RDD(2)

RDD转换算子

RDD根据数据处理方式的不同将算子整体上分为Value类型、双Value类型、Key-Value类型

value类型

map

函数签名
def map[U:ClassTag](f:T=>U):RDD[U]
函数说明
将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换
e.g.1

 val source = sparkContext.parallelize(Seq(1, 2, 3, 4, 5, 6))
    val map = source.map(item => item*10)
    val result = map.collect()
    result.foreach(println)

e.g.2

   val data1: RDD[Int] = sparkContext.parallelize(List(1, 2, 3, 4), 2)
//    val data2: RDD[Int] = sparkContext.parallelize(List(1, 2, 3, 4), 1)
    val rdd1: RDD[Int] = data1.map(
      num => {
        println(">>>" + num)
        num
      }
    )
    val rdd2: RDD[Int] = rdd1.map(
      num => {
        println("<<<" + num)
        num
      }
    )
    rdd2.collect()

note:
RDD计算同一分区内数据有序，不同分区数据无序

(func)从服务器日志数据apache.log中获取用户请求URL资源路径(例):

apache.log 提取码：unsk

83.149.9.216 - - 17/05/2015:10:05:12 +0000 GET /presentations/logstash-monitorama-2013/plugin/zoom-js/zoom.js
83.149.9.216 - - 17/05/2015:10:05:07 +0000 GET /presentations/logstash-monitorama-2013/plugin/notes/notes.js
83.149.9.216 - - 17/05/2015:10:05:34 +0000 GET /presentations/logstash-monitorama-2013/images/sad-medic.png

code:

val data = sparkContext.textFile("input/apache.log")
    val clean = data.map{
      item => {
        item.split(" ")(6)
      }
    }
    clean.foreach(println(_))

mapPartitions

函数签名

def mapPartitions[U:ClassTag](
  f:Iterator[T] =>Iterator[U],
  preservesPartitioning:Boolean = false):RDD[U]

函数说明
将待处理的数据以分区为单位发送到计算节点进行任意的处理(过滤数据亦可)
note: 函数会将整个分区的数据加载到内存中进行引用。内存较小、数据量较大的情况下，容易出现内存溢出。

val dataRDD1： RDD[Int]= dataRDD.mapPartitions(
  datas =>{            //遍历每个分区进行操作
    datas.filter(_==2) //过滤每个分区中值为2的数据
  }
)

(func)获取每个数据分区的最大值

code:

object getMaxFromArea {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Max")
    val sparkContext: SparkContext = new SparkContext(sparkConf)
    val source: RDD[Int] = sparkContext.parallelize(List(1, 2, 3, 4, 5, 6), 2)
    val mapPartition: RDD[Int] = source.mapPartitions(p => List(p.max).iterator)

    //多个分区获取最大值，使用迭代器
    val result: Array[Int] = mapPartition.collect()
    result.foreach(println)
    sparkContext.stop()

  }
}

comparison:

map和mapPartitions的区别

数据处理角度
Map 算子是分区内一个数据一个数据的执行，类似于串行操作。而 mapPartitions 算子
是以分区为单位进行批处理操作。

功能的角度
Map 算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。
MapPartitions 算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变，
所以可以增加或减少数据

性能的角度
Map 算子因为类似于串行操作，所以性能比较低，而是 mapPartitions 算子类似于批处
理，所以性能较高。但是 mapPartitions 算子会长时间占用内存，那么这样会导致内存可能
不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。使用 map 操作

mapPartitionsWithIndex

函数签名
def mapPartitionsWithIndex[U: ClassTag](f: (Int, Iterator[T]) => Iterator[U],preservesPartitioning: Boolean = false): RDD[U]
函数说明
将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据，在处理时同时可以获取当前分区索引。

val dataRDD1 = dataRDD.mapPartitionsWithIndex(
 (index, datas) => {
      datas.map(index, _)
 }
)

(func)获取第二个数据分区的数据

code:

object getSecondArea {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Sec")
    val sparkContext: SparkContext = new SparkContext(sparkConf)
    val source: RDD[Int] = sparkContext.parallelize(List(1, 2, 3, 4, 5, 6), 2)
    val mapPartitionsWithIndex: RDD[Int] = source.mapPartitionsWithIndex(
      (index, data) => {
        if (index == 1) {
          data
        } else {
          Nil.iterator
        }
      }
    ) 
    val result: Array[Int] = mapPartitionsWithIndex.collect()
    result.foreach(println(_))
    sparkContext.stop()
  }

}

(func)获取每个数据及其对应分区索引

code：

object getDataAndIndexOfArea {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Data")
    val sparkContext: SparkContext = new SparkContext(conf)

    val data: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4, 5, 6))
    val dataAndIndex: RDD[(Int, Int)] = data.mapPartitionsWithIndex(
      (index, iter) => {
        iter.map(data => (data, index))
      }
    )

    val result: Array[(Int, Int)] = dataAndIndex.collect()
    result.foreach(println)

  }

}

note：这里没有自定义分区数量，故默认最多分区数（与机器逻辑处理器数量相关),数据随机存储在这些分区中

flatMap

函数签名
def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]
函数说明
将处理的数据进行扁平化后再进行映射处理，所以算子也称作扁平映射

val dataRDD = sparkContext.makeRDD(
	List(List(1,2),List(3,4)),1)
val dataRDD1 = dataRDD.flatMap(list => list)
1
2
3
4

(func)将List(“Hello World”,”Hello Spark”)进行扁平化操作
1）字符扁平化

val data = sparkContext.makeRDD(List("Hello World","Hello Spark"),1)
    val rdd: RDD[Char] = data.flatMap(list => list)
    val result1: Array[Char] = rdd.collect()
    result1.foreach(item => print(item+" "))

2)字符串扁平化

val data1: RDD[String] = sparkContext.parallelize(List("Hello World", "Hello Spark"), 1)
    val rdd1: RDD[String] = data1.flatMap(list => {
      list.split(" ")
    })
    val result2: Array[String] = rdd1.collect()
    result2.foreach(item => println(item + " "))

(func)将List(List(1,2),3,List(4,5))进行扁平化操作

thinking：List(List(1,2),3,List(4,5)) => List(list,int,list) => RDD[Any]
当数据的格式不能够满足时我们可以使用match进行格式的匹配(类似java中的switch，case)

code：

val data2: RDD[Any] = sparkContext.parallelize(List(List(1, 2), 3, List(4, 5)))
    val rdd2: RDD[Any] = data2.flatMap {
   //  完整代码：
      //          dat =>{
      //            dat match {
      //              case list: List[_] => list
      //              case int => List(int)
      //            }
      //          }
      case list: List[_] => list
      case int => List(int)
    }
    val result3: Array[Any] = rdd2.collect()
    result3.foreach(item => print(item + " "))

result:

glom

函数签名
def glom(): RDD[Array[T]]
函数说明

将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变

e.g.

val dataRDD: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4), 1)
    val glomer: RDD[Array[Int]] = dataRDD.glom()
    val result: Array[Array[Int]] = glomer.collect()
    result.foreach(data => println(data.mkString(",")))

result: 1,2,3,4

(func)计算所有分区最大值之和
thinking：原分区 (1,2),(3,4)=>glom()=>List(1,2),List(3,4)=>map()=>取出每个里面的最大值，最后求和

code:

 val data: RDD[Int] = sc.parallelize(List(1, 2, 3, 4), 2)

    //  method1:mapPartitions
    val mapPartition: RDD[Int] = data.mapPartitions(p => List(p.max).iterator)
    val sum: Int = mapPartition.collect().sum
    println(sum)

    //  method2:glom
    val glom: RDD[Array[Int]] = data.glom()
    val max: RDD[Int] = glom.map(iter => iter.max)
    val sum1: Int = max.collect().sum
    print(sum1)

result:
6
6

groupby

函数签名
def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]
函数说明
将数据根据指定规则进行分组，分区默认不变，但数据会被打乱重新组合，这个操作被称作 shuffle

极限情况下，同一个组的数据可能被分在同一个分区中。注意，这并非意味着一个分区中只能有一个组。一如2-2-2组别分入两个分区。

 val dataRDD: RDD[Int] = sparkContext.parallelize(List(1, 2, 6, 3, 4, 5),5)
    val groupByer1: RDD[(Int, Iterable[Int])] = dataRDD.groupBy(_ % 3)
    val tuples: Array[(Int, Iterable[Int])] = groupByer1.collect()
    tuples.foreach(println(_))

result：
(0,CompactBuffer(6, 3))
(1,CompactBuffer(1, 4))
(2,CompactBuffer(2, 5))

(func)将List(“hello”,”spark”,”scala”,”hadoop”)根据单词首字母进行分组

code:

    val data: RDD[String] = sparkContext.makeRDD(List("hello", "spark", "scala", "hadoop"), 2)
    val gb: RDD[(Char, Iterable[String])] = data.groupBy(_.charAt(0))
    val tuples: Array[(Char, Iterable[String])] = gb.collect()
    tuples.foreach(println)

result:
(h,CompactBuffer(hello, hadoop))
(s,CompactBuffer(spark, scala))

(func)从服务器日志数据apache.log中获取每个时间段访问量
code:

 val data1: RDD[String] = sparkContext.textFile("input/apache.log")
    val split: RDD[(String, Int)] = data1.map(
      line => {
        val fields: Array[String] = line.split(" ")
        (fields(3).substring(11), 1)
      }
    )
    val gb1: RDD[(String, Iterable[(String, Int)])] = split.groupBy(_._1)
    val count: RDD[(String, Int)] = gb1.map(
      iter => {
        val count: Int = iter._2.size
        (iter._1, count)
      }
    )
    val tuples1: Array[(String, Int)] = count.collect()
    tuples1.foreach(println(_))

result：

(func)WordCount
code：

 val lines: RDD[String] = sparkContext.textFile("input/word.txt")
    val words: RDD[String] = lines.flatMap(_.split(" "))
    val groups: RDD[(String, Iterable[String])] = words.groupBy(word => word)
    val count1: RDD[(String, Int)] = groups.map {
      case (word, list) => {
        (word, list.size)
      }
    }
    val tuples2: Array[(String, Int)] = count1.collect()
    tuples2.foreach(println(_))

result：
(hello,2)
(world,1)
(spark,1)

filter

函数签名
def filter(f: T => **Boolean**): RDD[T]
函数说明
将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。
当数据进行筛选过滤后，分区不变，但是分区内的数据可能分布不均衡，生产环境下，可能出现数据倾斜
e.g.

  val dataRDD: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4), 1)
    val filter: RDD[Int] = dataRDD.filter(_ % 2 == 0)
    val result: Array[Int] = filter.collect()
    result.foreach(println(_))

result:
2
4

(func)从服务器日志数据apache.log中获取2015年5月17日的请求路径
code：

val data: RDD[String] = sparkContext.textFile("input/apache.log")
    val filter1: RDD[String] = data.filter(line => {
      val fields: Array[String] = line.split(" ")
      fields(3).startsWith("17/05/2015")
    })
    val strings: Array[String] = filter1.collect()
    strings.foreach(println(_))

result：
83.149.9.216 – – 17/05/2015:10:05:03 +0000 GET /presentations/logstash-monitorama-2013/images/kibana-search.png
83.149.9.216 – – 17/05/2015:10:05:43 +0000 GET /presentations/logstash-monitorama-2013/images/kibana-dashboard3.png
…………

sample

函数签名
def sample( withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong): RDD[T]
函数说明
根据指定的规则从数据集中抽取数据
e.g.

 val dataRDD = sparkContext.makeRDD(List(1,2,3,4,5,6,7,8,9,10))
    //抽取数据不放回（伯努利算法
    //伯努利算法：又叫0-1分布。譬如丢硬币，正反取其一
    //具体实现： 根据种子和随机算法算出一个数和第二个参数设置几率比较，小于第二个参数取得，大于第二个参数不取
    //第一个参数：是否放回，true:放回
    //第二个参数：数据源中每条数据被抽取的几率，范围在[0,1]之间，0：全不取；1：全取
    //基准值的概念
    //第三个参数：随机数种子
    //如果不填，将使用当前的系统时间作为种子数
    val sample1: String = dataRDD.sample(true, 2).collect().mkString(",")
    //随机取，放回
    println(sample1)

    //抽取数据放回（泊松算法
    //第一个参数：抽取的数据是否放回，true：放回
    //第二个参数：重复数据的几率，范围大于等于0，表示每一个元素被期望抽到的次数
    //第三个参数：随机数种子

    val sample2: String = dataRDD.sample(false, 0.4).collect().mkString(",")
    println(sample2)

result:
1,1,2,3,4,4,5,5,5,6,7,8,8,8,10,10
1,3,5,7,10

distinct

函数签名
def distinct()(implicit ord: Ordering[T] = null): RDD[T] def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
函数说明
将数据集内重复的数据去重
e.g.

val dataRDD: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 7, 5, 3, 3, 2, 1), 1)
    val dataRDD1: RDD[Int] = dataRDD.distinct()
    val result1: Array[Int] = dataRDD1.collect()
    result1.foreach({item =>print(item+" ")})

    println()

    val dataRDD2: RDD[Int] = dataRDD.distinct(2)
    val result2: Array[Int] = dataRDD2.collect()
    result2.foreach({item=>print(item+" ")})

result：
4 1 6 3 7 8 5 2
4 6 8 2 1 3 7 5

thinking：
distinct底层用到的是HashSet,dataRDD.distinct()替换思路如下
map(x=>(x,null)).reduceByKey((x,y)=>x,numPartitions).map(_,_1)

Spark源码

coalesce

函数签名
`
def coalesce(numPartitions: Int, shuffle: Boolean = false,
partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
(implicit ord: Ordering[T] = null)

`
函数说明
根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率
当Spark程序中，存在过多小任务时，可以通过coalesce方法，收缩合并分区，减少分区的个数，减小任务调度成本
note：coalesce算子不仅可以缩减分区，也可以扩充分区
但是coalesce不会将分区数据打乱组合，是将原先分区上数据整体迁移/增加分区
所以会产生数据倾斜。要让数据均衡，可以添加参数进行shuffle操作(增加分区必须进行shuffle操作，因为扩充过程中会打乱原有的组合)。
e.g.

 val dataRDD1: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4, 5, 6), 3)
    //    缩减分区
    val coalesce1: RDD[Int] = dataRDD1.coalesce(2)
    coalesce1.saveAsTextFile("output/output_coalesce_3_2")

    val coalesce1_2: RDD[Int] = dataRDD1.coalesce(2,shuffle = true)
    coalesce1_2.saveAsTextFile("output/output_coalesce_3_2_shuffle")

thinking:

repartition

函数签名
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
函数说明
该操作内部其实执行的是 coalesce 操作，参数 shuffle 的默认值为 true。无论是将分区数多的RDD转换为分区数少的RDD，还是将分区数少的 RDD 转换为分区数多的 RDD，repartition
操作都可以完成，因为无论如何都会经 shuffle 过程。

note:为了简化，扩充分区的时候常用repartition算子，直接重新分区。但其底层其实还是使用coalesce算子实现的。

/**
  * Return a new RDD that has exactly numPartitions partitions.
  *
  * Can increase or decrease the level of parallelism in this RDD. Internally, this uses
  * a shuffle to redistribute data.
  *
  * If you are decreasing the number of partitions in this RDD, consider using `coalesce`,
  * which can avoid performing a shuffle.
  */
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
  coalesce(numPartitions, shuffle = true)  // 底层使用的就是coalesce算子
}

softBy

函数签名
def sortBy[K](f: (T) => K,ascending: Boolean = true,numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
函数说明
该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为升序排列。排序后新产生的RDD的分区数与原RDD的分区数一致，中间存在shuffle的过程。
e.g.

 val dataRDD: RDD[(Int, String)] = context.parallelize(List((1, "11"), (2, "2"), (3, "13"), (4, "4"), (5, "15")))

    print("SortByNum:")

    val dataRDD1: RDD[(Int, String)] = dataRDD.sortBy(_._1)
    dataRDD1.collect().foreach({
      item=> print(item+" ")
    })
    
    println()

    print("SortByString:")

    val dataRDD2: RDD[(Int, String)] = dataRDD.sortBy(_._2)
    dataRDD2.collect().foreach({
      item=> print(item+" ")
    })

result:

双Value类型

双Value类型，两个数据源之间数据操作

intersection

函数签名
def intersection(other: RDD[T]): RDD[T]
函数说明
对源RDD和参数RDD求交集(取共有部分的数据集)后返回一个新的RDD

e.g.

val dataRDD1 = sc.makeRDD(List(1,2,3,4))
val dataRDD2 = sc.makeRDD(List(3,4,5,6))
val dataRDD = dataRDD1.intersection(dataRDD2)

note:如果两个RDD数据类型不一致，编译时会报错。所以RDD类型必须保持一致
下面的union、subtract算子同样适用这个准则

union

函数签名
def union(other: RDD[T]): RDD[T]
函数说明
对源RDD和参数RDD求并集(取全部数据集)后返回一个新的RDD
e.g.

val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6))
val dataRDD = dataRDD1.union(dataRDD2)

subtract

函数签名
def subtract(other: RDD[T]): RDD[T]
函数说明
以一个RDD元素为主，去除两个RDD中发生重复的元素(所有)，将其他元素保留下来，求差集
legend:

e.g.

val RDD1: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    val RDD2: RDD[Int] = sparkContext.makeRDD(List(3, 4, 5, 6))

    //    intersection(交
    print("intersection(交 ")
    val value: RDD[Int] = RDD1.intersection(RDD2)
    println(value.collect().mkString(","))

    //    union(并
    print("union(并 ")
    val value1: RDD[Int] = RDD1.union(RDD2)
    println(value1.collect().mkString(","))

    //    subtract(差
    println("subtract(差 ")
    val value2: RDD[Int] = RDD1.subtract(RDD2)
    println(value2.collect().mkString(","))
    val value3: RDD[Int] = RDD2.subtract(RDD1)
    println(value3.collect().mkString(","))

result:

zip

函数签名
def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]
函数说明
将两个RDD中的元素，以K-V(键值对)的形式合并。其中，键值对中的Key为第一个RDD中的元素，Value为第二个RDD中的相同位置的元素。
e.g.

 val dataRDD1: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4))
    val dataRDD2: RDD[Int] = sparkContext.makeRDD(List(3, 4, 5, 6))

    val dataRDD3: RDD[(Int, Int)] = dataRDD1.zip(dataRDD2)
    val dataRDD4: RDD[(Int, Int)] = dataRDD2.zip(dataRDD1)

    println(dataRDD3.collect().mkString(","))
    println(dataRDD4.collect().mkString(","))

result：

note：
①zip算子不要求两个RDD的数据类型保持一致
“(other: RDD[U]): RDD[(T, U)”
origin code:

  def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)] = withScope {
    zipPartitions(other, preservesPartitioning = false) { (thisIter, otherIter) =>
      new Iterator[(T, U)] {
        def hasNext: Boolean = (thisIter.hasNext, otherIter.hasNext) match {
          case (true, true) => true
          case (false, false) => false
          case _ => throw new SparkException("Can only zip RDDs with " +
            "same number of elements in each partition")
        }
        def next(): (T, U) = (thisIter.next(), otherIter.next())
      }
    }
  }

②zip算子要求两个数据源的分区必须保持一致，相关报错
“报错：java.lang.IllegalArgumentException: Can”t zip RDDs with unequal numbers of partitions: List(2, 4)”
③zip算子要求RDD相同分区下的数据量必须相同，相关报错
“报错：org.apache.spark.SparkException: Can only zip RDDs with same number of elements in each partition”

Key-Value类型

partitionBy

函数签名
`def partitionBy(partitioner: Partitioner): RDD[(K, V)]

// RDD中通过隐式转换将 partitionBy => PairRDDFunctions
implicit def rddToPairRDDFunctions[K, V](rdd: RDD[(K, V)])
(implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K] = null): PairRDDFunctions[K, V] = {
new PairRDDFunctions(rdd)
}
`
函数说明
将数据按照指定Partitioner重新进行分区。Spark中Partitioner算子默认的分区器是HashPartitioner,按照key的hash_code进行分区判别
spark大多数算子使用的都是默认分区器HashPartitioner，HashPartitioner会对数据的key进行 key.hascode%numpartitions 计算,得到的数值会放到对应的分区中，这样能较为平衡的分配数据到partition,笔者将相关赘述放在模块后面的分区器(Partitioner)专题中，如有需要可以查阅
e.g.

 val rdd: RDD[(Int, String)] = sparkContext.makeRDD(Array((1, "aaa"), (2, "bbb"), (3, "ccc")), 3)
    rdd.saveAsTextFile("output/output_partitionBy_3")

    val rdd2: RDD[(Int, String)] = rdd.partitionBy(new HashPartitioner(2))
    rdd2.saveAsTextFile("output/output_partitionBy_3_2")

result:

note:
①如果重分区的分区器和当前RDD的分区器一样怎么办？
origin code:

底层源码会先匹配分区器的类型，如果分区器类型和分区数均相同，则返回RDD本身；如分区器类型不同则会创建新的RDD （类型和分区数确认都不变就不变,分区器类型变就返回新的）
②如图所示，还有别的分区器，HashPartitioner只是大多数Spark算子默认的分区器
origin code:

// Partitioner 抽象类 ---> 实现类
abstract class Partitioner extends Serializable {
  def numPartitions: Int
  def getPartition(key: Any): Int
}

③如果有按照自己的方法进行数据分区的需求，需要自己创建分区器。

reduceByKey

函数签名
def reduceByKey(func: (V, V) => V): RDD[(K, V)] def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
函数说明
可以将数据按照相同的Key对Value进行聚合，scala语言中一般的聚合操作都是两两聚合
e.g.

  val dataRDD1: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("b", 2), ("c", 3), ("a", 1)))
//    val dataRDD2: RDD[(String, Int)] = dataRDD1.reduceByKey(_ + _)
    val dataRDD3: RDD[(String, Int)] = dataRDD1.reduceByKey(_ + _, 2)

    dataRDD3.collect().foreach(println(_))

result:
(b,2)
(a,2)
(c,3)

note:
可以看到reduceByKey接收一个func参数，而这个func参数接收两个V类型的参数并返回一个V类型的结果，这里的V其实就是初始RDD中的元素，这里需要传入的func就是元素两两聚合计算的逻辑。
并且在整个过程中，key组只有一个数据的不会参与计算，最终直接输出。
code:

val dataRDD1: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("b", 1)))
//    val dataRDD2: RDD[(String, Int)] = dataRDD1.reduceByKey(_ + _)
    val dataRDD3: RDD[(String, Int)] = dataRDD1.reduceByKey(_ + _, 2)

    dataRDD3.collect().foreach(println(_))

result:
(b,1)
(a,6)

如上所述，我们在进行reduceByKey算子操作的时候进行过程计算数据的输出，可见只有a的数据进行了累加，b的并没有参与计算。

groupByKey

函数签名

def groupByKey(): RDD[(K, Iterable[V])]
def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]

函数说明
将数据源的数据根据key对value进行分组
e.g.

val dataRDD1: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("b", 3)))
    val dataRDD2: RDD[(String, Iterable[Int])] = dataRDD1.groupByKey()
    val dataRDD3: RDD[(String, Iterable[Int])] = dataRDD1.groupByKey(2)
    val dataRDD4: RDD[(String, Iterable[Int])] = dataRDD1.groupByKey(new HashPartitioner(2))

    dataRDD2.collect().foreach(println(_))
    println()
    dataRDD3.collect().foreach(println(_))
    println()
    dataRDD4.collect().foreach(println(_))

result:
(a,CompactBuffer(1, 2))
(b,CompactBuffer(3))

(b,CompactBuffer(3))
(a,CompactBuffer(1, 2))

(b,CompactBuffer(3))
(a,CompactBuffer(1, 2))

comparison:
①groupBy 与 groupByKey 区别：
e.g.1
code:

 val dataRDD1: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("b", 3)))
    //    将数据源中相同的key的数据分在一个组，形成一个对偶元组
    //    (key,相同key的元素值集合)
    val dataRDD2: RDD[(String, Iterable[Int])] = dataRDD1.groupByKey()
    //    groupBy相比groupByKey更加灵活，能够对不同的value进行分组，而groupByKey只能对Key进行分组
    //    两者的结构有所不同：(key,相同key的元素集合)
    val dataRDD3: RDD[(String, Iterable[(String, Int)])] = dataRDD1.groupBy(_._1)

    println()
    dataRDD2.collect().foreach(println(_))
    println()
    dataRDD3.collect().foreach(println(_))

result:
(a,CompactBuffer(1, 2, 3))
(b,CompactBuffer(3))

(a,CompactBuffer((a,1), (a,2), (a,3)))
(b,CompactBuffer((b,3)))

虽然都是按照key值进行分组，但是得到的分组的数据结构有区别

e.g.2
code:

// 计算初始RDD不同首字母开头的元素数量
val rdd: RDD[String] = sc.makeRDD(List("Hello", "Java", "Python", "PHP", "Help"))
// ("H", ("Hello", "Help")), ("J", ("Java")), ("P", ("Python", "PHP"))
val groupRDD: RDD[(Char, Iterable[String])] = rdd.groupBy(_.charAt(0))
// ("H", 2), ("J", 1), ("P", 2)
val sizeRDD: RDD[Int] = groupRDD.map(_._2.size)
sizeRDD.collect().foreach(println)

result:
2
2
1

参考下图，第二步添加了File落盘动作，因为groupBy操作会计算每个分区所有单词的首字母并缓存下来，如果放在内存中若数据过多则会产生内存溢出；再就是第三步从文件读取回来，并不一定是三个分区，这里只是为了便于理解。
legend:

②groupByKey和reduceByKey的区别
legend:
groupByKey

reduceByKey
从下图中的第一张图看相对于groupByKey只是少了map的步骤将它整合在reduceByKey中，但是实际上reduceByKey的作用不止于此，第二张图才是实际的运行模式，它提供了Combine预聚合的功能，支持在分区中先进行聚合，称作分区内聚合，然后再落盘等待分区间聚合。这样下来它不只是减少了map的操作，同时提供了分区内聚合使得shuffle落盘时的数据量尽量小，IO效率也会提高不少。最后它引出了分区内聚合和分区间聚合，reduceByKey的分区内聚合和分区间聚合是一样的。

1. 从 shuffle 角度：
   同：reduceByKey和groupByKey都存在shuffle的操作，
   异: reduceByKey可以在shuffle前对分区内相同key值的数据进行预聚合（combine）功能，这样会减少落盘的数据量
   groupByKey 只是进行分组，不存在数据量减少的问题
2. 从 功能 角度：
   reduceByKey包含 分组 和 聚合的功能。
   groupByKey****只能分组，不能聚合
   所以 分组聚合 场景，使用reduceByKey更好。如果仅仅需要分组，不需要聚合，只能使用groupByKey

note：
reduceByKey 分区内 与 分区间 的 计算规则相同，都是按Key值进行分组聚合。但是在存在分区内 和 分区间 计算规则不同的情况(如 先求 分区内最大值，再求各分区间最大值的总和)，这时reduceByKey不再适用，可以使用aggregateByKey算子实现

aggregateByKey

函数签名
def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,combOp: (U, U) => U): RDD[(K, U)]
函数说明
将数据根据不同的规则进行分区内计算和分区间计算
e.g.
code:

val dataRDD1: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("c", 3)))
    val dataRDD2: RDD[(String, Int)] = dataRDD1.aggregateByKey(0)(_ + _, _ + _)

    dataRDD2.collect().foreach(println(_))

result:
(a,3)
(c,3)
(func)：
①取出每个分区内相同key地最大值，然后分区间相加
thinking:
(a,[1,2]) (a,[3,4])
=> (a,2) (a,4)
=> (a,6)
aggregateByKey存在函数柯里化(Currying)，有两个参数列表

1. 第一个参数列表需要传递一个参数，表示初始值

• 主要用于碰见第一个key时，和value进行分区内的计算

2. 第二参数列表需要传递两个参数

• 参数1：表示分区内的计算规则
• 参数2：表示分区间的计算规则
  code:

 val dataRDD3: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("b", 3), ("b", 4), ("b", 5), ("a", 3), ("a", 4)), 2)
    dataRDD3.aggregateByKey(0)(
      (x, y) => math.max(x, y),
      (x, y) => x + y
    ).collect().foreach(println)

result:
(b,9)
(a,5)
legend:

②求出相同Key的数据的平均值
aggregateByKey 最终返回的数据结果应该和初始值的数据类型保持一致，且整个过程中Key保持不变
code：

 val dataRDD3: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("b", 3), ("b", 4), ("b", 5), ("a", 3), ("a", 4)), 2)
 // (0,0) => (初始值，记录同key组中数的数量 --- size)
    val dataRDD4: RDD[(String, (Int, Int))] = dataRDD3.aggregateByKey((0, 0))(
      // 分区内计算：t代表上一个（Int，Int）结果，v代表当前的 value 【即下一个（Int，Int）中第一个Int的值】
      // 前值相加；后数目加一，用于统计单个分区内的数据量
      (t, v) => {
        (t._1 + v, t._2 + 1)
      },
      // 分区间计算：t1,t2代表两个（Int，Int）的结果 （值与值相加，数的数量与数的数量相加）
      // 前值相加；后数目相加，用于求所有分区内的数据量之和
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2)
      }
    )
    //前为原数据分类，后为 值之总和 与 数据量之总和 求商 => 平均值
    val dataRDD5: RDD[(String, Int)] = dataRDD4.map(item => (item._1, item._2._1 / item._2._2))
    dataRDD5.collect().foreach(println)

    // 与如下map算子等效
    val dataRDD6: RDD[(String, Int)] = dataRDD4.mapValues {
      case (sum, count) => {
        sum / count
      }
    }
    dataRDD6.collect().foreach(println)

result：
(b,4)
(a,2)

foldByKey

函数签名
def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
函数说明
当 分区内 与 分区间 计算规则相同时，aggregateByKey 就可以简化为foldByKey
e.g.
code:

 val dataRDD: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 3), ("a", 5), ("c", 3), ("c", 5), ("c", 6)))
    dataRDD.foldByKey(0)(_+_).collect().foreach(println)

result:
(a,9)
(c,14)

combineByKey

函数签名
def combineByKey[C]( createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]
函数说明
最通用的对 key-value 型 rdd 进行聚集操作的聚集函数（aggregation function）。类似于aggregate()，combineByKey()允许用户返回值的类型与输入不一致。

(func)数据 List((“a”, 88), (“b”, 95), (“a”, 91), (“b”, 93), (“a”, 95), (“b”, 98)),求每个 key 的平均值:
code:

val source: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)))
    val value1: RDD[(String, (Int, Int))] = source.combineByKey(
      (_, 1),
      (t, v) => (t._1 + v, t._2 + 1),
      (t1, t2) => (t1._1 + t2._1, t1._2 + t2._2)
    )
    //      (a,(274,3))
    //      (b,(286,3))
    value1.map(t => (t._1, t._2._1 / t._2._2)).collect().foreach(println)

result:
(a,91)
(b,95)
legend:

comparison：
reduceByKey、foldByKey、aggregateByKey、combineByKey 的区别？

• reduceByKey: 相同 key 的第一个数据不进行任何计算，分区内和分区间计算规则相同
• foldByKey: 相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则相同
• aggregateByKey：相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则可以不相同
• combineByKey:当计算时，发现数据结构不满足要求时，可以让第一个数据转换结构，分区内和分区间计算规则不相同。

sortByKey

函数签名
def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]
函数说明
在一个(K,V)的 RDD 上调用，K 必须实现 Ordered 接口(特质)，返回一个按照 key 进行排序的

e.g.
code:

val dataRDD1 = sc.makeRDD(List(("a",1),("b",2),("c",3)))
val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(true)
val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(false)

(func)设置key为自定义类User

join

函数签名
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
函数说明
在类型为(K,V)和(K,W)的 RDD 上调用，返回一个相同 key 对应的所有元素连接在一起的(K,(V,W))的 RDD
e.g.
code:

val dataRDD: RDD[(Int, String)] = sparkContext.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))
    val dataRDD1: RDD[(Int, Int)] = sparkContext.makeRDD(Array((1, 3), (2, 4), (3, 5)))
    
    dataRDD.join(dataRDD1).collect().foreach(println)

result:
(1,(a,3))
(2,(b,4))
(3,(c,5))
note:
① 如果key存在不相等的情况，不相等的部分不参与join，其他部分正常参与
code：

val dataRDD2: RDD[(Int, String)] = sparkContext.makeRDD(Array((1, "a"), (2, "b"), (4, "c")))
    val dataRDD3: RDD[(Int, Int)] = sparkContext.makeRDD(Array((1, 3), (2, 4), (3, 5)))
    dataRDD2.join(dataRDD3).collect().foreach(println)

result：
(1,(a,3))
(2,(b,4))

②如果两个数据源中key没有匹配上，那么数据不会出现在结果中
③如果两个数据源中key有多个相同的，会依次匹配，产生笛卡尔积

leftOuterJoin/rightOuterJoin

函数签名
def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
def rightOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
函数说明
类似于 SQL 语句的左、右外连接
e.g.
code:

val dataRDD1 = sc.makeRDD(List(("a",1),("b",2),("c",4)))
val dataRDD2 = sc.makeRDD(List(("a",1),("b",2),("d",3)))
// leftOuterJoin
val rdd1 = dataRDD1.leftOuterJoin(dataRDD2)
println("dataRDD1:dataRDD2")
rdd1.collect().foreach(println)
val rdd2 = dataRDD2.leftOuterJoin(dataRDD1)
println("dataRDD2:dataRDD1")
rdd2.collect().foreach(println)
// rightOuterJoin
val rdd3 = dataRDD1.rightOuterJoin(dataRDD2)
println("dataRDD1:dataRDD2")
rdd3.collect().foreach(println)
val rdd4 = dataRDD2.rightOuterJoin(dataRDD1)
println("dataRDD2:dataRDD1")
rdd4.collect().foreach(println)

result:
leftOuterJoin:

rightOuterJoin:

cogroup

函数签名
def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]
函数说明
在类型为(K,V)和(K,W)的 RDD 上调用，返回一个(K,(Iterable

,Iterable

))类型的 RDD

note:cogroup = connect + group，最多支持四个RDD的操作

e.g.1

code:

val dataRDD1 = sc.makeRDD(List(("a",1),("a",2),("c",3)))
val dataRDD2 = sc.makeRDD(List(("a",1),("c",2),("d",3)))
val value= dataRDD1.cogroup(dataRDD2)
value.collect().foreach(println)

result:
(a,(CompactBuffer(1, 2),CompactBuffer(1)))
(c,(CompactBuffer(3),CompactBuffer(2)))
(d,(CompactBuffer(),CompactBuffer(3)))
首先按照key进行分组，然后将分好的组进行连接；可能会存在空组的情况，如上图所示
e.g.2
code:

val dataRDD1 = sc.makeRDD(List(("a",1),("a",2),("c",3)))
  val dataRDD2 = sc.makeRDD(List(("a",1),("c",2),("d",3)))
  val dataRDD3 = sc.makeRDD(List(("a",11),("c",22),("a",33)))
  val dataRDD4 = sc.makeRDD(List(("a",1),("a",2),("d",3)))

  dataRDD1.cogroup(dataRDD2,dataRDD3,dataRDD4).collect().foreach(println)

result:
(a,(CompactBuffer(1, 2),CompactBuffer(1),CompactBuffer(11, 33),CompactBuffer(1, 2)))
(c,(CompactBuffer(3),CompactBuffer(2),CompactBuffer(22),CompactBuffer()))
(d,(CompactBuffer(),CompactBuffer(3),CompactBuffer(),CompactBuffer(3)))