Spark系列--SparkCore(三)RDD基本操作

r囧r小猫 2022-05-20 05:06 160阅读 0赞

## 前言 ##

RDD的基本操作分为两种，一种是转换Transformation，一种是行动Action

RDD中的所有转换都是延迟加载的，也就是说，它们并不会直接计算结果。相反的，它们只是记住这些应用到基础数据集（例如一个文件）上的转换动作。只有当发生一个要求返回结果给Driver的动作时，这些转换才会真正运行。这种设计让Spark更加有效率地运行。

各算子的详细代码示例，参考如下：  
[Spark算子使用示例][Spark]

## 一、常用的Transformation ##

**1）map(func)**

> **返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成**

**2）filter(func)**

> **返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成**

**3）flatMap(func)**

> **类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）**

**4）mapPartitions(func)**

> **类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator\[T\] => Iterator\[U\]。假设有N个元素，有M个分区，那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区**

**5）mapPartitionsWithIndex(func)**

> **类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator\[T\]) => Iterator\[U\]**

**6）sample(withReplacement, fraction, seed)**

> **以指定的随机种子随机抽样出数量为fraction的数据，withReplacement表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样，seed用于指定随机数生成器种子。例子从RDD中随机且有放回的抽出50%的数据，随机种子值为3（即可能以1 2 3的其中一个起始值）**

**7）takeSample**

> **和Sample的区别是：takeSample返回的是最终的结果集合。**

**8）intersection(otherDataset)**

> **对源RDD和参数RDD求交集后返回一个新的RDD**

**9）distinct(\[numTasks\]))**

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

**10）partitionBy**

> **对RDD进行分区操作，如果原有的partionRDD和现有的partionRDD是一致的话就不进行分区， 否则会生成ShuffleRDD。**

**11）reduceByKey(func, \[numTasks\])**

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

**12）groupByKey**

> **groupByKey也是对每个key进行操作，但只生成一个sequence。**

**13）combineByKey\[C\]( createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C)**

> 对相同K，把V合并成一个集合.  
> **createCombiner**: combineByKey() 会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，要么就 和之前的某个元素的键相同。如果这是一个新的元素,combineByKey() 会使用一个叫作 createCombiner() 的函数来创建  
> 那个键对应的累加器的初始值  
> **mergeValue**: 如果这是一个在处理当前分区之前已经遇到的键， 它会使用 mergeValue() 方法将该键的累加器对应的当前值与这个新的值进行合并  
> **mergeCombiners**: 由于每个分区都是独立处理的， 因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器， 就需要使用用户提供的 mergeCombiners() 方法将各个分区的结果进行合并。

**14）aggregateByKey(zeroValue:U,\[partitioner: Partitioner\]) (seqOp: (U, V) => U,combOp: (U, U) => U)**

> 在kv对的RDD中，，按key将value进行分组合并，合并时，将每个value和初始值作为seq函数的参数，进行计算，返回的结果作为一个新的kv对，然后再将结果按照key进行合并，最后将每个分组的value传递给combine函数进行计算（先将前两个value进行计算，将返回结果和下一个value传给combine函数，以此类推），将key与计算结果作为一个新的kv对输出。  
> seqOp函数用于在每一个分区中用初始值逐步迭代value，combOp函数用于合并每个分区中的结果

**15）foldByKey(zeroValue: V)(func: (V, V) => V): RDD\[(K, V)\]**

> **aggregateByKey的简化操作，seqop和combop相同**

**16）sortByKey(\[ascending\], \[numTasks\])**

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

**17）sortBy(func,\[ascending\], \[numTasks\])**

> **与sortByKey类似，但是更灵活,可以用func先对数据进行处理，按照处理后的数据比较结果排序。**

**18）join(otherDataset, \[numTasks\])**

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

**19）cogroup(otherDataset, \[numTasks\])**

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

**20）cartesian(otherDataset)**

> **笛卡尔积**

**21）pipe(command, \[envVars\])**  
**对于每个分区，都执行一个perl或者shell脚本，返回输出的RDD**

**22）coalesce(numPartitions)**

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

**23）repartition(numPartitions)**

> **根据分区数，从新通过网络随机洗牌所有数据。**

**24）repartitionAndSortWithinPartitions(partitioner)**

> **repartitionAndSortWithinPartitions函数是repartition函数的变种，与repartition函数不同的是，repartitionAndSortWithinPartitions在给定的partitioner内部进行排序，性能比repartition要高。**

**25）glom**

> **将每一个分区形成一个数组，形成新的RDD类型时RDD\[Array\[T\]\]**

**26）mapValues**

> **针对于(K,V)形式的类型只对V进行操作**

**27）subtract**

> **计算差的一种函数去除两个RDD中相同的元素，不同的RDD将保留下来**

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

## 二、常用的Action ##

**1）reduce(func)**

> **通过func函数聚集RDD中的所有元素，这个功能必须是可交换且可并联的**

**2）collect()**

> **在驱动程序中，以数组的形式返回数据集的所有元素**