Spark算子略解

[TOC]

Spark算子

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

• Transformation 转换算子：不触发提交job，在内存中完成数据的转换
  • Value数据类型的Transformation算子，这种变换并不触发提交作业，针对处理的数据项是Value型的数据
  • Key-Value数据类型的Transfromation算子，这种变换并不触发提交作业，针对处理的数据项是Key-Value型的数据对。
• Action 行为算子：触发SparkContext提交job，完成shuffle操作

Transformations

​ 下表列出了 Spark 支持的一些常见转换。 参考 RDD API 文档 (Scala, [Java](https:// spark.apache.org/docs/latest/api/java/index.html?org/apache/spark/api/java/JavaRDD.html), [Python](https://spark.apache.org/docs/latest /api/python/pyspark.html#pyspark.RDD)、R) 和配对 RDD 函数文档 (Scala, [Java](https://spark.apache.org/docs/latest/ api/java/index.html?org/apache/spark/api/java/JavaPairRDD.html)) 了解详细信息。

Transformation	Meaning
map(func)	返回通过将源的每个元素通过函数 func 形成的新分布式数据集
filter(func)	返回通过选择源中 func 返回 true 的那些元素形成的新数据集。
flatMap(func)	与 map 类似，但每个输入项可以映射到 0 个或多个输出项（因此 func 应返回 Seq 而不是单个项）。
mapPartitions(func)	与 map 类似，但在 RDD 的每个分区（块）上单独运行，因此 func 在 T 类型的 RDD 上运行时必须是 Iterator => Iterator 类型。
mapPartitionsWithIndex(func)	与 mapPartitions 类似，但也为 func 提供了一个表示分区索引的整数值，因此 func 在 RDD 类型的 RDD 上运行时必须是 (Int, Iterator) => Iterator 类型 T。
sample(withReplacement, fraction, seed)	使用给定的随机数生成器种子对数据的一小部分 fraction 进行采样，无论是否有替换。
union(otherDataset)	返回一个新数据集，其中包含源数据集中元素和参数的并集。
intersection(otherDataset)	返回一个新的 RDD，它包含源数据集中元素和参数的交集。
distinct([numPartitions]))	返回包含源数据集不同元素的新数据集。
groupByKey([numPartitions])	在 (K, V) 对的数据集上调用时，返回 (K, Iterable) 对的数据集。 注意： 如果您分组是为了对每个键执行聚合（例如求和或平均值），使用 reduceByKey 或 aggregateByKey 会产生更好的性能。 注意： 默认情况下，输出中的并行级别取决于父 RDD 的分区数。 您可以传递可选的 numPartitions 参数来设置不同数量的任务。
reduceByKey(func, [numPartitions])	在 (K, V) 对的数据集上调用时，返回 (K, V) 对的数据集，其中每个键的值使用给定的 reduce 函数 func 聚合，该函数必须是 (V,V) 类型 => V. 就像在 groupByKey 中一样，reduce 任务的数量可以通过可选的第二个参数进行配置。
aggregateByKey(zeroValue)(seqOp, combOp, [numPartitions])	当在 (K, V) 对的数据集上调用时，返回 (K, U) 对的数据集，其中每个键的值使用给定的组合函数和中性“零”值聚合。 允许与输入值类型不同的聚合值类型，同时避免不必要的分配。 就像在 groupByKey 中一样，reduce 任务的数量可以通过可选的第二个参数进行配置。
sortByKey([ascending], [numPartitions])	当在 K 实现 Ordered 的 (K, V) 对数据集上调用时，返回一个 (K, V) 对数据集，按升序或降序键排序，如布尔 ascending 参数中指定的那样。
join(otherDataset, [numPartitions])	当调用类型为 (K, V) 和 (K, W) 的数据集时，返回一个 (K, (V, W)) 对的数据集，其中包含每个键的所有元素对。 通过leftOuterJoin、rightOuterJoin和fullOuterJoin支持外连接。
cogroup(otherDataset, [numPartitions])	当调用类型为 (K, V) 和 (K, W) 的数据集时，返回一个包含 (K, (Iterable, Iterable)) 元组的数据集。 此操作也称为“groupWith”。
cartesian(otherDataset)	当调用类型为 T 和 U 的数据集时，返回一个 (T, U) 对（所有元素对）的数据集。
pipe(command, [envVars])	通过 shell 命令管道 RDD 的每个分区，例如 Perl 或 bash 脚本。 RDD 元素被写入进程的标准输入，输出到标准输出的行作为字符串的 RDD 返回。
coalesce(numPartitions)	将 RDD 中的分区数减少到 numPartitions。 对过滤大型数据集后更有效地运行操作很有用。
repartition(numPartitions)	随机重组 RDD 中的数据以创建更多或更少的分区并在它们之间进行平衡。 这总是在网络上打乱所有数据。
repartitionAndSortWithinPartitions(partitioner)	根据给定的分区器对 RDD 进行重新分区，并在每个结果分区内，按键对记录进行排序。 这比调用 repartition 然后在每个分区内排序更有效，因为它可以将排序推到 shuffle 机器中。

Actions

下表列出了 Spark 支持的一些常见操作。 参考 RDD API 文档 (Scala, Java, Python, R) ，和配对 RDD 函数文档 (Scala, Java) 了解详情。

Action	Meaning
reduce(func)	使用函数 func（接受两个参数并返回一个参数）聚合数据集的元素。 该函数应该是可交换的和关联的，以便它可以被正确地并行计算。
collect()	在驱动程序中将数据集的所有元素作为数组返回。 这通常在过滤器或其他返回足够小的数据子集的操作之后很有用。
count()	返回数据集中元素的数量。
first()	返回数据集的第一个元素（类似于 take(1)）。
take(n)	返回一个数组，其中包含数据集的前 n 个元素。
takeSample(withReplacement, num, [seed])	返回一个数组，其中包含数据集的 num 个元素的随机样本，有或没有替换，可以选择预先指定随机数生成器种子。
takeOrdered(n, [ordering])	使用自然顺序或自定义比较器返回 RDD 的前 n 个元素。
saveAsTextFile(path)	将数据集的元素作为文本文件（或一组文本文件）写入本地文件系统、HDFS 或任何其他 Hadoop 支持的文件系统中的给定目录中。 Spark 将对每个元素调用 toString 以将其转换为文件中的一行文本。
saveAsSequenceFile(path) (Java and Scala)	将数据集的元素作为 Hadoop SequenceFile 写入本地文件系统、HDFS 或任何其他 Hadoop 支持的文件系统中的给定路径中。 这在实现 Hadoop 的 Writable 接口的键值对的 RDD 上可用。 在 Scala 中，它也可用于隐式转换为 Writable 的类型（Spark 包括对 Int、Double、String 等基本类型的转换）。
saveAsObjectFile(path) (Java and Scala)	使用 Java 序列化以简单格式编写数据集的元素，然后可以使用 SparkContext.objectFile() 加载。
countByKey()	仅适用于 (K, V) 类型的 RDD。 返回 (K, Int) 对的哈希图，其中包含每个键的计数。
foreach(func)	在数据集的每个元素上运行一个函数 func。 这通常是为了副作用，例如更新 Accumulator 或与外部存储系统交互。 注意：在 foreach() 之外修改除累加器之外的变量可能会导致未定义的行为。 有关更多详细信息，请参阅 了解闭包 。

​ Spark RDD API 还公开了一些操作的异步版本，例如 foreach 的 foreachAsync，它会立即向调用者返回一个 FutureAction，而不是在操作完成时阻塞。 这可用于管理或等待动作的异步执行。