spark系列6：常用RDD介绍与演示

柔光的暖阳◎ 2022-11-10 08:12 179阅读 0赞

## 0. RDD 的算子 ##

目标

1.  理解 RDD 的算子分类, 以及其特性
2.  理解常见算子的使用

分类

RDD 中的算子从功能上分为两大类

1.  Transformation(转换) 它会在一个已经存在的 RDD 上创建一个新的 RDD, 将旧的 RDD 的数据转换为另外一种形式后放入新的 RDD
2.  Action(动作) 执行各个分区的计算任务, 将的到的结果返回到 Driver 中

RDD 中可以存放各种类型的数据, 那么对于不同类型的数据, RDD 又可以分为三类

*  针对基础类型(例如 String)处理的普通算子
 *  针对 `Key-Value` 数据处理的 `byKey` 算子
 *  针对数字类型数据处理的计算算子

特点

*  Spark 中所有的 Transformations 是 Lazy(惰性) 的, 它们不会立即执行获得结果. 相反, 它们只会记录在数据集上要应用的操作. 只有当需要返回结果给 Driver 时, 才会执行这些操作, 通过 DAGScheduler 和 TaskScheduler 分发到集群中运行, 这个特性叫做 **惰性求值**
 *  默认情况下, 每一个 Action 运行的时候, 其所关联的所有 Transformation RDD 都会重新计算, 但是也可以使用 `presist` 方法将 RDD 持久化到磁盘或者内存中. 这个时候为了下次可以更快的访问, 会把数据保存到集群上.

### 1. Transformations 算子 ###

<table> 
 <tbody> 
 <tr> 
 <th> Transformation function </th> 
 <td> 解释 </td> 
 </tr> 
 <tr> 
 <th> <code>map(T ⇒ U)</code> </th> 
 <td> <pre><code>sc.parallelize(Seq(1, 2, 3))
 .map( num =&gt; num * 10 )
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/57c2f77284bfa8f99ade091fdd7e9f83.png" rel="nofollow"><img alt="57c2f77284bfa8f99ade091fdd7e9f83" src="https://img-blog.csdnimg.cn/img_convert/56242e63e25bf9b4cd311cb606d7f755.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/c59d44296918b864a975ebbeb60d4c04.png" rel="nofollow"><img alt="c59d44296918b864a975ebbeb60d4c04" src="https://img-blog.csdnimg.cn/img_convert/70abe4c905a9915ca847575ea46362fc.png" width="800"></a> 作用 
 <ul> 
 <li> 把 RDD 中的数据 一对一 的转为另一种形式 </li> 
 </ul>签名 <code>def map[U: ClassTag](f: T ⇒ U): RDD[U]</code> 参数 
 <ul> 
 <li> <code>f</code>&nbsp;→ Map 算子是&nbsp;<code>原RDD → 新RDD</code>&nbsp;的过程, 传入函数的参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据 </li> 
 </ul>注意点 
 <ul> 
 <li> Map 是一对一, 如果函数是&nbsp;<code>String → Array[String]</code>&nbsp;则新的 RDD 中每条数据就是一个数组 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>flatMap(T ⇒ List[U])</code> </th> 
 <td> <pre><code>sc.parallelize(Seq("Hello lily", "Hello lucy", "Hello tim"))
 .flatMap( line =&gt; line.split(" ") )
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ec39594f30ca4d59e2ef5cdc60387866.png" rel="nofollow"><img alt="ec39594f30ca4d59e2ef5cdc60387866" src="https://img-blog.csdnimg.cn/img_convert/3565df2f3226bebe7aa28e936eb57917.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/f6c4feba14bb71372aa0cb678067c6a8.png" rel="nofollow"><img alt="f6c4feba14bb71372aa0cb678067c6a8" src="https://img-blog.csdnimg.cn/img_convert/a5c14a5995828f9afbd7fcb0649639e0.png" width="800"></a> 作用 
 <ul> 
 <li> FlatMap 算子和 Map 算子类似, 但是 FlatMap 是一对多 </li> 
 </ul>调用 <code>def flatMap[U: ClassTag](f: T ⇒ List[U]): RDD[U]</code> 参数 
 <ul> 
 <li> <code>f</code>&nbsp;→ 参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据, 需要注意的是返回值是一个集合, 集合中的数据会被展平后再放入新的 RDD </li> 
 </ul>注意点 
 <ul> 
 <li> flatMap 其实是两个操作, 是&nbsp;<code>map + flatten</code>, 也就是先转换, 后把转换而来的 List 展开 </li> 
 <li> Spark 中并没有直接展平 RDD 中数组的算子, 可以使用&nbsp;<code>flatMap</code>&nbsp;做这件事 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>filter(T ⇒ Boolean)</code> </th> 
 <td> <pre><code>sc.parallelize(Seq(1, 2, 3))
 .filter( value =&gt; value &gt;= 3 )
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/25a7aef5e2b8a39145d503f4652cc945.png" rel="nofollow"><img alt="25a7aef5e2b8a39145d503f4652cc945" src="https://img-blog.csdnimg.cn/img_convert/f851669e39bb33a26cd7b902c2cce8a9.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/05cdb79abd41a7b5baa41a4c62870d73.png" rel="nofollow"><img alt="05cdb79abd41a7b5baa41a4c62870d73" src="https://img-blog.csdnimg.cn/img_convert/9320375ce58b33813e0eadfda2d3ab54.png" width="800"></a> 作用 
 <ul> 
 <li> <code>Filter</code>&nbsp;算子的主要作用是过滤掉不需要的内容 </li> 
 <li> scala&gt; sc.parallelize(Seq("Hello lily", "Hello lucy", "Hello tim")).flatMap(_.split(" ")).filter(x=&gt;x.length &gt;4).collect res8: Array[String] = Array(Hello, Hello, Hello) &nbsp; </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>mapPartitions(List[T] ⇒ List[U])</code> </th> 
 <td> RDD[T] ⇒ RDD[U]&nbsp;和 map 类似, 但是针对整个分区的数据转换 </td> 
 </tr> 
 <tr> 
 <th> <code>mapPartitionsWithIndex</code> </th> 
 <td> 和 mapPartitions 类似, 只是在函数中增加了分区的 Index </td> 
 </tr> 
 <tr> 
 <th> <code>mapValues</code> </th> 
 <td> <pre><code>sc.parallelize(Seq(("a", 1), ("b", 2), ("c", 3)))
 .mapValues( value =&gt; value * 10 )
 .collect()</code></pre> <img alt="7a8b280a054fdab8e8d14549f67b85f9" src="https://img-blog.csdnimg.cn/img_convert/7534c05485e52ccdc5f15116b6defe35.png"> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/5551847febe453b134f3a4009df01bec.png" rel="nofollow"><img alt="5551847febe453b134f3a4009df01bec" src="https://img-blog.csdnimg.cn/img_convert/ee4de5f7961199638e13b881790b67e2.png" width="800"></a> 作用 
 <ul> 
 <li> MapValues 只能作用于 Key-Value 型数据, 和 Map 类似, 也是使用函数按照转换数据, 不同点是 MapValues 只转换 Key-Value 中的 Value </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>sample(withReplacement, fraction, seed)</code> </th> 
 <td> <pre><code>sc.parallelize(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
 .sample(withReplacement = true, 0.6, 2)
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/03139edb0211652195dccea955f3a9b3.png" rel="nofollow"><img alt="03139edb0211652195dccea955f3a9b3" src="https://img-blog.csdnimg.cn/img_convert/cec94fd0ef059e0704d3c5751e309c80.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ccd1ae121f6f6852158c044441437f04.png" rel="nofollow"><img alt="ccd1ae121f6f6852158c044441437f04" src="https://img-blog.csdnimg.cn/img_convert/a2a94d578a828adf7cb82c29ac7990a1.png" width="800"></a> 作用 
 <ul> 
 <li> Sample 算子可以从一个数据集中抽样出来一部分, 常用作于减小数据集以保证运行速度, 并且尽可能少规律的损失 </li> 
 </ul>参数 
 <ul> 
 <li> Sample 接受第一个参数为&nbsp;<code>withReplacement</code>, 意为是否取样以后是否还放回原数据集供下次使用, 简单的说, 如果这个参数的值为 true, 则抽样出来的数据集中可能会有重复 </li> 
 <li> Sample 接受第二个参数为&nbsp;<code>fraction</code>, 意为抽样的比例 </li> 
 <li> Sample 接受第三个参数为&nbsp;<code>seed</code>, 随机数种子, 用于 Sample 内部随机生成下标, 一般不指定, 使用默认值 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>union(other)</code> </th> 
 <td> <pre><code>val rdd1 = sc.parallelize(Seq(1, 2, 3))
val rdd2 = sc.parallelize(Seq(4, 5, 6))
rdd1.union(rdd2)
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/5f31c2c44aa66db3027fea4624a3c4eb.png" rel="nofollow"><img alt="5f31c2c44aa66db3027fea4624a3c4eb" src="https://img-blog.csdnimg.cn/img_convert/b986cfdd42523402a49a249374585a33.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/2a8b7d10930251ae32d6d276ab7f41f8.png" rel="nofollow"><img alt="2a8b7d10930251ae32d6d276ab7f41f8" src="https://img-blog.csdnimg.cn/img_convert/f70a10166c639c6e9ba119588fcb257e.png" width="800"></a> </td> 
 </tr> 
 <tr> 
 <th> <code>intersection(other)</code> </th> 
 <td> <pre><code>val rdd1 = sc.parallelize(Seq(1, 2, 3, 4, 5))
val rdd2 = sc.parallelize(Seq(4, 5, 6, 7, 8))
rdd1.intersection(rdd2)
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a4475b1193be01efc305ef3c39f4b1e8.png" rel="nofollow"><img alt="a4475b1193be01efc305ef3c39f4b1e8" src="https://img-blog.csdnimg.cn/img_convert/c6f38b14a3378b292f7c7bbd070e5890.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/76a9873eae8de8a9ed5223921da7c245.png" rel="nofollow"><img alt="76a9873eae8de8a9ed5223921da7c245" src="https://img-blog.csdnimg.cn/img_convert/d53a4c0b3eaca117fefa9d30f9f2e217.png" width="800"></a> 作用 
 <ul> 
 <li> Intersection 算子是一个集合操作, 用于求得 左侧集合 和 右侧集合 的交集, 换句话说, 就是左侧集合和右侧集合都有的元素, 并生成一个新的 RDD </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>subtract(other, numPartitions)</code> </th> 
 <td> (RDD[T], RDD[T]) ⇒ RDD[T]&nbsp;差集, 可以设置分区数 </td> 
 </tr> 
 <tr> 
 <th> <code>distinct(numPartitions)</code> </th> 
 <td> <pre><code>sc.parallelize(Seq(1, 1, 2, 2, 3))
 .distinct()
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a8cd033d9ce502337ba746d05ca94ae1.png" rel="nofollow"><img alt="a8cd033d9ce502337ba746d05ca94ae1" src="https://img-blog.csdnimg.cn/img_convert/9b9d17560c703e1aef9447792b5e3e0b.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/2bfefe5f5cab497d5aded3b7537a58ba.png" rel="nofollow"><img alt="2bfefe5f5cab497d5aded3b7537a58ba" src="https://img-blog.csdnimg.cn/img_convert/77de9dfc99fa21dd74ad1f09ef5d9e6f.png" width="800"></a> 作用 
 <ul> 
 <li> Distinct 算子用于去重 </li> 
 </ul>注意点 
 <ul> 
 <li> Distinct 是一个需要 Shuffled 的操作 </li> 
 <li> 本质上 Distinct 就是一个 reductByKey, 把重复的合并为一个 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>reduceByKey((V, V) ⇒ V, numPartition)</code> </th> 
 <td> <pre><code>sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
 .reduceByKey( (curr, agg) =&gt; curr + agg )
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a9b444d144d6996c83b33f6a48806a1a.png" rel="nofollow"><img alt="a9b444d144d6996c83b33f6a48806a1a" src="https://img-blog.csdnimg.cn/img_convert/623293a3d62958ac6bfb4c4c0d213a7c.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/07678e1b4d6ba1dfaf2f5df89489def4.png" rel="nofollow"><img alt="07678e1b4d6ba1dfaf2f5df89489def4" src="https://img-blog.csdnimg.cn/img_convert/ab17f3d5d34fa78eacdc7df797c99586.png" width="800"></a> 作用 
 <ul> 
 <li> 首先按照 Key 分组生成一个 Tuple, 然后针对每个组执行&nbsp;<code>reduce</code>&nbsp;算子 </li> 
 </ul>调用 <code>def reduceByKey(func: (V, V) ⇒ V): RDD[(K, V)]</code> 参数 
 <ul> 
 <li> func → 执行数据处理的函数, 传入两个参数, 一个是当前值, 一个是局部汇总, 这个函数需要有一个输出, 输出就是这个 Key 的汇总结果 </li> 
 </ul>注意点 
 <ul> 
 <li> ReduceByKey 只能作用于 Key-Value 型数据, Key-Value 型数据在当前语境中特指 Tuple2 </li> 
 <li> ReduceByKey 是一个需要 Shuffled 的操作 </li> 
 <li> 和其它的 Shuffled 相比, ReduceByKey是高效的, 因为类似 MapReduce 的, 在 Map 端有一个 Cominer, 这样 I/O 的数据便会减少 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>groupByKey()</code> </th> 
 <td> <pre><code>sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
 .groupByKey()
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/466c1ad2b738c4f0d27f2557ecedaf5b.png" rel="nofollow"><img alt="466c1ad2b738c4f0d27f2557ecedaf5b" src="https://img-blog.csdnimg.cn/img_convert/48b1261c8f59730be2e1398e616439a4.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/27de81df110abb6709bf1c5ffad184ab.png" rel="nofollow"><img alt="27de81df110abb6709bf1c5ffad184ab" src="https://img-blog.csdnimg.cn/img_convert/dc3a059c6863b6440464dcdb7c3d2206.png" width="800"></a> &nbsp; scala&gt; sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1))).groupByKey().collect res33: Array[(String, Iterable[Int])] = Array((a,CompactBuffer(1, 1)), (b,CompactBuffer(1))) scala&gt; sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1))).groupByKey().mapValues(_.reduce(_+_)).collect res34: Array[(String, Int)] = Array((a,2), (b,1)) 作用 
 <ul> 
 <li> GroupByKey 算子的主要作用是按照 Key 分组, 和 ReduceByKey 有点类似, 但是 GroupByKey 并不求聚合, 只是列举 Key 对应的所有 Value </li> 
 </ul>注意点 
 <ul> 
 <li> GroupByKey 是一个 Shuffled </li> 
 <li> GroupByKey 和 ReduceByKey 不同, 因为需要列举 Key 对应的所有数据, 所以无法在 Map 端做 Combine, 所以 GroupByKey 的性能并没有 ReduceByKey 好 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>combineByKey()</code> </th> 
 <td> <pre><code>val rdd = sc.parallelize(Seq(
 ("zhangsan", 99.0),
 ("zhangsan", 96.0),
 ("lisi", 97.0),
 ("lisi", 98.0),
 ("zhangsan", 97.0))
)

val combineRdd = rdd.combineByKey(
  score =&gt; (score, 1),
  (scoreCount: (Double, Int),newScore) =&gt; (scoreCount._1 + newScore, scoreCount._2 + 1),
  (scoreCount1: (Double, Int), scoreCount2: (Double, Int)) =&gt;
    (scoreCount1._1 + scoreCount2._1, scoreCount1._2 + scoreCount2._2)
)

val meanRdd = combineRdd.map(score =&gt; (score._1, score._2._1 / score._2._2))

meanRdd.collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/741d814a50e4c01686f394df079458bf.png" rel="nofollow"><img alt="741d814a50e4c01686f394df079458bf" src="https://img-blog.csdnimg.cn/img_convert/dd9a5a145b90ab24a5eacedac65eb217.png" width="800"></a> &nbsp; &nbsp; 作用 
 <ul> 
 <li> 对数据集按照 Key 进行聚合 </li> 
 </ul>调用 
 <ul> 
 <li> <code>combineByKey(createCombiner, mergeValue, mergeCombiners, [partitioner], [mapSideCombiner], [serializer])</code> </li> 
 </ul>参数 
 <ul> 
 <li> <code>createCombiner</code>&nbsp;将 Value 进行初步转换 </li> 
 <li> <code>mergeValue</code>&nbsp;在每个分区把上一步转换的结果聚合 </li> 
 <li> <code>mergeCombiners</code>&nbsp;在所有分区上把每个分区的聚合结果聚合 </li> 
 <li> <code>partitioner</code>&nbsp;可选, 分区函数 </li> 
 <li> <code>mapSideCombiner</code>&nbsp;可选, 是否在 Map 端 Combine </li> 
 <li> <code>serializer</code>&nbsp;序列化器 </li> 
 </ul>注意点 
 <ul> 
 <li> <code>combineByKey</code>&nbsp;的要点就是三个函数的意义要理解 </li> 
 <li> <code>groupByKey</code>,&nbsp;<code>reduceByKey</code>&nbsp;的底层都是&nbsp;<code>combineByKey</code> </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>aggregateByKey()</code> </th> 
 <td> <pre><code>val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
val result = rdd.aggregateByKey(0.8)(
 seqOp = (zero, price) =&gt; price * zero,
 combOp = (curr, agg) =&gt; curr + agg
).collect()
println(result)</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ee33b17dbc78705dbbd76d76ab4a9072.png" rel="nofollow"><img alt="ee33b17dbc78705dbbd76d76ab4a9072" src="https://img-blog.csdnimg.cn/img_convert/c8e17ad284d9289c1ef9152232a57eb9.png" width="800"></a> 作用 
 <ul> 
 <li> 聚合所有 Key 相同的 Value, 换句话说, 按照 Key 聚合 Value </li> 
 </ul>调用 
 <ul> 
 <li> <code>rdd.aggregateByKey(zeroValue)(seqOp, combOp)</code> </li> 
 </ul>参数 
 <ul> 
 <li> <code>zeroValue</code>&nbsp;初始值 </li> 
 <li> <code>seqOp</code>&nbsp;转换每一个值的函数 </li> 
 <li> <code>comboOp</code>&nbsp;将转换过的值聚合的函数 </li> 
 </ul>注意点&nbsp;* 为什么需要两个函数?&nbsp;aggregateByKey 运行将一个&nbsp;<code>RDD[(K, V)]</code>&nbsp;聚合为&nbsp;<code>RDD[(K, U)]</code>, 如果要做到这件事的话, 就需要先对数据做一次转换, 将每条数据从&nbsp;<code>V</code>&nbsp;转为&nbsp;<code>U</code>,&nbsp;<code>seqOp</code>&nbsp;就是干这件事的 ** 当&nbsp;<code>seqOp</code>&nbsp;的事情结束以后,&nbsp;<code>comboOp</code>&nbsp;把其结果聚合 
 <ul> 
 <li> 和 reduceByKey 的区别:: 
 <ul> 
 <li> aggregateByKey 最终聚合结果的类型和传入的初始值类型保持一致 </li> 
 <li> reduceByKey 在集合中选取第一个值作为初始值, 并且聚合过的数据类型不能改变 </li> 
 </ul></li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>foldByKey(zeroValue)((V, V) ⇒ V)</code> </th> 
 <td> <pre><code>sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
 .foldByKey(zeroValue = 10)( (curr, agg) =&gt; curr + agg )
 .collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/c00063a109a0f9e0b1c2b385c5e1cc47.png" rel="nofollow"><img alt="c00063a109a0f9e0b1c2b385c5e1cc47" src="https://img-blog.csdnimg.cn/img_convert/49e30ce3663b1827ca0a53175ab31f27.png" width="800"></a> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a406ff8395bb092e719007661b34d385.png" rel="nofollow"><img alt="a406ff8395bb092e719007661b34d385" src="https://img-blog.csdnimg.cn/img_convert/baf6c1b5b16b5a839dd57b8ae1094cf4.png" width="800"></a> 作用 
 <ul> 
 <li> 和 ReduceByKey 是一样的, 都是按照 Key 做分组去求聚合, 但是 FoldByKey 的不同点在于可以指定初始值 </li> 
 </ul>调用 <code>foldByKey(zeroValue)(func)</code> 参数 
 <ul> 
 <li> <code>zeroValue</code>&nbsp;初始值 </li> 
 <li> <code>func</code>&nbsp;seqOp 和 combOp 相同, 都是这个参数 </li> 
 </ul>注意点 
 <ul> 
 <li> FoldByKey 是 AggregateByKey 的简化版本, seqOp 和 combOp 是同一个函数 </li> 
 <li> FoldByKey 指定的初始值作用于每一个 Value </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>join(other, numPartitions)</code> </th> 
 <td> <pre><code>val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("b", 1)))
val rdd2 = sc.parallelize(Seq(("a", 10), ("a", 11), ("a", 12)))

rdd1.join(rdd2).collect()</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/bb3eda1410d3b0f6e1bff6d5e6a45879.png" rel="nofollow"><img alt="bb3eda1410d3b0f6e1bff6d5e6a45879" src="https://img-blog.csdnimg.cn/img_convert/d5e07e70f25c712065819aed123f32ff.png" width="800"></a> 作用 
 <ul> 
 <li> 将两个 RDD 按照相同的 Key 进行连接 </li> 
 </ul>调用 <code>join(other, [partitioner or numPartitions])</code> 参数 
 <ul> 
 <li> <code>other</code>&nbsp;其它 RDD </li> 
 <li> <code>partitioner or numPartitions</code>&nbsp;可选, 可以通过传递分区函数或者分区数量来改变分区 </li> 
 </ul>注意点 
 <ul> 
 <li> Join 有点类似于 SQL 中的内连接, 只会再结果中包含能够连接到的 Key </li> 
 <li> Join 的结果是一个笛卡尔积形式, 例如&nbsp;<code>"a", 1), ("a", 2</code>&nbsp;和&nbsp;<code>"a", 10), ("a", 11</code>&nbsp;的 Join 结果集是&nbsp;<code>"a", 1, 10), ("a", 1, 11), ("a", 2, 10), ("a", 2, 11</code> </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>cogroup(other, numPartitions)</code> </th> 
 <td> <pre><code>val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("a", 5), ("b", 2), ("b", 6), ("c", 3), ("d", 2)))
val rdd2 = sc.parallelize(Seq(("a", 10), ("b", 1), ("d", 3)))
val rdd3 = sc.parallelize(Seq(("b", 10), ("a", 1)))

val result1 = rdd1.cogroup(rdd2).collect()
val result2 = rdd1.cogroup(rdd2, rdd3).collect()

/*
执行结果:
Array(
  (d,(CompactBuffer(2),CompactBuffer(3))),
  (a,(CompactBuffer(1, 2, 5),CompactBuffer(10))),
  (b,(CompactBuffer(2, 6),CompactBuffer(1))),
  (c,(CompactBuffer(3),CompactBuffer()))
)
 */
println(result1)

/*
执行结果:
Array(
 (d,(CompactBuffer(2),CompactBuffer(3),CompactBuffer())),
 (a,(CompactBuffer(1, 2, 5),CompactBuffer(10),CompactBuffer(1))),
 (b,(CompactBuffer(2, 6),CompactBuffer(1),Co...
 */
println(result2)</code></pre> <a href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/42262ffe7f3ff35013fbe534d78e3518.png" rel="nofollow"><img alt="42262ffe7f3ff35013fbe534d78e3518" src="https://img-blog.csdnimg.cn/img_convert/6974b5094815cc83bdacb9030c77ed86.png" width="800"></a> 作用 
 <ul> 
 <li> 多个 RDD 协同分组, 将多个 RDD 中 Key 相同的 Value 分组 </li> 
 </ul>调用 
 <ul> 
 <li> <code>cogroup(rdd1, rdd2, rdd3, [partitioner or numPartitions])</code> </li> 
 </ul>参数 
 <ul> 
 <li> <code>rdd…</code>&nbsp;最多可以传三个 RDD 进去, 加上调用者, 可以为四个 RDD 协同分组 </li> 
 <li> <code>partitioner or numPartitions</code>&nbsp;可选, 可以通过传递分区函数或者分区数来改变分区 </li> 
 </ul>注意点 
 <ul> 
 <li> 对 RDD1, RDD2, RDD3 进行 cogroup, 结果中就一定会有三个 List, 如果没有 Value 则是空 List, 这一点类似于 SQL 的全连接, 返回所有结果, 即使没有关联上 </li> 
 <li> CoGroup 是一个需要 Shuffled 的操作 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>cartesian(other)</code> </th> 
 <td> (RDD[T], RDD[U]) ⇒ RDD[(T, U)]&nbsp;生成两个 RDD 的笛卡尔积 </td> 
 </tr> 
 <tr> 
 <th> <code>sortBy(ascending, numPartitions)</code> </th> 
 <td> <pre><code>val rdd1 = sc.parallelize(Seq(("a", 3), ("b", 2), ("c", 1)))
val sortByResult = rdd1.sortBy( item =&gt; item._2 ).collect()
val sortByKeyResult = rdd1.sortByKey().collect()

println(sortByResult)
println(sortByKeyResult)</code></pre> 作用 
 <ul> 
 <li> 排序相关相关的算子有两个, 一个是&nbsp;<code>sortBy</code>, 另外一个是&nbsp;<code>sortByKey</code> </li> 
 </ul>调用 <code>sortBy(func, ascending, numPartitions)</code> 参数 
 <ul> 
 <li> <code>func</code>&nbsp;通过这个函数返回要排序的字段 </li> 
 <li> <code>ascending</code>&nbsp;是否升序 </li> 
 <li> <code>numPartitions</code>&nbsp;分区数 </li> 
 </ul>注意点 
 <ul> 
 <li> 普通的 RDD 没有&nbsp;<code>sortByKey</code>, 只有 Key-Value 的 RDD 才有 </li> 
 <li> <code>sortBy</code>&nbsp;可以指定按照哪个字段来排序,&nbsp;<code>sortByKey</code>&nbsp;直接按照 Key 来排序 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>partitionBy(partitioner)</code> </th> 
 <td> 使用用传入的 partitioner 重新分区, 如果和当前分区函数相同, 则忽略操作 </td> 
 </tr> 
 <tr> 
 <th> <code>coalesce(numPartitions)</code> </th> 
 <td> 减少分区数 <pre><code>val rdd = sc.parallelize(Seq(("a", 3), ("b", 2), ("c", 1)))
val oldNum = rdd.partitions.length

val coalesceRdd = rdd.coalesce(4, shuffle = true)
val coalesceNum = coalesceRdd.partitions.length

val repartitionRdd = rdd.repartition(4)
val repartitionNum = repartitionRdd.partitions.length

print(oldNum, coalesceNum, repartitionNum)</code></pre> 作用 
 <ul> 
 <li> 一般涉及到分区操作的算子常见的有两个,&nbsp;<code>repartitioin</code>&nbsp;和&nbsp;<code>coalesce</code>, 两个算子都可以调大或者调小分区数量 </li> 
 </ul>调用 
 <ul> 
 <li> <code>repartitioin(numPartitions)</code> </li> 
 <li> <code>coalesce(numPartitions, shuffle)</code> </li> 
 </ul>参数 
 <ul> 
 <li> <code>numPartitions</code>&nbsp;新的分区数 </li> 
 <li> <code>shuffle</code>&nbsp;是否 shuffle, 如果新的分区数量比原分区数大, 必须 Shuffled, 否则重分区无效 </li> 
 </ul>注意点 
 <ul> 
 <li> <code>repartition</code>&nbsp;和&nbsp;<code>coalesce</code>&nbsp;的不同就在于&nbsp;<code>coalesce</code>&nbsp;可以控制是否 Shuffle </li> 
 <li> <code>repartition</code>&nbsp;是一个 Shuffled 操作 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>repartition(numPartitions)</code> </th> 
 <td> 重新分区 </td> 
 </tr> 
 <tr> 
 <th> <code>repartitionAndSortWithinPartitions</code> </th> 
 <td> 重新分区的同时升序排序, 在&nbsp;<code>partitioner</code>&nbsp;中排序, 比先重分区再排序要效率高, 建议使用在需要分区后再排序的场景使用 </td> 
 </tr> 
 </tbody> 
</table>

### 2. Action 算子 ###

<table> 
 <thead> 
 <tr> 
 <th>Action function</th> 
 <th>解释</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <th> <code>reduce( (T, T) ⇒ U )</code> </th> 
 <td> <pre><code>val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
val result = rdd.reduce((curr, agg) =&gt; ("总价", curr._2 + agg._2))
println(result)

sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))&nbsp; .reduceByKey( (curr, agg) =&gt; curr + agg )&nbsp; .collect()&nbsp;</code></pre> 作用 
 <ul> 
 <li> 对整个结果集规约, 最终生成一条数据, 是整个数据集的汇总 </li> 
 </ul>调用 
 <ul> 
 <li> <code>reduce( (currValue[T], agg[T]) ⇒ T )</code> </li> 
 </ul>注意点 
 <ul> 
 <li> reduce 和 reduceByKey 是完全不同的, reduce 是一个 action, 并不是 Shuffled 操作 </li> 
 <li> 本质上 reduce 就是现在每个 partition 上求值, 最终把每个 partition 的结果再汇总 </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>collect()</code> </th> 
 <td> 以数组的形式返回数据集中所有元素 </td> 
 </tr> 
 <tr> 
 <th> <code>count()</code> </th> 
 <td> 返回元素个数 </td> 
 </tr> 
 <tr> 
 <th> <code>first()</code> </th> 
 <td> 返回第一个元素 </td> 
 </tr> 
 <tr> 
 <th> <code>take( N )</code> </th> 
 <td> 返回前 N 个元素 </td> 
 </tr> 
 <tr> 
 <th> <code>takeSample(withReplacement, fract)</code> </th> 
 <td> 类似于 sample, 区别在这是一个Action, 直接返回结果 </td> 
 </tr> 
 <tr> 
 <th> <code>fold(zeroValue)( (T, T) ⇒ U )</code> </th> 
 <td> 指定初始值和计算函数, 折叠聚合整个数据集 </td> 
 </tr> 
 <tr> 
 <th> <code>saveAsTextFile(path)</code> </th> 
 <td> 将结果存入 path 对应的文件中 </td> 
 </tr> 
 <tr> 
 <th> <code>saveAsSequenceFile(path)</code> </th> 
 <td> 将结果存入 path 对应的 Sequence 文件中 </td> 
 </tr> 
 <tr> 
 <th> <code>countByKey()</code> </th> 
 <td> <pre><code>val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
val result = rdd.countByKey()
println(result)</code></pre> 作用 
 <ul> 
 <li> 求得整个数据集中 Key 以及对应 Key 出现的次数 </li> 
 </ul>注意点 
 <ul> 
 <li> 返回结果为&nbsp;<code>Map(key → count)</code> </li> 
 <li> 常在解决数据倾斜问题时使用, 查看倾斜的 Key </li> 
 </ul></td> 
 </tr> 
 <tr> 
 <th> <code>foreach( T ⇒ … )</code> </th> 
 <td> 遍历每一个元素 </td> 
 </tr> 
 </tbody> 
</table>

应用

```scala
    val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
    // 结果: Array((手机,10.0), (手机,15.0), (电脑,20.0))
    println(rdd.collect())
    // 结果: Array((手机,10.0), (手机,15.0))
    println(rdd.take(2))
    // 结果: (手机,10.0)
    println(rdd.first())
    ```

总结

RDD 的算子大部分都会生成一些专用的 RDD

*  `map`, `flatMap`, `filter` 等算子会生成 `MapPartitionsRDD`
 *  `coalesce`, `repartition` 等算子会生成 `CoalescedRDD`

常见的 RDD 有两种类型

*  转换型的 RDD, Transformation
 *  动作型的 RDD, Action

常见的 Transformation 类型的 RDD

*  map
 *  flatMap
 *  filter
 *  groupBy
 *  reduceByKey

常见的 Action 类型的 RDD

*  collect
 *  countByKey
 *  reduce