Spark中的Transformations和Actions介绍

深藏阁楼爱情的钟 2023-06-10 03:29 151阅读 0赞

#### RDD提供了两种类型的操作：`transformation`和`action` ####

1.  所有的`transformation`都是采用的懒策略，如果只是将`transformation`提交是不会执行计算的，计算只有在`action`被提交的时候才被触发。
2.  action操作：action是得到一个值，或者一个结果（直接将RDD cache到内存中）

**常见的转换操作有**

<table> 
 <thead> 
 <tr> 
 <th>Transformation 算子</th> 
 <th align="center">含义</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>map(func)</td> 
 <td align="center">对原 RDD 中每个元素运用 func 函数，并生成新的 RDD</td> 
 </tr> 
 <tr> 
 <td>filter(func)</td> 
 <td align="center">对原 RDD 中每个元素使用func 函数进行过滤，并生成新的 RDD</td> 
 </tr> 
 <tr> 
 <td>flatMap(func)</td> 
 <td align="center">与 map 类似，但是每一个输入的 item 被映射成 0 个或多个输出的 items（ func 返回类型需要为 Seq ）</td> 
 </tr> 
 <tr> 
 <td>mapPartitions(func)</td> 
 <td align="center">与 map 类似，但函数单独在 RDD 的每个分区上运行， func函数的类型为 Iterator&lt;T&gt; =&gt; Iterator&lt;U&gt; ，其中 T 是 RDD 的类型，即 RDD[T]</td> 
 </tr> 
 <tr> 
 <td>mapPartitionsWithIndex(func)</td> 
 <td align="center">与 mapPartitions 类似，但 func 类型为 (Int, Iterator&lt;T&gt;) =&gt; Iterator&lt;U&gt; ，其中第一个参数为分区索引</td> 
 </tr> 
 <tr> 
 <td>sample(withReplacement, fraction, seed)</td> 
 <td align="center">数据采样，有三个可选参数：设置是否放回（withReplacement）、采样的百分比（fraction）、随机数生成器的种子（seed）；</td> 
 </tr> 
 <tr> 
 <td>union(otherDataset)</td> 
 <td align="center">合并两个 RDD</td> 
 </tr> 
 <tr> 
 <td>intersection(otherDataset)</td> 
 <td align="center">求两个 RDD 的交集</td> 
 </tr> 
 <tr> 
 <td>distinct([numPartitions]))</td> 
 <td align="center">去重</td> 
 </tr> 
 <tr> 
 <td>groupByKey([numPartitions])</td> 
 <td align="center">按照 key 值进行分区，即在一个 (K, V) 对的 dataset 上调用时，返回一个 (K, Iterable&lt;V&gt;) Note: 如果分组是为了在每一个 key 上执行聚合操作（例如，sum 或 average)，此时使用 <code>reduceByKey</code> 或 <code>aggregateByKey</code> 性能会更好 Note: 默认情况下，并行度取决于父 RDD 的分区数。可以传入 <code>numTasks</code> 参数进行修改。</td> 
 </tr> 
 <tr> 
 <td>reduceByKey(func, [numPartitions])</td> 
 <td align="center">按照 key 值进行分组，并对分组后的数据执行归约操作。</td> 
 </tr> 
 <tr> 
 <td>aggregateByKey(zeroValue)(seqOp, combOp, [numPartitions])</td> 
 <td align="center">当调用（K，V）对的数据集时，返回（K，U）对的数据集，其中使用给定的组合函数和 zeroValue 聚合每个键的值。与 groupByKey 类似，reduce 任务的数量可通过第二个参数进行配置。</td> 
 </tr> 
 <tr> 
 <td>sortByKey([ascending], [numPartitions])</td> 
 <td align="center">按照 key 进行排序，其中的 key 需要实现 Ordered 特质，即可比较</td> 
 </tr> 
 <tr> 
 <td>join(otherDataset, [numPartitions])</td> 
 <td align="center">在一个 (K, V) 和 (K, W) 类型的 dataset 上调用时，返回一个 (K, (V, W)) pairs 的 dataset，等价于内连接操作。如果想要执行外连接，可以使用 <code>leftOuterJoin</code>, <code>rightOuterJoin</code> 和 <code>fullOuterJoin</code> 等算子。</td> 
 </tr> 
 <tr> 
 <td>cogroup(otherDataset, [numPartitions])</td> 
 <td align="center">在一个 (K, V) 对的 dataset 上调用时，返回一个 (K, (Iterable&lt;V&gt;, Iterable&lt;W&gt;)) tuples 的 dataset。</td> 
 </tr> 
 <tr> 
 <td>cartesian(otherDataset)</td> 
 <td align="center">在一个 T 和 U 类型的 dataset 上调用时，返回一个 (T, U) 类型的 dataset（即笛卡尔积）。</td> 
 </tr> 
 <tr> 
 <td>coalesce(numPartitions)</td> 
 <td align="center">将 RDD 中的分区数减少为 numPartitions。</td> 
 </tr> 
 <tr> 
 <td>repartition(numPartitions)</td> 
 <td align="center">随机重新调整 RDD 中的数据以创建更多或更少的分区，并在它们之间进行平衡。</td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <thead> 
 <tr> 
 <th>action算子</th> 
 <th align="center">含义</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>reduce(func)</td> 
 <td align="center">(func)使用函数func执行归约操作</td> 
 </tr> 
 <tr> 
 <td>collect()</td> 
 <td align="center">以一个 array 数组的形式返回 dataset 的所有元素，适用于小结果集。</td> 
 </tr> 
 <tr> 
 <td>count()</td> 
 <td align="center">返回 dataset 中元素的个数。</td> 
 </tr> 
 <tr> 
 <td>first()</td> 
 <td align="center">返回 dataset 中的第一个元素，等价于 take(1)。</td> 
 </tr> 
 <tr> 
 <td>take(n)</td> 
 <td align="center">将数据集中的前 n 个元素作为一个 array 数组返回。</td> 
 </tr> 
 <tr> 
 <td>takeSample(withReplacement, num, [seed])</td> 
 <td align="center">对一个 dataset 进行随机抽样**</td> 
 </tr> 
 <tr> 
 <td>takeOrdered**(n, [ordering])</td> 
 <td align="center">按自然顺序（natural order）或自定义比较器（custom comparator）排序后返回前 n 个元素。只适用于小结果集，因为所有数据都会被加载到驱动程序的内存中进行排序。</td> 
 </tr> 
 <tr> 
 <td>saveAsTextFile(path)</td> 
 <td align="center">将 dataset 中的元素以文本文件的形式写入本地文件系统、HDFS 或其它 Hadoop 支持的文件系统中。Spark 将对每个元素调用 toString 方法，将元素转换为文本文件中的一行记录。</td> 
 </tr> 
 <tr> 
 <td>saveAsSequenceFile(path)</td> 
 <td align="center">将 dataset 中的元素以 Hadoop SequenceFile 的形式写入到本地文件系统、HDFS 或其它 Hadoop 支持的文件系统中。该操作要求 RDD 中的元素需要实现 Hadoop 的 Writable 接口。对于 Scala 语言而言，它可以将 Spark 中的基本数据类型自动隐式转换为对应 Writable 类型。(目前仅支持 Java and Scala)</td> 
 </tr> 
 <tr> 
 <td>saveAsObjectFile(path)</td> 
 <td align="center">使用 Java 序列化后存储，可以使用 <code>SparkContext.objectFile()</code> 进行加载。(目前仅支持 Java and Scala)**</td> 
 </tr> 
 <tr> 
 <td>countByKey**()</td> 
 <td align="center">计算每个键出现的次数。</td> 
 </tr> 
 <tr> 
 <td>foreach(func)</td> 
 <td align="center">遍历 RDD 中每个元素，并对其执行fun函数</td> 
 </tr> 
 </tbody> 
</table>

**Transformations**

1.  `map(func)` 算子

var listRDD = sc.makeRDD(1 to 10)
     listRDD.map((_*2)).foreach(println)

// 输出结果： 2，4，6，8，10，12，14，16，18，20 （这里为了节省篇幅去掉了换行,后文亦同）

1.  `filter(func)`算子

var listRDD = sc.makeRDD(1 to 10)
    listRDD .filter(_%2==0).foreach(println)

// 输出结果： 2，4，6，8，10（这里为了节省篇幅去掉了换行,后文亦同）

1.  `flatMap(func)`  
    **flatMap(func)`与`map`类似，但每一个输入的 item 会被映射成 0 个或多个输出的 items（ *func* 返回类型需要为`Seq\`）**

val list = List(List(1, 2), List(3), List(), List(4, 5))
    sc.parallelize(list).flatMap(_.toList).map(_ * 10).foreach(println)

// 输出结果： 10，20，30，40，50（这里为了节省篇幅去掉了换行,后文亦同）

`flatmap` 流的扁平化，最终输出的数据类型为一维数组`Array[String]`。  
被分割出来的每个数据都作为同一个数组中的相同类型的元素，最终全部被分割出来的数据都存储于同一个一维数组中。  
结论：多行数据被分割后都存储到同一个一维数组Array\[String\]中。  
`flatMap` 这个算子在日志分析中使用概率非常高，这里进行一下演示：拆分输入的每行数据为单个单词，并赋值为 1，代表出现一次，之后按照单词分组并统计其出现总次数，代码如下：

var list = List("hello scala","hello python")
    sc.parallelize(list).flatMap(line=>line.split(" ")).map((_,1)).reduceByKey(_+_).foreach(println)

// 输出结果：(scala,1)(python,1)(hello,2)（这里为了节省篇幅去掉了换行,后文亦同）

1. `mapPartitions(func)` 
 `mapPartitions`算子，与 map 类似，但函数单独在 RDD 的每个分区上运行(map函数作用在每一条数据上)， *func*函数的类型为 `Iterator<T> => Iterator` (其中 T 是 RDD 的类型)，即输入和输出都必须是可迭代类型。

val listRdd: RDD[Int] = sc.makeRDD(1 to 10)
        //mappartition对一个rdd中所有的分区进行遍历
        //优于map算子，减少发到执行器的交互次数
        //可能内存溢出
        val partitions: RDD[Int] = listRdd.mapPartitions(data=>{ data.map(data=>data*2)})

//输出结果：2，4，6，8，10，12，1，16，18，20

1. `mapPartitionsWithIndex(func)` 
 与 mapPartitions 类似，但 *func* 类型为 `(Int, Iterator<T>) => Iterator` ，其中第一个参数为分区索引。

val listRdd: RDD[Int] = sc.makeRDD(1 to 10,2)
        val tupRDD: RDD[(Int, String)] = listRdd.mapPartitionsWithIndex { 
          case (num, datas) => { 
            datas.map((_, "分区号:" + num))
          }
        }

//输出结果(1,分区号:0)，(2,分区号:0)，(3,分区号:0)，(4,分区号:0)，(5,分区号:0)，(6,分区号:1，(7,分区号:1)，(8,分区号:1)，(9,分区号:1)，(10,分区号:1)

1.  `sample(withReplacement, fraction, seed)`  
    以指定的随机种子随机抽样出数量为`fraction`的数据，`withReplacement`表示是抽出的数据是否放回，`true`为有放回的抽样，`false`为无放回的抽样，`seed`用于指定随机数生成器种子。

val rdd: RDD[String] = sc.makeRDD(Array("hello1","hello1","hello2","hello3","hello4","hello5","hello6","hello1","hello1","hello2","hello3"))
        val sampleRDD: RDD[String] = rdd.sample(false,0.7)
        sampleRDD.foreach(println)

//输出结果hello1 hello6 hello1 hello2 hello3 hello5 hello1 hello3

1.  `union(otherDataset)`  
    对源`RDD`和参数`RDD`求并集后返回一个新的`RDD`

val rdd1 = sc.parallelize(1 to 5)
        val rdd2 = sc.parallelize(5 to 10)
        val rdd3 = rdd1.union(rdd2)
        rdd3.collect().foreach(println)

//输出结果1 2 3 4 5 5 6 7 8 9 10

1.  `intersection(otherDataset)`  
    对源`RDD`和参数`RDD`求交集后返回一个新的`RDD`

val rdd6 = sc.parallelize(1 to 7)
    val rdd7 = sc.parallelize(5 to 10)
    val rdd8 = rdd6.intersection(rdd7)
    rdd8.collect().foreach(println)

//输出结果5 6 7

1.  `distinct([numPartitions]))`  
    对源`RDD`进行去重后返回一个新的`RDD`。默认情况下，只有8个并行任务来操作，但是可以传入一个可选的`numTasks`参数改变它。

val makeRDD: RDD[Int] = sc.makeRDD(Array(1,3,2,5,6,3,2,1))
    val makedist: RDD[Int] = makeRDD.distinct() //不指定并行度
    /**对RDD（指定并行度为2） makeRDD.distinct(2)*/
    makedist.collect().foreach(println)

//输出结果1 2 3 5 6

1.  `groupByKey([numPartitions])`  
    `groupByKey`也是对每个`key`进行操作，但只生成一个`sequence`。

val words = Array("one","two","three","one","two","three")
        val listRdd: RDD[(String, Int)] = sc.makeRDD(words).map(word => (word,1))
        listRdd.collect().foreach(println)
        //groupByKey算子
        val group: RDD[(String, Iterable[Int])] = listRdd.groupByKey()
        group.collect().foreach(println)

//输出结果(two,CompactBuffer(1, 1)) (one,CompactBuffer(1, 1)) (three,CompactBuffer(1, 1))

1.  `reduceByKey(func, [numPartitions])`  
    在一个`(K,V)的RDD`上调用，返回一个`(K,V)的RDD`，使用指定的`reduce`函数，将相同`key`的值聚合到一起，`reduce`任务的个数可以通过第二个可选的参数来设置。

val words = Array("one","two","three","one","two","three")
     val listRdd: RDD[(String, Int)] = sc.makeRDD(words).map(word => (word,1))
     val reduceByKeyRDD: Array[(String, Int)] = listRdd.reduceByKey(_+_).collect()

//输出结果(two,2) (one,2) (three,2)

1.  `aggregateByKey(zeroValue)(seqOp, combOp, [numPartitions])`  
    参数描述：  
    （1）`zeroValue`：给每一个分区中的每一个key一个初始值；  
    （2）`seqOp`：函数用于在每一个分区中用初始值逐步迭代value；  
    （3）`combOp`：函数用于合并每个分区中的结果  
    在`kv对的RDD`中，，按`key`将`value`进行分组合并，合并时，将每个`value`和初始值作为`seq`函数的参数，进行计算，返回的结果作为一个新的kv对，然后再将结果按照key进行合并，最后将每个分组的`value`传递给`combine`函数进行计算（先将前两个`value`进行计算，将返回结果和下一个`value`传给`combine`函数，以此类推），将key与计算结果作为一个新的kv对输出。

val rdd = sc.parallelize(List(("a",3),("a",2),("c",4),("b",3),("c",6),("c",8)),2)
        val aggregateByKeyRDD: RDD[(String, Int)] = rdd.aggregateByKey(0)(math.max(_,_),_+_)
        aggregateByKeyRDD.collect().foreach(println)

//输出结果(b,3), (a,3), (c,12)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]

1.  `sortByKey([ascending], [numPartitions])`  
    在一个`(K,V)的RDD`上调用，K必须实现`Ordered`接口，返回一个按照key进行排序的`(K,V)的RDD`

val rdd = sc.parallelize(Array((3,"aa"),(6,"cc"),(2,"bb"),(1,"dd")))
        //根据key正序排
        rdd.sortByKey(true).collect().foreach(println)
        //倒序排序
        rdd.sortByKey(false).collect().foreach(println)

//输出结果(1,dd)(2,bb)(3,aa)(6,cc) 和 (6,cc) (3,aa) (2,bb) (1,dd)

1.  `join(otherDataset, [numPartitions])`  
    在类型为`(K,V)和(K,W)`的RDD上调用，返回一个相同key对应的所有元素对在一起的`(K,(V,W))的RDD`，等价于内连接操作。如果想要执行外连接，可以使用 `leftOuterJoin`, `rightOuterJoin` 和 `fullOuterJoin` 等算子。

val rdd = sc.parallelize(Array((3,"aa"),(6,"cc"),(2,"bb"),(1,"dd")))
     val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6)))
      rdd.join(rdd1).collect().foreach(println)

//输出结果(1,(dd,4)) (2,(bb,5)) (3,(aa,6))

1. `cogroup(otherDataset, [numPartitions])` 
 在类型为`(K,V)和(K,W)`的RDD上调用，返回一个`(K,(Iterable<V>,Iterable<W>))`类型的RDD

val rdd = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
        val rdd1 = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
        rdd.cogroup(rdd1).collect().foreach(println)

//输出结果(1,(CompactBuffer(a),CompactBuffer(a))) (2,(CompactBuffer(b),CompactBuffer(b))) (3,(CompactBuffer(c),CompactBuffer(c)))

1.  `cartesian(otherDataset)`  
    计算笛卡尔积

val list1 = List("A", "B", "C")
    val list2 = List(1, 2, 3)
    sc.parallelize(list1).cartesian(sc.parallelize(list2)).foreach(println)

//输出笛卡尔积(A,1)(A,2)(A,3)(B,1)(B,2)(B,3)(C,1)(C,2)(C,3)

1.  `coalesce(numPartitions)`  
    `coalesce`重新分区，可以选择是否进行`shuffle`过程。由参数`shuffle: Boolean = false/true`决定。

val makeRdd: RDD[Int] = sc.makeRDD(1 to 16,4)
        //缩减分区数，可以理解为合并分区，第三个和第四个的数据分区合并了
        val coalesceRdd: RDD[Int] = makeRdd.coalesce(3)
        coalesceRdd.saveAsTextFile("output")
        println(coalesceRdd.partitions.size)

//输出结果：3

1.  `repartition(numPartitions)`  
    根据分区数，重新通过网络随机洗牌所有数据。

val parallelize: RDD[Int] = sc.parallelize(1 to 16,4)
        println(parallelize.partitions.size)
        val rerdd = parallelize.repartition(2)
        println(rerdd.partitions.size)
        val glom: RDD[Array[Int]] = rerdd.glom()
        glom.collect().foreach(array=>{ 
          println(array.mkString(","))
        })

输出结果1,3,5,7,9,11,13,15        2,4,6,8,10,12,14,16

**Actions**

1.  `reduce(func)`  
    使用函数*func*执行归约操作：

val list = List(1, 2, 3, 4, 5)
    sc.parallelize(list).reduce((x, y) => x + y)
    sc.parallelize(list).reduce(_ + _)

// 输出结果 15

1.  `collect()`  
    在驱动程序中，以数组的形式返回数据集的所有元素。

val rdd = sc.parallelize(1 to 10)
        println(rdd.collect.mkString(","))

//输出结果 1,2,3,4,5,6,7,8,9,10

1.  `count()`  
    返回RDD中元素的个数

val rdd = sc.parallelize(1 to 10)
        println(rdd.count)

输出结果 10

1.  `first()`  
    返回RDD中的第一个元素

val rdd = sc.parallelize(1 to 10)
        println(rdd.first)

输出结果 1

1.  `take(n)`  
    返回一个由RDD的前n个元素组成的数组

val rdd = sc.parallelize(Array(2,5,4,6,8,3))
        println(rdd.take(3).mkString(","))

输出结果 2 5 4

1.  `takeOrdered(n)`  
    返回该RDD排序后的前n个元素组成的数组

val makeRDD: RDD[Int] = sc.makeRDD(Array(2,5,4,6,3,8))
        val ordered: Array[Int] = makeRDD.takeOrdered(3)
        ordered.foreach(println)

输出结果 2 3 4

1.  `saveAsTextFile`  
    将 `dataset` 中的元素以文本文件的形式写入本地文件系统、HDFS 或其它 Hadoop 支持的文件系统中。Spark 将对每个元素调用 toString 方法，将元素转换为文本文件中的一行记录。

val list = List(("hadoop", 10), ("hadoop", 10), ("storm", 3), ("storm", 3), ("azkaban", 1))
    sc.parallelize(list).saveAsTextFile("/usr/file/temp")

1.  `countByKey()`  
    针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。

val rdd = sc.parallelize(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)
        println(rdd.countByKey)

输出结果 Map(3 -> 2, 1 -> 3, 2 -> 1)

1.  `foreach(func)`  
    在数据集的每一个元素上，运行函数func进行更新

var rdd = sc.makeRDD(1 to 5,2)
        rdd.foreach(println(_))

输出结果 1 2 3 4 5

完整代码托管于[https://github.com/daizikaikou/learningSpark][https_github.com_daizikaikou_learningSpark]

参考:  
[https://spark.apache.org/docs/latest/rdd-programming-guide.html][https_spark.apache.org_docs_latest_rdd-programming-guide.html]  
[https://www.jianshu.com/p/64aab52fbb21][https_www.jianshu.com_p_64aab52fbb21]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]: /images/20230531/284bcd7a1c444df0b09ae148db07901b.png
[https_github.com_daizikaikou_learningSpark]: https://github.com/daizikaikou/learningSpark
[https_spark.apache.org_docs_latest_rdd-programming-guide.html]: https://spark.apache.org/docs/latest/rdd-programming-guide.html
[https_www.jianshu.com_p_64aab52fbb21]: https://www.jianshu.com/p/64aab52fbb21