目录

一.value类型

1.map()

 2.mapPartitions()

3.mapPartitionsWithIndex()

 4.flatMap()扁平化

 5.glom()

 6.groupBy()

 7.filter()

8.sample()

9. distinct()

10.coalesce()

11.repartition()

 12.sortBy()

13. pipe()

二、双value 

1.intersection()

2.union()

3.subtract()

 4.zip()

 三、KEY-VALUE类型

1.partitionBy()

2.自定义分区 

3.reduceByKey()

4.groupByKey()

5.aggregateByKey()

6.foldByKey()

7.combineByKey()

8.sortByKey()

 9.mapValues()

10.join() 

 11.cogroup()

 12.求top3

---

一.value类型

1.map()

映射：新RDD每一个元素都是由原来RDD中每一个元素依次应用函数f得到的

val value: RDD[Int] = sc.makeRDD(1 to 4, 2)
val rdd1: RDD[Int] = value.map(_*2)

 2.mapPartitions()

        以分区为单位执行Map

val rdd2: RDD[Int] = value.mapPartitions(list=> {println("计算一个数字") //运行两遍 因为两个分区list.map(i => {println("数字") //运行四遍i*2})})
rdd2.collect().foreach(println)

注：在空间资源较大时使用，一个分区数据处理完之后，原RDD分区中的数据才释放，可能导致OOM

3.mapPartitionsWithIndex()

        带分区号映射，创建一个RDD，使每个元素跟所在分区号形成一个元组，组成一个新的RDD

val rdd4: RDD[(Int, Int)] = value.mapPartitionsWithIndex((num,list)=> list.map(i=>(num,i)))
val rdd3: RDD[(Int, Int)] = value.mapPartitionsWithIndex((num,list)=>list.map((num,_)))
rdd3.collect().foreach(println)

 4.flatMap()扁平化

        将RDD中的每一个元素通过应用f函数依次转换为新的元素，并封装到RDD中。但在flatMap操作中，f函数的返回值是一个集合，并且会将每一个该集合中的元素拆分出来放到新的RDD中。

// 判断分区// flatMap不改变分区的情况  保持原分区val rdd3: RDD[(Int, Int)] = value1.mapPartitionsWithIndex((num,list)=>list.map((num,_)))rdd3.collect().foreach(println)// 对应 (长字符串 ,次数)  => (单词,次数),(单词,次数)val tupleRDD: RDD[(String, Int)] = sc.makeRDD(List(("hello world", 100), ("hello scala", 200)))val value3: RDD[(String, Int)] = tupleRDD.flatMap(tuple => {val strings: Array[String] = tuple._1.split(" ")strings.map(word => (word, tuple._2))})value3.collect().foreach(println)tupleRDD.flatMap(tuple => {tuple._1.split(" ").map(word => (word,tuple._2))})// 偏函数写法val value4: RDD[(String, Int)] = tupleRDD.flatMap(tuple => tuple match {case (line, count) => line.split(" ").map(word => (word, count))})value4.collect().foreach(println)val value5: RDD[(String, Int)] = tupleRDD.flatMap{case (line, count) => line.split(" ").map(word => (word, count))}

 5.glom()

        将RDD中每一个分区变成一个数组，并放置在新的RDD中，数组中元素的类型与原分区中元素类型一致

val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)
val value: RDD[Array[Int]] = listRDD.glom()
val arraydemo: RDD[Int] = value.map(Array => (Array.max))
arraydemo.collect().foreach(println)// 从一个sql脚本文件中  提取对应的sql
val lineRDD: RDD[String] = sc.textFile("input/1.sql",1)// 直接使用spark的算子读取sql脚本文件  里面的内容是打散为一行一行
val value2: RDD[Array[String]] = lineRDD.glom()//mkString 数组转化为字符串
value2.map(array => array.mkString).collect().foreach(println)

 6.groupBy()

        将相同K对应的值放入一个迭代器

    val sc: SparkContext = new SparkContext(conf)val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)val value: RDD[(Int, Iterable[Int])] = listRDD.groupBy(i => i%2)value.collect().foreach(println)//(0,CompactBuffer(2, 4))//(1,CompactBuffer(1, 3))//groupby实现wordcountval text: RDD[String] = sc.textFile("input/1.txt")val fmRDD: RDD[String] = text.flatMap(_.split(" "))val value: RDD[(String, Iterable[String])] = fmRDD.groupBy(s => s)value.collect().foreach(println)val value1: RDD[(String, Int)] = value.mapValues(list => list.size)value1.collect().foreach(println)val value2: RDD[(String, Int)] = value.map({case (word, list) => (word, list.size)})value2.collect().foreach(println)

 7.filter()

        接收一个返回值为布尔类型的函数作为参数。当某个RDD调用filter方法时，会对该RDD中每一个元素应用f函数，如果返回值类型为true，则该元素会被添加到新的RDD中。

    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)val value: RDD[Int] = listRDD.filter(i => i%2 ==0)value.collect().foreach(println)// 过滤// 保持分区不变

8.sample()

        采样，从大量的数据中采样

val listRDD: RDD[Int] = sc.makeRDD(1 to 10)// 抽取数据不放回（伯努利算法）// 伯努利算法：又叫0、1分布。例如扔硬币，要么正面，要么反面。// 具体实现：根据种子和随机算法算出一个数和第二个参数设置几率比较，小于第二个参数要，大于不要// 第一个参数：抽取的数据是否放回，false：不放回// 第二个参数：抽取的几率，范围在[0,1]之间,0：全不取；1：全取；// 第三个参数：随机数种子// 随机算法相同，种子相同，那么随机数就相同// 不输入参数，种子取的当前时间的纳秒值，所以随机结果就不相同了val value1: RDD[Int] = listRDD.sample(false,0.5,10)value1.collect().foreach(println)// 抽取数据放回（泊松算法）// 第一个参数：抽取的数据是否放回，true：放回；false：不放回// 第二个参数：重复数据的几率，范围大于等于0.表示每一个元素被期望抽取到的次数// 第三个参数：随机数种子val value2: RDD[Int] = listRDD.sample(true,2.3,100)value2.collect().foreach(println)

9. distinct()

        去重，将去重后的数据放入新的RDD中

 

val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4,5,4,2,3,1,6),2)
val value: RDD[Int] = listRDD.distinct()
//distinct会存在shuffle过程

10.coalesce()

        合并分区，缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。

    // 缩减分区// 多对一的关系 不走shuffleval coalrdd: RDD[Int] = listRDD.coalesce(2)val rdd3: RDD[(Int, Int)] =         coalrdd.mapPartitionsWithIndex((num,list)=>list.map((num,_)))rdd3.collect().foreach(println)println("===============")// 扩大分区// 必须走shuffle 不然没有意义val listRDD1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)val coalrdd2: RDD[Int] = listRDD1.coalesce(5,true)val rdd4: RDD[(Int, Int)] =         coalrdd2.mapPartitionsWithIndex((num,list)=>list.map((num,_)))rdd4.collect().foreach(println)

11.repartition()

        重新分区，该操作内部其实执行的是coalesce操作，参数shuffle的默认值为true。无论是将分区数多的RDD转换为分区数少的RDD，还是将分区数少的RDD转换为分区数多的RDD，repartition操作都可以完成，因为无论如何都会经shuffle过程。

    val sc: SparkContext = new SparkContext(conf)val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)val value: RDD[Int] = listRDD.repartition(4)val rdd3: RDD[(Int, Int)] = value.mapPartitionsWithIndex((num,list)=>list.map((num,_)))

 12.sortBy()

        排序，该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为正序排列。排序后新产生的RDD的分区数与原RDD的分区数一致。

    val listRDD: RDD[Int] = sc.makeRDD(List(1,6,5,4,2,3,4,9),2)val value: RDD[Int] = listRDD.sortBy(i => i)value.collect().foreach(println)println("==================")// spark的排序能够实现全局有序// 保证0号分区的数据都大于等于1号分区的数据// sortBy需要走shuffleval value2: RDD[Int] = listRDD.sortBy(i => i,false)val rdd3: RDD[(Int, Int)] = value2.mapPartitionsWithIndex((num,list)=>list.map((num,_)))rdd3.collect().foreach(println)
//    (0,9)
//    (0,6)
//    (0,5)
//    (1,4)
//    (1,4)
//    (1,3)
//    (1,2)
//    (1,1)

13. pipe()

        调用脚本，管道，针对每个分区，都调用一次shell脚本，返回输出的RDD

二、双value 

1.intersection()

        求交集

2.union()

         求并集

3.subtract()

        求差集

 4.zip()

        拉链，该操作可以将两个RDD中的元素，以键值对的形式进行合并。其中，键值对中的Key为第1个RDD中的元素，Value为第2个RDD中的元素。

        将两个RDD组合成Key/Value形式的RDD，这里默认两个RDD的partition数量以及元素数量都相同，否则会抛出异常。

    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)val listRDD1: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)// 求交集会打散重新分区  即需要走shuffle// 默认采用交集中较多的分区val demo01: RDD[Int] = listRDD.intersection(listRDD1)demo01.collect().foreach(println)// 求并集// 并集不走shuffle// 只是把两个RDD的分区数据拿到一起  分区的个数等于两个RDD分区个数之和val demo02: RDD[Int] = listRDD.union(listRDD1)val rdd3: RDD[(Int, Int)] = demo02.mapPartitionsWithIndex((num,list)=>list.map((num,_)))rdd3.collect().foreach(println)// 求差集// 需要重写分区  走shuffle  可以自己写分区数println("=======================")val demo03: RDD[Int] = listRDD.subtract(listRDD1)val rdd4: RDD[(Int, Int)] = demo03.mapPartitionsWithIndex((num,list)=>list.map((num,_)))rdd4.collect().foreach(println)// 将相同分区对应位置的元素拉链到一起  成为一个2元组// zip只能操作两个rdd具有相同的分区个数和元素个数val demo04: RDD[(Int, Int)] = listRDD.zip(listRDD1)demo04.mapPartitionsWithIndex((num,list) => list.map((num,_))).collect().foreach(println)

 三、KEY-VALUE类型

1.partitionBy()

        根据key值重新分区，如果原有的RDD和新的RDD是一致的话就不进行分区，否则会产生Shuffle过程。

    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)val value: RDD[(Int, Int)] = listRDD.map((_,1))//按照key对分区数取模分区val hp: RDD[(Int, Int)] = value.partitionBy(new HashPartitioner(2))hp.mapPartitionsWithIndex((num,list)=>list.map((num,_))).collect().foreach(println)

2.自定义分区 

        要实现自定义分区器，需要继承org.apache.spark.Partitioner类，并实现下面三个方法。

（1）numPartitions: Int:返回创建出来的分区数。

（2）getPartition(key: Any): Int:返回给定键的分区编号（0到numPartitions-1）。

（3）equals():Java 判断相等性的标准方法。这个方法的实现非常重要，Spark需要用这个方法来检查你的分区器对象是否和其他分区器实例相同，这样Spark才可以判断两个RDD的分区方式是否相同

    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)val value: RDD[(Int, Int)] = listRDD.map((_,1))val hp: RDD[(Int, Int)] = value.partitionBy(new MyPartition)hp.mapPartitionsWithIndex((num,list)=>list.map((num,_))).collect().foreach(println)sc.stop()}//自定义分区class MyPartition extends Partitioner{override def numPartitions: Int = 2// 获取分区号 => 根据元素的key值  判断分给哪个分区// spark 的分区器只能对key进行分区override def getPartition(key: Any): Int = {key match {case i:Int => i%2case _ => 0}

3.reduceByKey()

        元素按照相同的Key对Value进行聚合。其存在多种重载形式，还可以设置新RDD的分区数。

    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)val tuplerdd: RDD[(Int, Int)] = listRDD.map( (_ ,1))val result1: RDD[(Int, Int)] = tuplerdd.reduceByKey((res,elem) => res - elem)result1.collect().foreach(println)// 验证结果// 需要进行两次归约  一次分区内  一次分区间// 分区间的第一个元素取决于分区的编号   编号越小越靠前val value1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 1), ("a", 1), ("b", 1), ("b", 1), ("b", 1), ("b", 1), ("a", 1),("c",1)), 2)val result2: RDD[(String, Int)] = value1.reduceByKey((res,elem) => res - elem)result2.collect().foreach(println)

4.groupByKey()

        按照key重新分组，对每个key进行操作，但只生成一个seq，并不进行聚合

    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4,1,2,3,4),2)val tupleRDD: RDD[(Int, Int)] = listRDD.map((_, 1))val result: RDD[(Int, Iterable[(Int, Int)])] = tupleRDD.groupBy(tuple => tuple._1)result.collect().foreach(println)//(4,CompactBuffer((4,1), (4,1)))//(2,CompactBuffer((2,1), (2,1)))//(1,CompactBuffer((1,1), (1,1)))//(3,CompactBuffer((3,1), (3,1)))val result2: RDD[(Int, Iterable[Int])] = tupleRDD.groupByKey()result2.collect().foreach(println)//(4,CompactBuffer(1, 1))//(2,CompactBuffer(1, 1))//(1,CompactBuffer(1, 1))//(3,CompactBuffer(1, 1))

5.aggregateByKey()

        按照Key处理分区内和分区间逻辑

    val value1: RDD[(String, Int)] = sc.makeRDD(List(("a", 10), ("b", 7), ("a", 11), ("b", 21)), 4)val result1: RDD[(String, Int)] = value1.aggregateByKey(10)(_ + _,_ + _)result1.collect().foreach(println)

6.foldByKey()

        即分区内和分区间相同的aggregateByKey()

    val value1: RDD[(String, Int)] = sc.makeRDD(List(("a", 10), ("b", 7), ("a", 11), ("b", 21)), 4)value1.foldByKey(10)(_ + _).collect().foreach(println)value1.foldByKey(10)((res,elem) => math.max(res,elem)).collect().foreach(println)

7.combineByKey()

        转换结构后分区内和分区间操作，针对相同Key，将Value合并成一个集合。

    val value1: RDD[(String, Int)] = sc.makeRDD(List(("a", 10), ("b", 7), ("a", 11), ("b", 21)), 4)// 对上面的元素进行归约  (单词,("product",21))val value: RDD[(String, (String, Int))] = value1.combineByKey(i => ("product", i),// 分区内计算  是转换结构后的初始值和分区相同key的元素值进行归约(res: (String, Int), elem: Int) => (res._1, res._2 * elem),// 分区间计算  将每个分区相同key的res值进行合并(res: (String, Int), elem: (String, Int)) => (res._1, res._2 * elem._2))value.collect().foreach(println)println("==========================")val list: List[(String, Int)] = List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))val listRDD1: RDD[(String, Int)] = sc.makeRDD(list)val result2: RDD[(String, (Int, Int))] = listRDD1.combineByKey(// 将(a,88) => (a,(88,1)) 因为算子已经内部按照key聚合了  所以写的时候只写valuei => (i, 1),(res: (Int, Int), elem: Int) => (res._1 + elem, res._2 + 1),// 分区内累加   将相同分区相同key的值合并  (88,1)和91  => (179,2)// 分区间累加   将不同分区相同key的二元组合并在一起  (179,2) 和 (95,1) => (274,3)(res: (Int, Int), elem: (Int, Int)) => (res._1 + elem._1, res._2 + elem._2))result2.mapValues({case (res,elem) => res.toDouble/elem}).collect().foreach(println)

区别：

8.sortByKey()

        按照K进行排序，在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD

    val value: RDD[(String, Int)] = sc.makeRDD(List(("a", 110), ("b", 27), ("a", 11), ("b", 21)), 2)// 默认使用range分区器（尽量保证每个分区中数据量的均匀，而且分区与分区之间是有序的，一个分区中的元素肯定都是比另一个分区内的元素小或者大；但是分区内的元素是不能保证顺序的。）// 固定使用二元组中的key进行排序   不会使用valuevalue.sortByKey(false).collect().foreach(println)// 使用value排序// 可以使用sortBy 底层仍为sortbykeyvalue.sortBy(_._2).collect().foreach(println)

 9.mapValues()

       只对value进行修改

    val value: RDD[(String, Int)] = sc.makeRDD(List(("a", 110), ("b", 27), ("a", 11), ("b", 21)), 2)value.mapValues(i => i*2).collect().foreach(println)

10.join() 

        在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

    val value: RDD[(String, Int)] = sc.makeRDD(List(("a", 110), ("b", 27), ("a", 11), ("b", 21)), 2)val value1: RDD[(String, Int)] = sc.makeRDD(List(("a", 110), ("b", 27), ("a", 11), ("b", 21)), 2)// 将相同key合并// join走shuffle   使用hash分区器// 尽量保证join之前key是不重复的  如果有重复  会造成最终结果也是重复的value.join(value1).collect().foreach(println)//解决重复value.groupByKey().join(value1).collect().foreach(println)

 11.cogroup()

        在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD

    val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1,"a"),(2,"b"),(3,"c")))val rdd1: RDD[(Int, Int)] = sc.makeRDD(Array((1,4),(2,5),(4,6)))rdd.cogroup(rdd1).collect().foreach(println)

 12.求top3

   val listRDD: RDD[String] = sc.textFile("input/agent.log")val result1: RDD[(String, Int)] = listRDD.map({line => {val strings: Array[String] = line.split(" ")(strings(1) + "-" + strings(2), 1)}}).reduceByKey(_ + _)result1.collect().foreach(println)val result2: RDD[(String, Iterable[(String, Int)])] = result1.map({ case (res, sum) => {val strings01: Array[String] = res.split("-")(strings01(0), (strings01(1), sum))}}).groupByKey()val result3: RDD[(String, List[(String, Int)])] = result2.mapValues({datas => {datas.toList.sortWith((left, right) => {left._2 > right._2}).take(3)}})
// 简写result3.mapValues(_.toList.sortWith(_._2 > _._2).take(3))result3.collect().foreach(println)