Spark RDD 算子

偏执的太偏执、 2023-06-11 03:12 165阅读 0赞

**目录**

Basics

Passing Functions to Spark

闭包

Example

Local vs. cluster modes

Printing elements of an RDD-打印RDD中的每个元素

Working with Key-Value Pairs

Transformations

Actions

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

RDD 支持两种类型的操作：transformations (从现有数据集创建新数据集)和 actions (在数据集上运行计算后向驱动程序返回一个值)。例如，map 是一个 transformation 算子，它通过一个函数传递每个数据集元素并返回一个表示结果的新 RDD。另一方面，reduce 是一个 action 算子，它使用某个函数聚合 RDD 的所有元素，并将最终结果返回给驱动程序(尽管还有一个返回分布式数据集的 reduceByKey)。

Spark 中的所有 transformations 算子都是惰性的，因为它们不会立即计算结果。相反，他们只记得应用到某些基本数据集(例如文件)的转换。只有当操作需要将结果返回给驱动程序时，才会计算转换。这种设计使 Spark 能够更有效地运行。例如，我们可以意识到，通过 map 创建的数据集将在 reduce 中使用，并且只将 reduce 的结果返回给驱动程序，而不是更大的映射数据集。

默认情况下，每次对每个转换后的 RDD 运行一个操作时，都可以重新计算它。但是，也可以使用 persist (或cache)方法将 RDD 持久化到内存中，在这种情况下，Spark 会将元素保存在集群中，以便在下一次查询时更快地访问它。还支持在磁盘上持久存储 RDD，或跨多个节点复制 RDD。

## Basics ##

为了说明RDD的基础，考虑下面的简单程序：

val lines = sc.textFile("data.txt") // 读取本地文件data.txt
    val lineLengths = lines.map(s => s.length) // 返回每一行字符串的长度
    val totalLength = lineLengths.reduce((a, b) => a + b) // 将每个字符串的长度累加并返回

第一行定义了来自外部文件的基本 RDD。这个数据集没有加载到内存中，也没有在其他地方执行：行只是指向文件的指针。第二行将 lineLengths 定义为映射转换的结果。同样，由于懒惰，lineLengths 不是立即计算的。最后，运行 reduce，这是一个 action 算子。此时，Spark 将计算分解为在不同的机器上运行的任务，每台机器都运行其部分映射和本地聚合，只向 Driver 返回其结果。

如果我们稍后还想再次使用 lineLengths，我们可以添加：

lineLengths.persist()

在 reduce 之前，这会导致 lineLengths 在第一次计算之后保存到内存中。

## Passing Functions to Spark ##

Spark 的 API 严重依赖于驱动程序中传递的在集群上运行的函数。有两种推荐的方法：

*  匿名函数语法，可用于一小段代码。
 *  全局单例对象中的静态方法。例如，您可以定义对象 MyFunctions，然后传递 MyFunctions。func1如下所示：

object MyFunctions {
      def func1(s: String): String = { ... }
    }
    
    myRdd.map(MyFunctions.func1)

注意，虽然也可以将引用传递给类实例中的方法(与单例对象相反)，但这需要同时发送包含该类的对象和方法。例如考虑：

class MyClass {
      def func1(s: String): String = { ... }
      def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
    }

在这里，如果我们创建一个新的 MyClass 实例并在其上调用 doStuff，那么其中的映射将引用 MyClass 实例的 func1 方法，因此需要将整个对象发送到集群。它类似于编写 `rdd.map(x => this.func1(x))`。

以类似的方式，访问外部对象的字段将引用整个对象：

class MyClass {
      val field = "Hello"
      def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
    }

等同于编写 `rdd.map(x => this.field + x)`，它引用了所有这些。为了避免这个问题，最简单的方法是将字段复制到一个局部变量中，而不是从外部访问它：

def doStuff(rdd: RDD[String]): RDD[String] = {
      val field_ = this.field
      rdd.map(x => field_ + x)
    }

##  闭包 ##

Spark 的难点之一是理解跨集群执行代码时变量和方法的范围和生命周期。在范围之外修改变量的 RDD 操作可能经常引起混淆。在下面的示例中，我们将查看使用 foreach() 递增计数器的代码，但是其他操作也可能出现类似的问题。

### Example ###

考虑下面简单的 RDD 元素 sum，它的行为可能会根据是否在相同的 JVM 中执行而有所不同。一个常见的例子是在本地模式下运行Spark (- master = local\[n\]) 与将Spark应用程序部署到集群(例如，通过Spark -submit to YARN)：

var counter = 0
    var rdd = sc.parallelize(data)
    
    // Wrong: Don't do this!!
    rdd.foreach(x => counter += x)
    
    println("Counter value: " + counter)

### Local vs. cluster modes ###

上述代码的行为是未定义的，可能无法按预期工作。为了执行作业，Spark 将 RDD 算子的处理分解为任务，每个任务由执行程序执行。在执行之前，Spark 计算任务的闭包。闭包是那些执行程序在 RDD 上执行其计算时必须可见的变量和方法(在本例中为foreach())。这个闭包被序列化并发送给每个执行器。

传递给每个执行器的闭包中的变量现在是副本，因此，当在 foreach 函数中引用 counter 时，它不再是驱动节点上的计数器。在驱动节点的内存中仍然有一个计数器，但它对执行器不再可见!执行者只看到来自序列化闭包的副本。因此，counter 的最终值仍然是零，因为 counter 上的所有操作都引用了序列化闭包中的值。

在本地模式下，在某些情况下，foreach 函数实际上会在与驱动程序相同的 JVM 中执行，并引用相同的原始计数器，并可能实际更新它。

为了确保在这类场景中定义良好的行为，应该使用累加器。Spark中的累加器专门用于提供一种机制，以便在集群中的工作节点之间执行分割时安全地更新变量。

一般来说，像循环或局部定义方法这样的闭包结构不应该用来改变全局状态。Spark 不定义或保证闭包外部引用的对象的突变行为。一些这样做的代码可能在本地模式下工作，但那只是偶然的，而且这样的代码在分布式模式下不会像预期的那样工作。如果需要全局聚合，则使用累加器。

### Printing elements of an RDD-打印RDD中的每个元素 ###

另一个常见的习惯用法是尝试使用 RDD.foreach(println) 或 RDD.map(println)打印出 RDD 的元素。在一台机器上，这将生成预期的输出并打印所有的 RDD 元素。但是，在集群模式下，执行器调用的 stdout 输出现在是写入执行器的 stdout，而不是写入驱动程序上的 stdout，所以驱动程序上的 stdout 不会显示这些要打印驱动程序上的所有元素，可以使用 collect()方法首先将 RDD 带到驱动程序节点，如下所示：RDD.collect().foreach(println)。这可能会导致驱动程序耗尽内存。

## Working with Key-Value Pairs ##

虽然大多数 Spark 算子在包含任何类型对象的 RDD 上工作，但是只有少数特殊算子在键值对的 RDD 上可用。最常见的是分布式 shuffle 算子，如按键分组或聚合元素。

在 Scala 中，这些算子在包含 Tuple2 对象(语言中内置的元组，通过简单的编写(a, b)创建)的 RDD 上自动可用。键值对操作在 PairRDDFunctions 类中是可用的，它自动包装元组的 RDD。

例如，下面的代码使用对键-值对的 reduceByKey 操作来计算文件中每行文本出现的次数：

val lines = sc.textFile("data.txt")
    val pairs = lines.map(s => (s, 1))
    val counts = pairs.reduceByKey((a, b) => a + b)

例如，我们还可以使用 counts.sortByKey() 来按字母顺序排序对数据，最后使用 counts.collect() 将它们作为对象数组返回到 Driver。

> 注意：在使用自定义对象作为键-值对操作中的键时，必须确保自定义 equals()方法附带匹配的 hashCode()方法。

## Transformations ##

<table> 
 <tbody> 
 <tr> 
 <th style="width:287px;">算子</th> 
 <th style="width:110px;">返回类型</th> 
 <th style="width:452px;">描述</th> 
 </tr> 
 <tr> 
 <td style="width:287px;">map(func)</td> 
 <td style="width:110px;">org.apache.spark.rdd.RDD[U]</td> 
 <td style="width:452px;">通过函数func传递源的每个元素，返回一个新的分布式数据集。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">filter(func)</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[T]</code></pre> </td> 
 <td style="width:452px;">通过选择func返回true的源元素来返回一个新的数据集。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">flatMap(func)</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[U]</code></pre> </td> 
 <td style="width:452px;">与map类似，但是每个输入项可以映射到0或多个输出项(因此func应该返回一个Seq，而不是多个单项)。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">mapPartitions(func) <a rel="nofollow"></a></td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[U]</code></pre> </td> 
 <td style="width:452px;">与map类似，但是在RDD的每个分区(块)上单独运行，所以func在类型为T的RDD上运行时必须是类型Iterator&lt;T&gt; =&gt; Iterator。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">mapPartitionsWithIndex(func)</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[U]</code></pre> </td> 
 <td style="width:452px;">与mappartition类似，但也为func提供了一个表示分区索引的整数值，所以func的类型必须是(Int, Iterator&lt;T&gt;) =&gt;迭代器，当运行在类型为T的RDD上时。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">sample(withReplacement, fraction, seed)</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[T]</code></pre> </td> 
 <td style="width:452px;">使用给定的随机数生成器种子对数据的一部分进行采样，不论是否替换。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">union(otherDataset)</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[T]</code></pre> </td> 
 <td style="width:452px;">返回一个新数据集，该数据集包含源数据集中的元素和参数的并集。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">intersection(otherDataset)</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[T]</code></pre> </td> 
 <td style="width:452px;">返回一个新的RDD，其中包含源数据集中的元素和参数的交集。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">distinct([numPartitions]))</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[T]</code></pre> </td> 
 <td style="width:452px;">返回包含源数据集的不同元素的新数据集。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">groupByKey([numPartitions]) <a rel="nofollow"></a></td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[scala.Tuple2[K, scala.Iterable[V]]]</code></pre> </td> 
 <td style="width:452px;">当调用一个(K, V)对的数据集时，返回一个(K，可迭代&lt;V&gt;)对的数据集。 Note:&nbsp;如果分组是为了对每个键执行聚合(比如求和或平均值)，那么使用 reduceByKey 或 aggregateByKey 将产生更好的性能。 Note: 默认情况下，输出中的并行度取决于父 RDD 的分区数。您可以传递一个可选的 numPartitions 参数来设置不同数量的任务。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">reduceByKey(func, [numPartitions]) <a rel="nofollow"></a></td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[scala.Tuple2[K, V]]</code></pre> </td> 
 <td style="width:452px;">当对一个(K, V)对的数据集调用时，返回一个(K, V)对的数据集，其中每个键的值使用给定的 reduce 函数 func 进行聚合，与 groupByKey 类似，reduce 任务的数量可以通过第二个可选参数进行配置。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">aggregateByKey(zeroValue)(seqOp, combOp, [numPartitions]) <a rel="nofollow"></a></td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[scala.Tuple2[K, U]]</code></pre> </td> 
 <td style="width:452px;">当对一个(K, V)对的数据集调用时，返回一个(K, U)对的数据集，其中每个键的值使用给定的 combine 函数和一个中性的“0”值进行聚合。允许与输入值类型不同的聚合值类型，同时避免不必要的分配。与 groupByKey 类似，reduce 任务的数量可以通过第二个可选参数进行配置。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">sortByKey([ascending], [numPartitions]) <a rel="nofollow"></a></td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[scala.Tuple2[K, V]]</code></pre> </td> 
 <td style="width:452px;">当调用一个(K, V)对的数据集时，其中K实现有序，返回按键升序或降序排序的(K, V)对的数据集，如布尔升序参数中指定的。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">join(otherDataset, [numPartitions]) <a rel="nofollow"></a></td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[scala.Tuple2[K, scala.Tuple2[V, W]]]</code></pre> </td> 
 <td style="width:452px;">当在类型(K, V)和(K, W)的数据集上调用时，返回一个(K， (V, W))对的数据集，其中包含每个键的所有对元素。外部连接由 leftOuterJoin、rightOuterJoin 和 fullOuterJoin 支持。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">cogroup(otherDataset, [numPartitions]) <a rel="nofollow"></a></td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[scala.Tuple2[K, scala.Tuple4[scala.Iterable[V], scala.Iterable[W1], scala.Iterable[W2], scala.Iterable[W3]]]]</code></pre> </td> 
 <td style="width:452px;">当调用类型(K, V)和(K, W)的数据集时，返回一个(K， (Iterable&lt;V&gt;， Iterable&lt;W&gt;))元组的数据集。这个操作也称为 groupWith。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">cartesian(otherDataset)</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[scala.Tuple2[T, U]]</code></pre> </td> 
 <td style="width:452px;">在类型为T和U的数据集上调用时，返回一个(T, U)对(所有对元素)的数据集。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">pipe(command, [envVars])</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[scala.Predef.String]</code></pre> </td> 
 <td style="width:452px;">通过 shell 命令(例如 Perl 或 bash 脚本)对 RDD 的每个分区进行管道传输。将 RDD 元素写入进程的stdin，并将输出到其 stdout 的行作为字符串的 RDD 返回。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">coalesce(numPartitions) <a rel="nofollow"></a></td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[T]</code></pre> </td> 
 <td style="width:452px;">将 RDD 中的分区数量减少到 numpartition。适用于过滤大型数据集后更有效地执行算子。</td> 
 </tr> 
 <tr> 
 <td style="width:287px;">repartition(numPartitions)</td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[T]</code></pre> </td> 
 <td style="width:452px;">随机地重新洗牌 RDD 中的数据，以创建更多或更少的分区，并在它们之间进行平衡。这总是通过网络打乱所有数据。 <a rel="nofollow"></a></td> 
 </tr> 
 <tr> 
 <td style="width:287px;">repartitionAndSortWithinPartitions(partitioner) <a rel="nofollow"></a></td> 
 <td style="width:110px;"> <pre><code>org.apache.spark.rdd.RDD[scala.Tuple2[K, V]]</code></pre> </td> 
 <td style="width:452px;">根据给定的分区程序重新划分RDD，在每个结果分区中，根据键对记录进行排序。这比&nbsp;<code>repartition 然后在每个分区内 sorting 的效率更高，</code>因为它可以将排序向下推到 shuffle 制中。</td> 
 </tr> 
 </tbody> 
</table>

## Actions ##

<table> 
 <tbody> 
 <tr> 
 <th>算子</th> 
 <th>描述</th> 
 </tr> 
 <tr> 
 <td>reduce(func)</td> 
 <td>使用函数func(它接受两个参数并返回一个)聚合数据集的元素。这个函数应该是交换的和结合的，这样它才能被正确地并行计算。</td> 
 </tr> 
 <tr> 
 <td>collect()</td> 
 <td>在驱动程序中以数组的形式返回数据集的所有元素。这通常在筛选器或其他操作返回足够小的数据子集之后才有用。</td> 
 </tr> 
 <tr> 
 <td>count()</td> 
 <td>返回数据集中元素的数量。</td> 
 </tr> 
 <tr> 
 <td>first()</td> 
 <td>返回数据集的第一个元素(类似于take(1))。</td> 
 </tr> 
 <tr> 
 <td>take(n)</td> 
 <td>返回数据集的前n个元素的数组。</td> 
 </tr> 
 <tr> 
 <td>takeSample(withReplacement, num, [seed])</td> 
 <td>返回一个数组，其中包含数据集的随机num元素样本，可以替换，也可以不替换，可以预先指定随机数生成器种子。</td> 
 </tr> 
 <tr> 
 <td>takeOrdered(n, [ordering])</td> 
 <td>使用RDD的自然顺序或自定义比较器返回RDD的前n个元素。</td> 
 </tr> 
 <tr> 
 <td>saveAsTextFile(path)</td> 
 <td>将数据集的元素作为文本文件(或文本文件集)写入本地文件系统、HDFS或任何其他hadoop支持的文件系统的给定目录中。Spark将对每个元素调用toString，将其转换为文件中的一行文本。</td> 
 </tr> 
 <tr> 
 <td>saveAsSequenceFile(path) (Java and Scala)</td> 
 <td>将数据集的元素作为Hadoop SequenceFile写入本地文件系统、HDFS或任何其他Hadoop支持的文件系统的给定路径中。这在实现Hadoop的可写接口的键值对的RDDs上是可用的。在Scala中，它也可用于隐式转换为可写的类型(Spark包括对Int、Double、String等基本类型的转换)。</td> 
 </tr> 
 <tr> 
 <td>saveAsObjectFile(path) (Java and Scala)</td> 
 <td>使用Java序列化以简单的格式编写数据集的元素，然后可以使用SparkContext.objectFile()加载这些元素。</td> 
 </tr> 
 <tr> 
 <td>countByKey() <a rel="nofollow"></a></td> 
 <td>仅在类型(K, V)的RDDs上可用。返回(K, Int)对的hashmap和每个键的计数。</td> 
 </tr> 
 <tr> 
 <td>foreach(func)</td> 
 <td>对数据集的每个元素运行函数func。这通常是为了解决诸如更新累加器或与外部存储系统交互等副作用。 注意：在foreach()之外修改除累加器之外的其他变量可能会导致未定义的行为。</td> 
 </tr> 
 </tbody> 
</table>