map

/*
 map(func: RDD元素类型=> B ): 映射
map一般用于数据转换[类型的转换、值的转换]
 map里面的函数是针对RDD每个元素进行操作,后续会将RDD每个元素传递函数的参数,RDD中元素有多少个,函数就会调用多少次
val rdd2 = rdd1.map(…) 此时rdd2元素的个数 = rdd1元素个数
 map的应用场景: 一对一
*/

  1.   @Test
  2.   def map(): Unit ={
  4.     val rdd = sc.parallelize(List(10,2,6,3,7,9))
  6.     val rdd2 = rdd.map( x => {
  7.       println(s"${Thread.currentThread().getName}")
  8.       x * x
  9.     })

mapPartitions

/*
 mapPartitions(func: 迭代器[RDD元素类型]=>迭代器[B])
mapPartitions里面的函数是针对一整个分区进行处理,后续会将一个分区所有数据用迭代器封装之后传递给函数的参数
 rdd有多少分区,mapPartitions里面的函数就会调用多少次
处理大量数据的时候，mapPartitions因为是对每个分区中的所有数据进行处理，不用频繁开启资源、关闭资源，效率会快很多
（map的缺陷：针对每个元素进行操作，假如map方法体中用jdbc连接数据库，那么每个元素的获取值的时候需要启动和关闭jdbc）

map与mapPartitions的区别:
 1、函数针对的对象不一样:
map中的函数针对的是RDD每个元素
 mapPartitions里面的函数针对的是RDD每个分区
2、返回类型不一样:
 map要求返回一个元素,对于类型没有要求,map生成的新RDD元素个数等于原来RDD元素个数.
mapPartitions要求返回一个迭代器,mapPartitions生成的新RDD元素个数可能不等于原来RDD元素个数.
 3、元素内存释放的时机不一样
map里面的函数是针对单个元素处理,元素处理完成之后就可以回收元素内存
 mapPartitions里面的函数针对的是RDD一个分区的所有数据的迭代器,需要等到一个分区所有数据全部处理完成之后才能回收内存。mapPartitions又能出现内存溢出，出现内存溢出之后可以用map代替【完成比完美要重要】
*/

  1.   @Test
  2.   def mapPartitions(): Unit ={
  3.     //userid
  4.     val rdd = sc.parallelize(List(10,2,6,3,7,9))
  5.     //获取用户的信息[姓名、年龄、地址、id],用户的信息在mysql表中
  9.     rdd.map(id=>{
  11.       var name:String = null
  12.       var address:String = null
  13.       var age:Int = -1
  15.       var connection:Connection = null
  16.       var statement:PreparedStatement = null
  17.       try{
  19.         connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test")
  20.         statement = connection.prepareStatement("select * from person where id=?")
  21.         statement.setInt(1,id)
  22.         val set = statement.executeQuery()
  24.         while (set.next()){
  25.           name = set.getString("name")
  26.           address = set.getString("address")
  27.           age = set.getInt("age")
  28.         }
  29.         (id,name,age,address)
  30.       }catch {
  31.         case e:Exception => (id,name,age,address)
  32.       }finally {
  33.         if(statement!=null)
  34.           statement.close()
  35.         if(connection!=null)
  36.           connection.close()
  37.       }
  38.       id
  39.     }).collect()
  43.     val rdd3 = rdd.mapPartitions(it=>{
  44.       println("mapPartitions-"+Thread.currentThread().getName)
  45.       it.filter(_%2==0)
  46.       /*var connection:Connection = null
  47.       var statement:PreparedStatement = null
  48.       var list = List[(String,Int,String,Int)]()
  49.       try{
  51.         connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test")
  52.         statement = connection.prepareStatement("select * from person where id=?")
  53.         var name:String = null
  54.         var address:String = null
  55.         var age:Int = -1
  56.         it.foreach(id=>{
  57.           statement.setInt(1,id)
  58.           val set = statement.executeQuery()
  60.           while (set.next()){
  61.             name = set.getString("name")
  62.             address = set.getString("address")
  63.             age = set.getInt("age")
  64.           }
  65.           list = (name,age,address,id) :: list
  66.         })
  67.       }catch {
  68.         case e:Exception =>  list
  69.       }finally {
  70.         if(statement!=null)
  71.           statement.close()
  72.         if(connection!=null)
  73.           connection.close()
  74.       }
  76.       list.iterator*/
  77.     })
  79.     println(rdd3.collect().toList)
  80.   }

mapPartitionsWithIndex

/*
 mapPartitionsWithIndex(func: (Int,迭代器[RDD元素类型]) =>迭代器[B])
mapPartitionsWithIndex与mapPartitions的区别:
 mapPartitionsWithIndex里面的函数比mapPartitions里面的函数多了一个索引号
mapPartitionsWith里面的函数是针对一整个分区进行处理,后续会将一个分区所有数据用迭代器封装之后传递给函数的第二个参数
 rdd有多少分区,mapPartitionsWithIndex里面的函数就会调用多少次
*/
image.png

  2.   @Test
  3.   def mapPartitionsWithIndex(): Unit ={
  5.     val rdd = sc.parallelize(List(10,3,6,4,2,9))
  7.     val result = rdd.mapPartitionsWithIndex( (index,it) => {
  8.       println(s"index=${index}  data=${it.toList}")
  9.       it
  10.     } ).collect()
  12.     println(result.toList)
  13.   }

flatmap

/*
 flatMap = map + flatten
flatMap(func: RDD元素类型 => 集合[B] )
 flatMap里面的函数也是针对RDD每个元素操作,元素有多少个函数就调用多少次
*/

  2.   @Test
  3.   def flatMap(): Unit ={
  5.     val rdd = sc.parallelize(List( List(1,2),List(2,3) ))
  7.     val rdd2 = rdd.flatMap( x=> x )
  9.     println(rdd2.collect().toList)
  10.   }

glom

/*
 glom: 将一个分区所有数据用数组封装
glom生成的RDD元素个数 = 分区数
/

  2.   @Test
  3.   def glom(): Unit ={
  4.     val rdd = sc.parallelize(List(1,2,3,4,5,6,6,8,8,9,0))
  6.     rdd.mapPartitionsWithIndex((index,it)=>{
  7.       println(s"index:${index} data:${it.toList}")
  8.       it
  9.     }).collect()
  11.     val rdd2 = rdd.glom()
  13.     val result = rdd2.collect()
  15.     result.foreach(x=>println(x.toList))
  16.   }

groupBy

/*
 groupBy(func: RDD元素类型=>K): 根据指定字段分组
groupBy后续是根据函数的返回值进行分组
 groupBy针对的也是RDD每个元素操作,元素有多少个,函数调用多少次

 MR 整个过程: file -> inputformat -> map -> 缓冲区 -> 缓冲区分区排序 -> 溢写 -> combiner -> 磁盘 -> reduce拉取数据 -> 归并排序 -> reduce -> 磁盘
shuffle过程: 缓冲区 -> 缓冲区分区排序 -> 溢写 -> combiner -> 磁盘 ->归并-> reduc e拉取数据 -> 归并排序
 Spark shuffle过程: 缓冲区 -> 缓冲区分区[排序] -> 溢写 -> combiner -> 磁盘 -> rdd分区拉取数据

/

  1. @Test
  2.   def groupBy(): Unit ={
  4.     val rdd = sc.parallelize(List("hello","spark","spark","java"))
  6.     val rdd2 = rdd.groupBy(x=>x)
  8.     println(rdd2.collect().toList)
  9.   }

filter

/*
 filter(func: RDD元素类型=>Boolean ): 过滤
filter保留的是函数返回值为true的数据
 filter中的函数是针对RDD每个元素操作,元素有多少个,函数就调用多少次
*/

  2.   @Test
  3.   def filter(): Unit ={
  4.     val rdd = sc.parallelize(List(10,2,5,7,9,3,20,4))
  6.     val rdd2 = rdd.filter(x=> x%2==0)
  8.     println(rdd2.collect().toList)
  9.   }

sample

/*
 sample: 采样 【一般用于数据倾斜场景】
withReplacement: 同一个数据是否可以多次采样 [true代表同一个数据可能被采样多次, false代表同一个数据最多被采样一次] 【工作中一般设置为false，设置为ture判断不了是否产生数据倾斜】
 fraction:
withReplacement=true, 此时fraction代表元素期望被采样的次数[>=0]
 withReplacement=false,此时fraction每个元素被采样的概率[0,1] 【工作中一般设置为0.1/0.2】

/
image.png

  2.   @Test
  3.   def sample(): Unit ={
  4.     val rdd = sc.parallelize(List(10,2,5,7,9,3,20,4,11,55,22,33,88,66))
  6.     val rdd2 = rdd.sample(true,5)
  7.     //println(rdd2.collect().toList)
  9.     val rdd3 = rdd.sample(false,0.6)
  10.     println(rdd3.collect().toList)
  13.   }

distinct去重

/*
 distinct: 去重
distinct有shuffle操作

*/

  2.   @Test
  3.   def distinct(): Unit ={
  4.     val rdd = sc.parallelize(List(10,2,5,7,9,3,2,6,3,2))
  6.     val rdd2 = rdd.distinct()
  8.     println(rdd2.collect().toList)
  10.     val rdd3 = mydistinct(rdd)
  12.     println(rdd3.collect().toList)
  13.     Thread.sleep(100000)
  15.   }
  16.  def mydistinct(rdd:RDD[Int]):RDD[Int] = {
  18.     val rdd2 = rdd.groupBy(x=>x)
  20.     rdd2.map(_._1)
  21.   }

coalesce合并分区

/*
 coalesce: 合并分区
coalesce默认情况下只能减少分区
 coalesce如果想要增大分区,需要将shuffle参数设置为true
shuffle设置为true代表会产生shuffle操作
 工作中一般搭配filter使用。
*/
减少分区不涉及shuffle操作，增大分区设计shuffle操作
原因 ：shuffle定义：一个父RDD将数据传输给多个子RDD，而减少分区算法中只能是一个子RDD接收多个父RDD：
image.png

  2.   @Test
  3.   def coalesce(): Unit ={
  4.     val rdd = sc.parallelize(List(10,2,5,7,9,3,2,6,3,2),5)
  6.     println(new Random(hashing.byteswap32(0)).nextInt(5))
  8.     rdd.mapPartitionsWithIndex((index,it)=>{
  9.       println(s"rdd index:${index} datas:${it.toList}")
  10.       it
  11.     }).collect()
  12.     println(rdd.partitions.length)
  14.     val rdd2 = rdd.coalesce(3)
  15.     rdd2.mapPartitionsWithIndex((index,it)=>{
  16.       println(s"rdd2 index:${index} datas:${it.toList}")
  17.       it
  18.     }).collect()
  20.     val rdd3 = rdd.coalesce(8,true)
  21.     rdd3.mapPartitionsWithIndex((index,it)=>{
  22.       println(s"rdd3 index:${index} datas:${it.toList}")
  23.       it
  24.     }).collect()
  25.     println(rdd2.partitions.length)
  26.   }

repartition重分区

/*
 repartition: 重分区
repartition既可以增大分区也可以减少分区,都会产生shuffle
 coleasce与repartition的区别:
coleasce默认只能减少分区,coaleasce的shuffle如果没有设置为true，是没有shuffle操作的
 repartition既可以增大分区也可以减少分区,都会产生shuffle,底层就是使用的coaleasce
如果减少分区推荐使用coleasce,coleasce一般搭配filter使用
 如果增大分区推荐使用repartition,因为简单
*/

  2.   @Test
  3.   def repartition(): Unit ={
  4.     val rdd = sc.parallelize(List(10,2,5,7,9,3,2,6,3,2),5)
  6.     val rdd2 = rdd.repartition(3)
  8.     println(rdd2.partitions.length)
  9.   }

sortBy

/*
 sortBy(func: RDD元素类型=>K，ascding=ture ): 根据指定字段排序
sortBy后续是根据函数的返回值进行排序
 sortBy里面的函数是针对集合每个元素进行操作
sortBy有shuffle操作
/

  2.   @Test
  3.   def sortBy(): Unit ={
  4.     val rdd = sc.parallelize(List(10,2,5,7,9,3,2,6,3,2),5)
  6.     val rdd2 = rdd.sortBy(x=>x,false)
  8.     rdd2.mapPartitionsWithIndex((index,it)=>{
  9.       println(s"index:${index} data:${it.toList}")
  10.       it
  11.     }).collect()
  12.     println(rdd2.collect().toList)
  13.   }

pipe

/*
 pipe: 调用指定的脚本
pipe生成的新RDD的元素是由脚本中通过echo返回的
 pipe在调用脚本的时候是每个分区调用一次
*/

  2.   @Test
  3.   def pipe(): Unit ={
  4.     val rdd = sc.parallelize(List(10,2,5,7,9,3,2,6,3,2),5)
  6.   }