Spark中的转换算子transformation和动作算子action

柔情只为你懂 2021-12-22 13:09 350阅读 0赞

主要记录了Spark中转换算子transformation和动作算子action具体含义和使用方法。

### 转换算子-transformation ###

*  map(func)  
    返回一个新的分布式数据集，其中每个元素都是由源RDD中一个元素经func转换得到的。
 *  filter(func)  
    返回一个新的数据集，其中包含的元素来自源RDD中元素经func过滤后（func返回true时才选中）的结果。
 *  flatMap(func)  
    类似于map，但每个输入元素可以映射到0到n个输出元素（所以要求func必须返回一个Seq而不是单个元素）。
 *  mapPartitions(func)  
    类似于map，但基于每个RDD分区（或者数据block）独立运行，所以如果RDD包含元素类型为T，则 func 必须是 Iterator<T> => Iterator<U> 的映射函数。
 *  mapPartitionsWithIndex(func)  
    类似于 mapPartitions，只是func 多了一个整型的分区索引值，因此如果RDD包含元素类型为T，则 func 必须是 Iterator => Iterator<U> 的映射函数。
 *  sample(withReplacement, fraction, seed)  
    采样部分（比例取决于 fraction ）数据，同时可以指定是否使用回置采样（withReplacement），以及随机数种子(seed)
 *  union(otherDataset)  
    返回源数据集和参数数据集（otherDataset）的并集。
 *  intersection(otherDataset)  
    返回源数据集和参数数据集（otherDataset）的交集。
 *  distinct(\[numTasks\]))  
    返回对源数据集做元素去重后的新数据集。
 *  groupByKey(\[numTasks\])  
    只对包含键值对的RDD有效，如源RDD包含 (K, V) 对，则该算子返回一个新的数据集包含 (K, Iterable) 对。注意：如果你需要按key分组聚合的话（如sum或average），推荐使用 reduceByKey或者 aggregateByKey 以获得更好的性能。注意：默认情况下，输出计算的并行度取决于源RDD的分区个数。当然，你也可以通过设置可选参数 numTasks 来指定并行任务的个数。
 *  reduceByKey(func, \[numTasks\])  
    如果源RDD包含元素类型 (K, V) 对，则该算子也返回包含(K, V) 对的RDD，只不过每个key对应的value是经过func聚合后的结果，而func本身是一个 (V, V) => V 的映射函数。另外，和 groupByKey 类似，可以通过可选参数 numTasks 指定reduce任务的个数。
 *  aggregateByKey(zeroValue)(seqOp, combOp, \[numTasks\])  
    如果源RDD包含 (K, V) 对，则返回新RDD包含 (K, U) 对，其中每个key对应的value都是由 combOp 函数 和 一个“0”值zeroValue 聚合得到。允许聚合后value类型和输入value类型不同，避免了不必要的开销。和 groupByKey 类似，可以通过可选参数 numTasks 指定reduce任务的个数。
 *  sortByKey(\[ascending\], \[numTasks\])  
    如果源RDD包含元素类型 (K, V) 对，其中K可排序，则返回新的RDD包含 (K, V) 对，并按照 K 排序（升序还是降序取决于 ascending 参数）。
 *  join(otherDataset, \[numTasks\])  
    如果源RDD包含元素类型 (K, V) 且参数RDD（otherDataset）包含元素类型(K, W)，则返回的新RDD中将包含内关联后key对应的 (K, (V, W)) 对。外关联(Outer joins)操作请参考 leftOuterJoin、rightOuterJoin 以及 fullOuterJoin 算子。
 *  cogroup(otherDataset, \[numTasks\])  
    如果源RDD包含元素类型 (K, V) 且参数RDD（otherDataset）包含元素类型(K, W)，则返回的新RDD中包含 (K, (Iterable, Iterable))。该算子还有个别名：groupWith。
 *  cartesian(otherDataset)  
    如果源RDD包含元素类型 T 且参数RDD（otherDataset）包含元素类型 U，则返回的新RDD包含前二者的笛卡尔积，其元素类型为 (T, U) 对。
 *  pipe(command, \[envVars\])  
    以shell命令行管道处理RDD的每个分区，如：Perl 或者 bash 脚本。RDD中每个元素都将依次写入进程的标准输入（stdin），然后按行输出到标准输出（stdout），每一行输出字符串即成为一个新的RDD元素。
 *  coalesce(numPartitions)  
    将RDD的分区数减少到numPartitions。当以后大数据集被过滤成小数据集后，减少分区数，可以提升效率。
 *  repartition(numPartitions)  
    将RDD数据重新混洗（reshuffle）并随机分布到新的分区中，使数据分布更均衡，新的分区个数取决于numPartitions。该算子总是需要通过网络混洗所有数据。
 *  repartitionAndSortWithinPartitions(partitioner)  
    根据partitioner（spark自带有HashPartitioner和RangePartitioner等）重新分区RDD，并且在每个结果分区中按key做排序。这是一个组合算子，功能上等价于先 repartition 再在每个分区内排序，但这个算子内部做了优化（将排序过程下推到混洗同时进行），因此性能更好。

### 动作算子-action ###

*  reduce(func)  
    将RDD中元素按func进行聚合（func是一个 (T,T) => T 的映射函数，其中T为源RDD元素类型，并且func需要满足 交换律 和 结合律 以便支持并行计算）
 *  collect()  
    将数据集中所有元素以数组形式返回驱动器（driver）程序。通常用于，在RDD进行了filter或其他过滤操作后，将一个足够小的数据子集返回到驱动器内存中。
 *  count()  
    返回数据集中元素个数
 *  first()  
    返回数据集中首个元素（类似于 take(1) ）
 *  take(n)  
    返回数据集中前 n 个元素
 *  takeSample(withReplacement,num, \[seed\])  
    返回数据集的随机采样子集，最多包含 num 个元素，withReplacement 表示是否使用回置采样，最后一个参数为可选参数seed，随机数生成器的种子。
 *  takeOrdered(n, \[ordering\])  
    按元素排序（可以通过 ordering 自定义排序规则）后，返回前 n 个元素
 *  saveAsTextFile(path)  
    将数据集中元素保存到指定目录下的文本文件中（或者多个文本文件），支持本地文件系统、HDFS 或者其他任何Hadoop支持的文件系统。保存过程中，Spark会调用每个元素的toString方法，并将结果保存成文件中的一行。
 *  saveAsSequenceFile(path)(Java and Scala)  
    将数据集中元素保存到指定目录下的Hadoop Sequence文件中，支持本地文件系统、HDFS 或者其他任何Hadoop支持的文件系统。适用于实现了Writable接口的键值对RDD。在Scala中，同样也适用于能够被隐式转换为Writable的类型（Spark实现了所有基本类型的隐式转换，如：Int，Double，String 等）
 *  saveAsObjectFile(path)(Java and Scala)  
    将RDD元素以Java序列化的格式保存成文件，保存结果文件可以使用 SparkContext.objectFile 来读取。
 *  countByKey()  
    只适用于包含键值对(K, V)的RDD，并返回一个哈希表，包含 (K, Int) 对，表示每个key的个数。
 *  foreach(func)  
    在RDD的每个元素上运行 func 函数。通常被用于累加操作，如：更新一个累加器（Accumulator ） 或者 和外部存储系统互操作。