3.2.4 spark体系之分布式计算-spark-core之离线计算-Spark的算子总结

红太狼 2022-11-21 09:56 16阅读 0赞

**目录**

1.写在前面

2.功能分类

（一）Value数据类型的Transformation算子：

（二）Key-Value数据类型的Transformation算子

（三）Action算子

Value数据类型的Transformation算子

1.map算子

2.flatMap算子

3.mapPartitions算子

4.mapPartitionsWithIndex(function)算子

5.union算子

6.cartesian算子

7.groupBy算子

8.filter算子

9.distinct算子

10.subtract算子

11.sample算子

(12.13)cache算子、persist算子

Key-Value数据类型的Transformation算子

14.mapValues算子

15.flatMapValues算子

16.combineByKey算子

17.reduceByKey算子

18.repartition算子

19.cogroup算子

20.join算子

21.leftOutJoin、rightOutJoin算子、fullOuterJoin算子

22.intersection算子

Action行动算子

23.foreach算子

24.saveAsTextFile算子

25.saveAsObjectFile算子

26.collect算子

27.collectAsMap算子

28.count算子

29.countByKey算子、countByValue算子

30、take、takeSample算子

31、reduce算子

32.Aggregate算子----transformation类算子

33.zip算子 ---Transformation类算子

33、zipWithIndex算子---Transformation类算子

--------------------

# 1.写在前面 #

从大方向说，Spark算子大致可以分为以下两类：

（1）Transformation变换/转换算子：这种变换并不触发提交作业，这种算子是延迟执行的，也就是说从一个RDD转换生成另一个RDD的转换操作不是马上执行，需要等到有Action操作的时候才会真正触发。

（2）Action行动算子：这类算子会触发SparkContext提交job作业，并将数据输出到Spark系统。

从小方向说，Spark算子大致可以分为以下三类：

（1）**Value数据类型的Transformation算子**，这种变换并不触发提交作业，针对处理的数据项是Value型的数据

（2）**Key-Value 数据类型的Transformation算子**，这种变换并不触发提交作业，针对处理的数据项是Key-Value型的数据对

（3）**Action算子**：这类算子会触发SparkContext提交Job作业

# 2.功能分类 #

## **（一）Value数据类型的Transformation算子：** ##

**（1）输入分区与输出分区一对一型：**

1.map算子

2.flatMap算子

3.mapPartitions算子

4.mapPartitionsWithIndex算子

**（2）输入分区与输出分区多对一型**

5.union算子

6.cartesian算子

**（3）输入分区与输出分区多对多型**

7.groupBy算子、groupByKey算子

**（4）输出分区是输入分区子集类型**

8.filter算子

9.distinct算子

10.subtract算子

11.sample算子

**（5）Cache算子**

12.cache算子

13.persist算子

## **（二）Key-Value数据类型的Transformation算子** ##

**（1）输入分区与输出分区一对一**

14.mapValues算子

15.flatMapValues算子

**（2）对单个RDD或者两个RDD聚集**

单个RDD聚集

16.combineByKey算子

17.reduceByKey算子

18.repartition算子

两个RDD聚集

19.cogroup算子

**（3）连接**

20.join算子

21.leftOutJoin和rightOutJoin算子、fullOuterJoin算子

22.intersection算子

## **（三）Action算子** ##

**（1）无输出**

23.foreach算子

**（2）HDFS**

24.saveAsTextFile算子

25.saveAsObjectFile算子

**（3）Scala集合和数据类型**

26.collect算子

27.collectAsMap算子

28.count算子

29.countByKey，CountByValue算子

30.take、takeSample算子

31.reduce算子

32.aggregate算子

33.zip、zipWithIndex算子

# **Value数据类型的Transformation算子** #

## **1.map算子** ##

处理数据是一对一的关系，进入一条数据，出去的还是一条数据。map的输入变换函数应用于RDD中所有的元素，而mapPartitions应用于所有分区。区别于mapPartitions主要在于调度粒度不同。如parallelize（1 to 10 ，3），map函数执行了10次，而mapPartitions函数执行了3次。

val infos: RDD[String] = sc.parallelize(Array[String]("hello spark","hello hdfs","hello HBase"))
    val result: RDD[Array[String]] = infos.map(one => {
      one.split(" ")
    })
    result.foreach(arr =>{arr.foreach(println)})

执行结果：

![97992402218be7e786855ec8c8cdd75b.png][]

## **2.flatMap算子** ##

flatMap是一对多的关系，处理一条数据得到多条结果

将原来 RDD 中的每个元素通过函数 f 转换为新的元素，并将生成的 RDD 的每个集合中的元素合并为一个集合。

![ee71f6b88813e3f4b2f4035392eed33e.png][]

val infos: RDD[String] = sc.makeRDD(Array[String]("hello spark","hello hdfs","hello MapReduce"))
    val rdd1: RDD[String] = infos.flatMap(one => {
      one.split(" ")
    })
    rdd1.foreach(println)

## **3.mapPartitions算子** ##

mapPartitions遍历的是每一个分区中的数据，一个个分区的遍历。获 取 到 每 个 分 区 的 迭 代器，在 函 数 中 通 过 这 个 分 区 整 体 的 迭 代 器 对整 个 分 区 的 元 素 进 行 操 作，相对于map一条条处理数据，性能比较高，可获取返回值。

可以通过函数f(iter) =>iter.filter(\_>=3)对分区中所有的数据进行过滤，大于和等于3的数据保留，一个方块代表一个RDD分区，含有1,2,3的分区过滤，只剩下元素3。

![8a2447cb33b297b95d5ca2a107186012.png][]

## **4.mapPartitionsWithIndex(function)算子** ##

拿到每个RDD中的分区，以及分区中的数据

val lines: RDD[String] = sc.textFile("./data/words",5)
        val result: RDD[String] = lines.mapPartitionsWithIndex((index, iter) => {
          val arr: ArrayBuffer[String] = ArrayBuffer[String]()
          iter.foreach(one => {
    //        one.split(" ")
            arr.append(s"partition = [$index] ,value = $one")
          })
          arr.iterator
        }, true)
        result.foreach(println)

![add4bf339fec01b6ec5d5cc8f65de1b0.png][]

## **5.union算子** ##

union合并两个RDD，两个RDD必须是同种类型，不一定是K，V格式的RDD

val rdd1: RDD[String] = sc.parallelize(List[String]("zhangsan","lisi","wangwu","maliu"),3)
    val rdd2: RDD[String] = sc.parallelize(List[String]("a","b","c","d"),4)
    val unionRDD: RDD[String] = rdd1.union(rdd2)
    unionRDD.foreach(println)

## **6.cartesian算子** ##

求笛卡尔积，该操作不会执行shuffle操作，但最好别用，容易触发OOM

## **7.groupBy算子** ##

按照指定的规则，将数据分组

![0139cdd6777e21292dff1b90e6591365.png][]

**groupByKey算子**

根据key去将相同的key对应的value合并在一起（K,V）=>(K,\[V\])

val rdd: RDD[(String, Double)] = sc.parallelize(List[(String,Double)](("zhangsan",66.5),("lisi",33.2),("zhangsan",66.7),("lisi",33.4),("zhangsan",66.8),("wangwu",29.8)))
    val rdd1: RDD[(String, Iterable[Double])] = rdd.groupByKey()
    rdd1.foreach(info=>{
      val name: String = info._1
      val value: Iterable[Double] = info._2
      val list: List[Double] = info._2.toList
      println("name = "+name+",value ="+list)
    })

![7c9747344f05763a2a1f76bb54f4e8e5.png][]

## **8.filter算子** ##

过滤数据，返回true的数据会被留下

val infos: RDD[String] = sc.makeRDD(List[String]("hehe","hahha","zhangsan","lisi","wangwu"))
    val result: RDD[String] = infos.filter(one => {
      !one.equals("zhangsan")
    })
    result.foreach(println)

![c03b00988170a5db09b543b85d1ed849.png][]

## **9.distinct算子** ##

distinct去重，有shuffle产生，内部实际是map+reduceByKey+map实现

val infos: RDD[String] = sc.parallelize(List[String]("a","a","b","a","b","c","c","d"),4)
    val result: RDD[String] = infos.distinct()
    result.foreach(println)

![fab8038489d8c1dcff053f160355fa51.png][]

## **10.subtract算子** ##

取RDD的差集，subtract两个RDD的类型要一致，结果RDD的分区数与subtract算子前面的RDD分区数多的一致。

val rdd1 = sc.parallelize(List[String]("zhangsan","lisi","wangwu"),5)
    val rdd2 = sc.parallelize(List[String]("zhangsan","lisi","maliu"),4)
    val subtractRDD: RDD[String] = rdd1.subtract(rdd2)
    subtractRDD.foreach(println)
    println("subtractRDD partition length = "+subtractRDD.getNumPartitions)

![25f87e6ab8e15e7cba0fe8d3e5807776.png][]

## **11.sample算子** ##

sample随机抽样，参数sample（withReplacement:有无放回抽样，fraction:抽样的比例，seed:用于指定的随机数生成器的种子）

有种子和无种子的区别：

有种子是只要针对数据源一样，都是指定相同的参数，那么每次抽样到的数据都是一样的

没有种子是针对同一个数据源，每次抽样都是随机抽样

![71369955bedaf9a972fa810da2379c8d.png][]

![5c96230dc8bc119b259b68c633e78134.png][]

## **(12.13)cache算子、persist算子** ##

![54250e1147982a3dc2c804cb8e4d9405.png][]

package core.persist
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.storage.StorageLevel
    import org.apache.spark.{SparkConf, SparkContext}
    
    /**
      * cache()和persist()注意问题
      * 1.cache()和persist()持久化单位是partition，cache()和persist()是懒执行算子，需要action算子触发执行
      * 2.对一个RDD使用cache或者persist之后可以赋值给一个变量，下次直接使用这个变量就是使用持久化的数据。
      * 也可以直接对RDD进行cache或者persist，不赋值给一个变量
      * 3.如果采用第二种方法赋值给变量的话，后面不能紧跟action算子
      * 4.cache()和persist()的数据在当前application执行完成之后会自动清除
      */
    object CacheAndPersist {
      def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setMaster("local").setAppName("cache")
        val sc: SparkContext = new SparkContext(conf)
        sc.setLogLevel("Error")
        val lines: RDD[String] = sc.textFile("./data/persistData.txt")
    
        //    val linescache: RDD[String] = lines.persist(StorageLevel.MEMORY_ONLY)
        val linescache: RDD[String] = lines.cache()
    
        val startTime1: Long = System.currentTimeMillis()
        val count1: Long = linescache.count()
        val endTime1: Long = System.currentTimeMillis()
        println("count1 = "+count1+". time = "+(endTime1-startTime1) + "mm")
    
        val starttime2: Long = System.currentTimeMillis()
        val count2: Long = linescache.count()
        val endTime2: Long = System.currentTimeMillis()
        println("count2 = "+count2+", time = "+(endTime2-starttime2) + "ms")
        sc.stop()
      }
    }

# **Key-Value数据类型的Transformation算子** #

## **14.mapValues算子** ##

针对K，V格式的数据，只对Value做操作，Key保持不变

val infos: RDD[(String, String)] = sc.makeRDD(
        List[(String,String)](
        ("zhangsan","18"),("lisi","20"),("wangwu","30")
        ))
    val result: RDD[(String, String)] = infos.mapValues(s => {
      s + " " + "zhangsan18"
    })
    result.foreach(println)
    sc.stop()

![d074661c5175c28f05e53d3e908e7856.png][]

## **15.flatMapValues算子** ##

（K,V）->(K,V)，作用在K,V格式的RDD上，对一个Key的一个Value返回多个Value

val infos: RDD[(String, String)] = sc.makeRDD(
        List[(String,String)](
        ("zhangsan","18"),("lisi","20"),("wangwu","30")
        ))
        val transInfo: RDD[(String, String)] = infos.mapValues(s => {
          s + " " + "zhangsan18"
        })
    //    transInfo.foreach(println)
        val result: RDD[(String, String)] = transInfo.flatMapValues(s => {
          //按空格切分
          s.split(" ")
        })
        result.foreach(println)
        sc.stop()

![f006dba94f5f4d3b9e0ba2616d037667.png][]

## **16.combineByKey算子** ##

首先给RDD中每个分区中的每一个key一个初始值

其次在RDD每个分区内部相同的key聚合一次

再次在RDD不同的分区之间将相同的key结果聚合一次

val rdd1: RDD[(String, Int)] = sc.makeRDD(List[(String, Int)](
      ("zhangsan", 10), ("zhangsan", 20), ("wangwu", 30),
      ("lisi", 40), ("zhangsan", 50), ("lisi", 60),
      ("wangwu", 70), ("wangwu", 80), ("lisi", 90)
    ),3)
    rdd1.mapPartitionsWithIndex((index,iter)=>{
      val arr: ArrayBuffer[(String, Int)] = ArrayBuffer[(String,Int)]()
      iter.foreach(tp=>{
        arr.append(tp)
        println("rdd1 partition index ="+index+".value ="+tp)
      })
      arr.iterator
    }).count()
    println("++++++++++++++++++++++++++++++++++++")
    val result: RDD[(String, String)] = rdd1.combineByKey(v=>{v+"hello"}, (s:String, v)=>{s+"@"+v}, (s1:String, s2:String)=>{s1+"#"+s2})
    result.foreach(println)

![eee4438f9f86a85ad4e84f707ea4a66b.png][]

## **17.reduceByKey算子** ##

首先会根据key去分组，然后在每一组中将value聚合，作用在KV格式的RDD上

首先会根据key去分组，然后在每一组中将value聚合，作用在KV格式的RDD上
    val infos: RDD[(String, Int)] = sc.parallelize(
        List[(String,Int)](("zhangsan",1),("zhangsan",2),
        ("zhangsan",3),("lisi",100),("lisi",200)),5
        )
    val result: RDD[(String, Int)] = infos.reduceByKey((v1, v2)=>{v1+v2})
    result.foreach(println)
    sc.stop()

![164f16d94269c72e30aec99cffb2fa65.png][]

## **18.repartition算子** ##

重新分区，可以将RDD的分区增多或者减少，会产生shuffle，coalesc(num,true) = repartition(num)

val rdd1: RDD[String] = sc.parallelize(List[String](
      "love1", "love2", "love3", "love4",
      "love5", "love6", "love7", "love8",
      "love9", "love10", "love11", "love12"
    ), 3)
    val rdd2: RDD[String] = rdd1.mapPartitionsWithIndex((index, iter) => {
      val list: ListBuffer[String] = ListBuffer[String]()
      iter.foreach(one => {
        list.append(s"rdd1 partition = [$index] ,value = [$one]")
      })
      list.iterator
    }, true)
    val rdd3: RDD[String] = rdd2.repartition(3)
    val rdd4: RDD[String] = rdd3.mapPartitionsWithIndex((index, iter) => {
      val arr: ArrayBuffer[String] = ArrayBuffer[String]()
      iter.foreach(one => {
        arr.append(s"rdd3 partition = [$index] ,value = [$one]")
      })
      arr.iterator
    })
    val results: Array[String] = rdd4.collect()
    results.foreach(println)
    sc.stop()

![eebd068295e0770571dadf8dc22f77a8.png][]

## **19.cogroup算子** ##

合并两个RDD,生成一个新的RDD。分区数与分区数多个那个RDD保持一致

val rdd1 = sc.parallelize(List[(String,String)](("zhangsan","female"),("zhangsan","female1"),("lisi","male"),("wangwu","female"),("maliu","male")),3)
    val rdd2 = sc.parallelize(List[(String,Int)](("zhangsan",18),("lisi",19),("lisi",190),("wangwu",20),("tianqi",21)),4)
    val resultRDD: RDD[(String, (Iterable[String], Iterable[Int]))] = rdd1.cogroup(rdd2)
    
    resultRDD.foreach(info=>{
      val key = info._1
      val value1: List[String] = info._2._1.toList
      val value2: List[Int] = info._2._2.toList
      println("key ="+key+",value"+value1+", value2 = "+value2)
    })
    println("resultRDD partition length ="+resultRDD.getNumPartitions)
    sc.stop()

![66fdfce40334380b162734f11df33afa.png][]

## **20.join算子** ##

会产生shuffle,(K,V)格式的RDD和(K,V)格式的RDD按照相同的K，join得到(K,(V,W))格式的数据，分区数按照大的来。

val nameRDD: RDD[(String, String)] = sc.parallelize(List[(String,String)](("zhangsan","female"),("lisi","male"),("wangwu","female")),3)
    val scoreRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",18),("lisi",19),("wangwu",20)),2)
    val joinRDD: RDD[(String, (String, Int))] = nameRDD.join(scoreRDD)
    println(joinRDD.getNumPartitions)
    joinRDD.foreach(println)

![b3151ae437cbc210e9395e374f3f1ac0.png][]

## **21.leftOutJoin、rightOutJoin算子、fullOuterJoin算子** ##

leftOuterJoin(K,V)格式的RDD和(K,V)格式的RDD，使用leftOuterJoin结合，以左边的RDD出现的key为主 ，得到(K,(V,Option(W)))

val nameRDD: RDD[(String, String)] = sc.parallelize(
        List[(String,String)](("zhangsan","female"),
        ("lisi","male"),("wangwu","female"),("maliu","male")
        ))
    val scoreRDD: RDD[(String, Int)] = sc.parallelize(
        List[(String,Int)](("zhangsan",22),("lisi",19),
        ("wangwu",20),("tianqi",21)
        ))
    val leftOutJoin: RDD[(String, (String, Option[Int]))] = nameRDD.leftOuterJoin(scoreRDD)
    leftOutJoin.foreach(println)
    sc.stop()

![0ec1f4037161bd9dda490fa76ae489e3.png][]

rightOuterJoin(K,V)格式的RDD和（K,W）格式的RDD使用rightOuterJoin结合以右边的RDD出现的key为主，得到(K,(Option(V),W))

val nameRDD: RDD[(String, String)] = sc.parallelize(
        List[(String,String)](("zhangsan","female"),("lisi","male")
        ,("wangwu","female"),("maliu","male")),3
        )
      val scoreRDD: RDD[(String, Int)] = sc.parallelize(
          List[(String,Int)](("zhangsan",18),("lisi",19),
          ("wangwu",20),("tianqi",21)),4
          )
      val rightOuterJoin: RDD[(String, (Option[String], Int))] = nameRDD.rightOuterJoin(scoreRDD)
      rightOuterJoin.foreach(println)
      println("rightOuterJoin RDD partiotion length = "+rightOuterJoin.getNumPartitions)
      sc.stop()

![cfca0374a6fec3069117e73c7eb28fc7.png][]

fullOuterJoin算子(K,,V)格式的RDD和(K,V)格式的RDD，使用fullOuterJoin结合是以两边的RDD出现的key为主，得到(K(Option(V),Option(W)))

val nameRDD: RDD[(String, String)] = sc.parallelize(List[(String,String)](("zhangsan","female"),("lisi","male"),("wangwu","female"),("maliu","male")),3)
    val ageRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",18),("lisi",16),("wangwu",20),("tianqi",21)),4)
    val fullOuterJoin: RDD[(String, (Option[String], Option[Int]))] = nameRDD.fullOuterJoin(ageRDD)
    fullOuterJoin.foreach(println)
    println("fullOuterJoin RDD partition length = "+fullOuterJoin.getNumPartitions)
    sc.stop()

![3b81327e71a3c698d7192826ffb91a62.png][]

## **22.intersection算子** ##

intersection取两个RDD的交集，两个RDD的类型要一致，结果RDD的分区数要与两个父RDD多的那个一致

val rdd1: RDD[String] = sc.parallelize(List[String]("zhangsan","lisi","wangwu"),5)
    val rdd2: RDD[String] = sc.parallelize(List[String]("zhangsan","lisi","maliu"),4)
    val result: RDD[String] = rdd1.intersection(rdd2)
    result.foreach(println)
    println("intersection partition length = "+ result.getNumPartitions)
    sc.stop()

![333827b7d471b17a0334d74bdbcea31b.png][]

# Action行动算子 #

## **23.foreach算子** ##

foreach遍历RDD中的每一个元素

val lines: RDD\[String\] = sc.textFile("./data/words") lines.foreach(println)

## **24.saveAsTextFile算子** ##

将DataSet中的元素以文本的形式写入本地文件系统或者HDFS中，Spark将会对每个元素调用toString方法，将数据元素转换成文本文件中的一行数据，若将文件保存在本地文件系统，那么只会保存在executor所在机器的本地目录

val infos: RDD\[String\] = sc.parallelize(List\[String\]("a","b","c","e","f","g"),4)

infos.saveAsTextFile("./data/infos")

![7db4b10573ad20f8dcb0b7766aaf2214.png][]

![8395278e61d6f4f3482272838f379265.png][]

## **25.saveAsObjectFile算子** ##

将数据集中元素以ObjectFile形式写入本地文件系统或者HDFS中

infos.saveAsObjectFile("./data/infosObject")

![8db65e358cd6941c41e9dff2619dfe10.png][]

## **26.collect算子** ##

collect回收算子，会将结果回收到Driver端，如果结果比较大,就不要回收，这样的话会造成Driver端的OOM

val lines: RDD[String] = sc.textFile("./data/words")
    sc.setLogLevel("Error")
    val result: Array[String] = lines.collect()
    result.foreach(println)

![754eba2193526d0c3b10888ed46d6f0d.png][]

## **27.collectAsMap算子** ##

将K、V格式的RDD回收到Driver端作为Map使用

val weightInfos: RDD[(String, Double)] = sc.parallelize(
        List[(String,Double)](new Tuple2("zhangsan",99),
            new Tuple2("lisi",78.6),
            new Tuple2("wangwu",122.2323)
            )
            )
    val stringToDouble: collection.Map[String, Double] = weightInfos.collectAsMap()
    stringToDouble.foreach(tp=>{
      println(tp._1+"**************"+tp._2)
    })
    sc.stop()

![5ea6cdcf3a4eb0a388c4c678e0feb45a.png][]

## **28.count算子** ##

count统计RDD共有多少行数据

val lines: RDD\[String\] = sc.textFile("./data/sampleData.txt")

val result: Long = lines.count()

println(result)

sc.stop()

直接给出结果行数

## **29.countByKey算子、countByValue算子** ##

countByKey统计相同的key出现的个数

val rdd: RDD[(String, Integer)] = sc.makeRDD(List[(String,Integer)](
        ("a",1),("a",100),("a",1000),("b",2),("b",200),("c",3),("c",4),("d",122)
        ))
    val result: collection.Map[String, Long] = rdd.countByKey()
    result.foreach(println)

![252f4a5dad2ae9009aa9f7bc96b02153.png][]

countByValue统计RDD中相同的Value出现的次数，不要求数据必须为RDD格式

val rdd = sc.makeRDD(List[(String,Integer)](
        ("a",1),("a",1),("a",1000),("b",2),("b",200),("c",3),("c",3)
        ))
    val result: collection.Map[(String, Integer), Long] = rdd.countByValue()
    result.foreach(println)

![d0bd7aa6ba05a2d226eab3ebd161117d.png][]

## **30、take、takeSample算子** ##

take取出RDD中的前N个元素

val lines: RDD\[String\] = sc.textFile("./data/words")

val array: Array\[String\] = lines.take(3)

array.foreach(println)

takeSapmle(withReplacement,num,seed)，随机抽样将数据结果拿回Driver端使用，返回Array，

withReplacement:有无放回抽样，num:抽样的条数，seed:种子

val lines: RDD\[String\] = sc.textFile("./data/words")

val result: Array\[String\] = lines.takeSample(false,3,10)

result.foreach(println)

![55f180488e5b2bc29e98ba84e5189cb7.png][]

## **31、reduce算子和fold算子** ##

// 模拟数据
            val data_rdd: RDD[Int] = sc.parallelize(List(1, 2, 3, 4), 1)
    
            // 测试reduce（返回值必须与rdd的数据类型相同）
            println("测试reduce")
            // 这个x，它代指的是返回值，而y是对rdd各元素的遍历
            val reduce_data: Int = data_rdd.reduce((x, y) => x + y)
            println(reduce_data) //1+2+3+4 =10
    
            // 测试fold（返回值必须与rdd的数据类型相同）
            println("测试fold")
            // scala的语法确实是比较奇怪的，既然有两个参数，你就不能把两个参数放在一个括号里吗？
            // 也是醉了，这种写法确实容易让人迷惑
            val fold_data: Int = data_rdd.fold(0)((x, y) => x + y)
            println(fold_data) // zeroValue表示初始值

reduce和fold基本上没有区别，fold需要传两个参数，第一个是初始值，但是这个初始值和我们的理解还是不太一样的（先对每个分区操作，再对多个分区合并操作，每一个操作都要加一次这个初始值）。比如我们分区为1的时候：

初始值为10：

*  第一步： 先对分区内元素做op操作， zeroValue作为初始值,  即 10+1+2+3+4 = 20
 *  第二步： 再对多个分区做op操作, zeroValue也作为初始值,  即 10+20 =30

## **32.Aggregate算子---Transformation类算子** ##

reduce和fold要求返回值和rdd类型要求一致，但是aggregate更加灵活，可以得到我们想要的东西，比如求平均值操作，我们需要得到sum的同时还要得到数的个数：

// 测试aggregate
            println("测试aggregate")
            //返回值是(Int, Int)，初始值设置为（0,0）
            //用x._1和x._2来代指这两个元素的，y指代遍历的元素
            val aggregate_tuple: (Int, Int) = data_rdd.aggregate(0, 0)((x, y) => (x._1 + y, x._2 + 1), (x, y) => (x._1 + y._1, x._2 + y._2))
            val aveage: Double = aggregate_tuple._1 / aggregate_tuple._2.toDouble
            println(aveage)

首先是给定RDD的每一个分区一个初始值，然后RDD中每一个分区中按照相同的key，结合初始值去合并，最后RDD之间相同的key聚合

val rdd1: RDD[(String, Int)] = sc.makeRDD(List[(String, Int)](
      ("zhangsan", 10), ("zhangsan", 20), ("wangwu", 30),
      ("lisi", 40), ("zhangsan", 50), ("lisi", 60),
      ("wangwu", 70), ("wangwu", 80), ("lisi", 90)
    ), 3)
    rdd1.mapPartitionsWithIndex((index,iter)=>{
      val arr: ArrayBuffer[(String, Int)] = ArrayBuffer[(String,Int)]()
      iter.foreach(tp=>{
        arr.append(tp)
        println("rdd1 partition index ="+index+", value ="+tp)
      })
      arr.iterator
    }).count()
    val result: RDD[(String, String)] = 
        rdd1.aggregateByKey("hello")(
            (s, v)=>{s+"~"+v}, (s1, s2)=>{s1+"#"+s2}
            )
    result.foreach(println)

![c3fd413169cf1e38bf91755296938ced.png][]

mapPartitionsWithIndex注释掉执行结果：

![92a79f568f8553309f82c3dfdf499bc8.png][]

## **33.zip算子 ---Transformation类算子** ##

将两个RDD合成一个K，V格式的RDD，分区数要相同，每个分区中的元素必须相同

val rdd1: RDD[String] = sc.parallelize(List[String]("a","b","c","d"),2)
    val rdd2: RDD[Int] = sc.parallelize(List[Int](1,2,3,4),2)
    val result: RDD[(String, Int)] = rdd1.zip(rdd2)
    result.foreach(println)

![ac4d187e745017ae9c1157337e4c7793.png][]

## **33、zipWithIndex算子---Transformation类算子** ##

val rdd1 = sc.parallelize(List[String]("a","b","c"),2)
    val rdd2 = sc.parallelize(List[Int](1,2,3),numSlices = 2)
    val result: RDD[(String, Long)] = rdd1.zipWithIndex()
    val result2: RDD[(Int, Long)] = rdd2.zipWithIndex()
    result.foreach(println)
    result2.foreach(println)

![dc6faf7b06f07d5e3846d34e03990e18.png][]

[97992402218be7e786855ec8c8cdd75b.png]: /images/20221120/70f79db1bf724a6bac10a4d1ab0c8a8e.png
[ee71f6b88813e3f4b2f4035392eed33e.png]: /images/20221120/a4db671911dc41009bc1002650d428f5.png
[8a2447cb33b297b95d5ca2a107186012.png]: https://img-blog.csdnimg.cn/img_convert/8a2447cb33b297b95d5ca2a107186012.png
[add4bf339fec01b6ec5d5cc8f65de1b0.png]: https://img-blog.csdnimg.cn/img_convert/add4bf339fec01b6ec5d5cc8f65de1b0.png
[0139cdd6777e21292dff1b90e6591365.png]: https://img-blog.csdnimg.cn/img_convert/0139cdd6777e21292dff1b90e6591365.png
[7c9747344f05763a2a1f76bb54f4e8e5.png]: https://img-blog.csdnimg.cn/img_convert/7c9747344f05763a2a1f76bb54f4e8e5.png
[c03b00988170a5db09b543b85d1ed849.png]: https://img-blog.csdnimg.cn/img_convert/c03b00988170a5db09b543b85d1ed849.png
[fab8038489d8c1dcff053f160355fa51.png]: https://img-blog.csdnimg.cn/img_convert/fab8038489d8c1dcff053f160355fa51.png
[25f87e6ab8e15e7cba0fe8d3e5807776.png]: https://img-blog.csdnimg.cn/img_convert/25f87e6ab8e15e7cba0fe8d3e5807776.png
[71369955bedaf9a972fa810da2379c8d.png]: https://img-blog.csdnimg.cn/img_convert/71369955bedaf9a972fa810da2379c8d.png
[5c96230dc8bc119b259b68c633e78134.png]: https://img-blog.csdnimg.cn/img_convert/5c96230dc8bc119b259b68c633e78134.png
[54250e1147982a3dc2c804cb8e4d9405.png]: https://img-blog.csdnimg.cn/img_convert/54250e1147982a3dc2c804cb8e4d9405.png
[d074661c5175c28f05e53d3e908e7856.png]: https://img-blog.csdnimg.cn/img_convert/d074661c5175c28f05e53d3e908e7856.png
[f006dba94f5f4d3b9e0ba2616d037667.png]: https://img-blog.csdnimg.cn/img_convert/f006dba94f5f4d3b9e0ba2616d037667.png
[eee4438f9f86a85ad4e84f707ea4a66b.png]: https://img-blog.csdnimg.cn/img_convert/eee4438f9f86a85ad4e84f707ea4a66b.png
[164f16d94269c72e30aec99cffb2fa65.png]: https://img-blog.csdnimg.cn/img_convert/164f16d94269c72e30aec99cffb2fa65.png
[eebd068295e0770571dadf8dc22f77a8.png]: https://img-blog.csdnimg.cn/img_convert/eebd068295e0770571dadf8dc22f77a8.png
[66fdfce40334380b162734f11df33afa.png]: https://img-blog.csdnimg.cn/img_convert/66fdfce40334380b162734f11df33afa.png
[b3151ae437cbc210e9395e374f3f1ac0.png]: https://img-blog.csdnimg.cn/img_convert/b3151ae437cbc210e9395e374f3f1ac0.png
[0ec1f4037161bd9dda490fa76ae489e3.png]: https://img-blog.csdnimg.cn/img_convert/0ec1f4037161bd9dda490fa76ae489e3.png
[cfca0374a6fec3069117e73c7eb28fc7.png]: https://img-blog.csdnimg.cn/img_convert/cfca0374a6fec3069117e73c7eb28fc7.png
[3b81327e71a3c698d7192826ffb91a62.png]: https://img-blog.csdnimg.cn/img_convert/3b81327e71a3c698d7192826ffb91a62.png
[333827b7d471b17a0334d74bdbcea31b.png]: https://img-blog.csdnimg.cn/img_convert/333827b7d471b17a0334d74bdbcea31b.png
[7db4b10573ad20f8dcb0b7766aaf2214.png]: https://img-blog.csdnimg.cn/img_convert/7db4b10573ad20f8dcb0b7766aaf2214.png
[8395278e61d6f4f3482272838f379265.png]: https://img-blog.csdnimg.cn/img_convert/8395278e61d6f4f3482272838f379265.png
[8db65e358cd6941c41e9dff2619dfe10.png]: https://img-blog.csdnimg.cn/img_convert/8db65e358cd6941c41e9dff2619dfe10.png
[754eba2193526d0c3b10888ed46d6f0d.png]: https://img-blog.csdnimg.cn/img_convert/754eba2193526d0c3b10888ed46d6f0d.png
[5ea6cdcf3a4eb0a388c4c678e0feb45a.png]: https://img-blog.csdnimg.cn/img_convert/5ea6cdcf3a4eb0a388c4c678e0feb45a.png
[252f4a5dad2ae9009aa9f7bc96b02153.png]: https://img-blog.csdnimg.cn/img_convert/252f4a5dad2ae9009aa9f7bc96b02153.png
[d0bd7aa6ba05a2d226eab3ebd161117d.png]: https://img-blog.csdnimg.cn/img_convert/d0bd7aa6ba05a2d226eab3ebd161117d.png
[55f180488e5b2bc29e98ba84e5189cb7.png]: https://img-blog.csdnimg.cn/img_convert/55f180488e5b2bc29e98ba84e5189cb7.png
[c3fd413169cf1e38bf91755296938ced.png]: https://img-blog.csdnimg.cn/img_convert/c3fd413169cf1e38bf91755296938ced.png
[92a79f568f8553309f82c3dfdf499bc8.png]: https://img-blog.csdnimg.cn/img_convert/92a79f568f8553309f82c3dfdf499bc8.png
[ac4d187e745017ae9c1157337e4c7793.png]: https://img-blog.csdnimg.cn/img_convert/ac4d187e745017ae9c1157337e4c7793.png
[dc6faf7b06f07d5e3846d34e03990e18.png]: https://img-blog.csdnimg.cn/img_convert/dc6faf7b06f07d5e3846d34e03990e18.png