1-3 Spark RDD 编程指南

概述

与 Spark 链接

使用Shell

初始化 Spark

并行集合

外部数据集

基本

将函数传递给 Spark

例子

本地模式与集群模式

打印 RDD 的元素

了解闭包

使用键值对

Transformations

Actions

背景

性能影响

Shuffle operations

RDD操作

选择哪个存储级别？

删除数据

RDD持久化

弹性分布式数据集 (RDD)

广播变量

累加器

共享变量

部署到集群

从 Java/Scala 启动 Spark 作业

单元测试

其他

目录

从较高的层面来看，每个 Spark 应用程序都包含一个驱动程序，该程序运行用户的函数并在集群上main执行各种并行操作。Spark 提供的主要抽象是弹性分布式数据集(RDD)，它是跨集群节点分区的元素集合，可以并行操作。RDD 是通过从 Hadoop 文件系统（或任何其他 Hadoop 支持的文件系统）中的文件或驱动程序中现有的 Scala 集合开始，然后对其进行转换来创建的。用户还可以要求 Spark 将RDD持久保存在内存中，从而使其能够在并行操作中高效地重用。最后，RDD 会自动从节点故障中恢复。Spark 中的第二个抽象是可在并行操作中使用的共享变量。默认情况下，当 Spark 将函数作为一组任务在不同节点上并行运行时，它会将函数中使用的每个变量的副本发送给每个任务。有时，需要在任务之间或任务和驱动程序之间共享变量。Spark 支持两种类型的共享变量：广播变量（可用于在所有节点上的内存中缓存值）和累加器（仅“添加”变量，例如计数器和总和）。本指南以 Spark 支持的每种语言展示了其中的每一项功能。bin/spark-shell如果您启动 Spark 的交互式 shell（无论是 Scala shell 还是 bin/pysparkPython shell），那么最容易遵循。

Spark 2.3.0 的构建和分发默认与 Scala 2.11 配合使用。（Spark 也可以与其他版本的 Scala 一起使用。）要在 Scala 中编写应用程序，您需要使用兼容的 Scala 版本（例如 2.11.X）。要编写 Spark 应用程序，您需要添加 Spark 的 Maven 依赖项。Spark 可通过 Maven Central 获取：groupId = org.apache.sparkartifactId = spark-core_2.11version = 2.3.0此外，如果您希望访问 HDFS 集群，则需要 hadoop-client为您的 HDFS 版本添加依赖项。groupId = org.apache.hadoopartifactId = hadoop-clientversion = <your-hdfs-version>最后，您需要将一些 Spark 类导入到您的程序中。添加以下行：import org.apache.spark.SparkContextimport org.apache.spark.SparkConf（在 Spark 1.3.0 之前，您需要显式import org.apache.spark.SparkContext._启用必要的隐式转换。）

与 Spark 链接

Spark 程序必须做的第一件事是创建一个SparkContext对象，它告诉 Spark 如何访问集群。要创建 SparkConf，SparkContext您首先需要构建一个包含有关您的应用程序的信息的SparkConf对象。每个 JVM 只能有一个 SparkContext 处于活动状态。stop()在创建新 SparkContext 之前，您必须先激活 SparkContext。val conf = new SparkConf().setAppName(appName).setMaster(master)new SparkContext(conf)该appName参数是您的应用程序在集群 UI 上显示的名称。 master是Spark、Mesos 或 YARN 集群 URL，或在本地模式下运行的特殊“本地”字符串。实际上，当在集群上运行时，您不会希望master在程序中进行硬编码，而是希望启动应用程序并spark-submit在那里接收它。但是，对于本地测试和单元测试，您可以通过“local”在进程内运行 Spark。

初始化 Spark

在 Spark shell 中，已经为您创建了一个特殊的解释器感知 SparkContext，位于名为 的变量中sc。制作自己的 SparkContext 是行不通的。您可以使用参数设置上下文连接到哪个主机--master，并且可以通过将逗号分隔的列表传递给参数来将 JAR 添加到类路径--jars。您还可以通过向参数提供以逗号分隔的 Maven 坐标列表来将依赖项（例如 Spark 包）添加到 shell 会话中--packages。可能存在依赖关系的任何其他存储库（例如 Sonatype）都可以传递给参数--repositories。例如，要bin/spark-shell在四个核心上运行，请使用：$ ./bin/spark-shell --master local[4]或者，要添加code.jar到其类路径中，请使用：$ ./bin/spark-shell --master local[4] --jars code.jar要使用 Maven 坐标包含依赖项：$ ./bin/spark-shell --master local[4] --packages \"org.example:example:0.1\"要获得完整的选项列表，请运行spark-shell --help。在幕后， spark-shell调用更通用的spark-submit脚本。

Spark 围绕弹性分布式数据集(RDD)的概念，它是可以并行操作的容错元素集合。创建 RDD 有两种方法：并行 化驱动程序中的现有集合，或引用外部存储系统中的数据集，例如共享文件系统、HDFS、HBase 或任何提供 Hadoop InputFormat 的数据源。

Spark 可以从 Hadoop 支持的任何存储源创建分布式数据集，包括本地文件系统、HDFS、Cassandra、HBase、Amazon S3等。Spark 支持TexfFiles、SequenceFiles和任何其他 Hadoop InputFormat。文本文件 RDD 可以使用SparkContext的textFile方法创建。此方法采用文件的 URI（计算机上的本地路径，或 hdfs://和s3a://URI等）并将其作为行集合读取。下面是一个调用示例：scala> val distFile = sc.textFile(\"data.txt\

外部Datasets

RDD 支持两种类型的操作：转换（从现有数据集创建新数据集）和操作（在对数据集运行计算后将值返回给驱动程序）。例如，map是一种转换，它将每个数据集元素传递给函数并返回表示结果的新 RDD。另一方面，reduce是使用某个函数聚合 RDD 的所有元素并将最终结果返回给驱动程序的操作（尽管也有reduceByKey返回分布式数据集的并行操作）。Spark 中的所有转换都是惰性的，因为它们不会立即计算结果。相反，他们只记住应用于某些基础数据集（例如文件）的转换。仅当操作需要将结果返回到驱动程序时才计算转换。这样的设计使得Spark能够更高效地运行。例如，我们可以意识到通过创建的数据集map将在 a 中使用reduce，并且仅将 a 的结果返回reduce给驱动程序，而不是更大的映射数据集。默认情况下，每次对每个转换后的 RDD 运行操作时，都可能会重新计算它。但是，您也可以使用persistcache(或Cache) 方法将RDD保留在内存中，在这种情况下，Spark 会将元素保留在集群上，以便下次查询时更快地访问。还支持将 RDD 持久保存在磁盘上或跨多个节点复制。

为了说明 RDD 基础知识，请考虑下面的简单程序：val lines = sc.textFile(\"data.txt\

Basics

Spark的API很大程度上依赖于驱动程序中传递函数来在集群上运行。有两种推荐的方法可以做到这一点：匿名函数语法，可用于短代码。全局单例对象中的静态方法。例如，您可以定义object 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它，map里面会引用 该实例的func1方法，因此需要将整个对象发送到集群。这与写作类似。MyClassrdd.map(x => this.func1(x))以类似的方式，访问外部对象的字段将引用整个对象：class MyClass {  val field = \"Hello\"  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }}相当于writing rdd.map(x => this.field + x)，它引用了所有的this. 为了避免这个问题，最简单的方法是复制field到局部变量中，而不是从外部访问它：def doStuff(rdd: RDD[String]): RDD[String] = {  val field_ = this.field  rdd.map(x => field_ + x)}

给Spark 传递函数

Spark 的难点之一是理解跨集群执行代码时变量和方法的范围和生命周期。修改超出其范围的变量的 RDD 操作可能是造成混乱的常见原因。在下面的示例中，我们将查看用于foreach()递增计数器的代码，但其他操作也可能会出现类似的问题。例子考虑下面的朴素 RDD 元素总和，它的行为可能会有所不同，具体取决于执行是否发生在同一 JVM 中。一个常见的示例是在local模式 ( --master = local[n]) 下运行 Spark 与将 Spark 应用程序部署到集群（例如，通过 Spark-submit 到 YARN）：var counter = 0var rdd = sc.parallelize(data)// Wrong: Don't do this!!rdd.foreach(x => counter += x)println(\"Counter value: \" + counter)

上述代码的行为未定义，可能无法按预期工作。为了执行作业，Spark 将 RDD 操作的处理分解为任务，每个任务都由执行器执行。在执行之前，Spark 计算任务的闭包。闭包是执行器在 RDD 上执行计算时必须可见的变量和方法（在本例中foreach()）。该闭包被序列化并发送到每个执行器。发送到每个执行器的闭包中的变量现在是副本，因此，当在函数内引用计数器foreach时，它不再是驱动程序节点上的计数器。驱动程序节点的内存中仍然有一个计数器，但执行程序不再可见！执行者只能看到序列化闭包中的副本。因此，计数器的最终值仍然为零，因为计数器上的所有操作都引用序列化闭包内的值。在本地模式下，在某些情况下，该foreach函数实际上将在与驱动程序相同的 JVM 中执行，并将引用相同的原始计数器，并且实际上可能会更新它。为了确保在这些场景中定义良好的行为，应该使用Accumulator. Spark 中的累加器专门用于提供一种机制，用于在集群中的工作节点之间分割执行时安全地更新变量。本指南的累加器部分更详细地讨论了这些内容。一般来说，闭包——像循环或本地定义的方法这样的结构，不应该被用来改变一些全局状态。Spark 不定义或保证从闭包外部引用的对象的突变行为。执行此操作的某些代码可能在本地模式下工作，但这只是偶然的，并且此类代码在分布式模式下不会按预期运行。如果需要一些全局聚合，请改用累加器。

dd.foreach(println)另一个常见的习惯用法是尝试使用或打印出 RDD 的元素rdd.map(println)。在一台机器上，这将生成预期的输出并打印 RDD 的所有元素。但是，在cluster模式下，stdout执行程序调用的输出现在写入执行程序stdout，而不是驱动程序上的输出，因此stdout驱动程序不会显示这些！要打印驱动程序上的所有元素，可以使用以下collect()方法首先将 RDD 带到驱动程序节点：rdd.collect().foreach(println)。不过，这可能会导致驱动程序内存不足，因为collect()将整个 RDD 提取到一台机器上；如果您只需要打印 RDD 的几个元素，更安全的方法是使用take(): rdd.take(100).foreach(println)。

打印Rdd元素

下表列出了 Spark 支持的一些常见转换。有关详细信息，请参阅 RDD API 文档（Scala、 Java、 Python、 R）和配对 RDD 函数文档（Scala、 Java ）。

回一个新的分布式数据集，该数据集通过将源的每个元素传递给函数func来形成。【一对一】

map(func)

返回一个新的数据集，该数据集是通过选择func返回 true的源元素而形成的。

filter(func)

与map类似，但每个输入项可以映射到0个或多个输出项（因此func应该返回一个Seq而不是单个项）。

flatMap(func)

与map类似，但在RDD的每个分区（块）上单独运行，因此当在T类型的RDD上运行时， func必须是 Iterator<T> => Iterator<U>类型。

mapPartitions(func)

mapPartitionsWithIndex(func)

返回一个新数据集，其中包含源数据集中元素和参数的并集。

union(otherDataset)

回一个新的 RDD，其中包含源数据集中元素与参数的交集。

intersection(otherDataset)

返回一个新数据集，其中包含源数据集的不同元素。

distinct([numPartitions]))

groupByKey([numPartitions])

join

cogroup

cartesian

通过 shell 命令（例如 Perl 或 bash 脚本）对 RDD 的每个分区进行管道传输。RDD 元素被写入进程的 stdin，并且输出到其 stdout 的行作为字符串 RDD 返回。

pipe

将 RDD 中的分区数量减少到 numPartitions。对于过滤大型数据集后更有效地运行操作很有用。

coalesce

随机重新整理 RDD 中的数据以创建更多或更少的分区并在它们之间进行平衡。这总是会打乱网络上的所有数据。

parttitions

根据给定的分区程序对 RDD 进行重新分区，并在每个结果分区中按记录的键对记录进行排序。这比repartition在每个分区内调用然后排序更有效，因为它可以将排序下推到洗牌机制中。

repartitionAndSortWithinPartitions

使用函数func聚合数据集的元素（它接受两个参数并返回一个）。该函数应该是可交换的和关联的，以便可以正确地并行计算。

reduce

在驱动程序中将数据集的所有元素作为数组返回。在返回足够小的数据子集的过滤器或其他操作之后，这通常很有用。

collect

返回数据集中的元素数量。

count

返回数据集的第一个元素（类似于 take(1)）。

first

返回包含数据集前n 个元素的数组。

take(n)

返回一个数组，其中包含数据集num 个元素的随机样本，有或没有替换，可以选择预先指定随机数生成器种子。

使用自然顺序或自定义比较器返回 RDD 的前n 个元素。

将数据集的元素作为 Hadoop SequenceFile 写入本地文件系统、HDFS 或任何其他 Hadoop 支持的文件系统的给定路径中。这在实现 Hadoop 的 Writable 接口的键值对 RDD 上可用。在 Scala 中，它也适用于可隐式转换为 Writable 的类型（Spark 包括对 Int、Double、String 等基本类型的转换）。

saveAsTextFile(path)

saveAsSequenceFile(path)(Java and Scala)

使用 Java 序列化以简单格式写入数据集的元素，然后可以使用 SparkContext.objectFile().

saveAsObjectFile(path)(Java and Scala)

countByKey

对数据集的每个元素运行函数func 。这样做通常是为了防止副作用，例如更新累加器或与外部存储系统交互。注意：修改累加器之外的变量foreach()可能会导致未定义的行为。有关更多详细信息，请参阅了解闭包。

foreach

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

Spark 中的某些操作会触发称为 shuffle 的事件。shuffle 是 Spark 重新分配数据的机制，以便数据在分区之间以不同的方式分组。这通常涉及跨Executors和machines复制数据，从而使Shuffle成为复杂且成本高昂的操作。

为了了解Shuffle期间发生的情况，我们可以考虑该 reduceByKey操作的示例。reduceByKey操作会生成一个新的 RDD，其中单个键的所有值都组合成一个元组 （键以及针对与该键关联的所有值执行reduce函数的结果）。难点在于，单个键的所有值不一定都位于同一分区，甚至同一台机器上，但它们必须位于同一位置才能计算结果。在 Spark 中，数据通常不会跨分区分布到特定操作的必要位置。在计算过程中，单个任务将在单个分区上运行 - 因此，为了组织单个reduce任务执行educeByKey的所有数据r，Spark需要执行all-to-all操作。它必须从所有分区中读取以查找所有键的所有值，然后将跨分区的值汇集在一起​​以计算每个键的最终结果 - 这称为shuffle。虽然新打乱的数据的每个分区中的元素集是确定性的，分区本身的顺序也是确定性的，但这些元素的顺序却不是。如果需要在洗牌后得到可预测的有序数据，那么可以使用：mapPartitions对每个分区进行排序，例如，.sortedrepartitionAndSortWithinPartitions有效地对分区进行排序，同时重新分区sortBy制作一个全局有序的 RDD可能会引起Suffffle的操作有， 重新repartitioin的操作，比如： repartition 和coalesce; ByKey 的操作（除了Counting的操作），比如groupByKey 和reduceByKey; join 的操作比如cogroup 和join.

Shuffle Operations

Spark 最重要的功能之一是跨操作在内存中持久保存（或缓存）数据集。当您持久化 RDD 时，每个节点都会将其计算的任何分区存储在内存中，并在该数据集（或从其派生的数据集）上的其他操作中重用它们。这使得未来的操作速度更快（通常快 10 倍以上）。缓存是迭代算法和快速交互使用的关键工具。您可以使用persist()或cache()方法将RDD 标记为要持久化。第一次在操作中计算时，它将保存在节点的内存中。Spark 的缓存是容错的——如果 RDD 的任何分区丢失，它将使用最初创建它的转换自动重新计算。此外，每个持久化的 RDD 都可以使用不同的存储级别进行存储，例如，允许您将数据集持久化在磁盘上、将其持久化在内存中、但作为序列化的 Java 对象（以节省空间）、跨节点复制。这些级别是通过将 StorageLevel对象（Scala、 Java、 Python）传递给persist(). 该cache()方法是使用默认存储级别的简写，即StorageLevel.MEMORY_ONLY（将反序列化对象存储在内存中）。完整的存储级别集是：

MEMORY_ONLY： 将 RDD 作为反序列化的 Java 对象存储在 JVM 中。如果 RDD 不适合内存，某些分区将不会被缓存，并且会在每次需要时动态重新计算。这是默认级别。

MEMORY_AND_DISK(Java and Scala)： 将 RDD 作为反序列化的 Java 对象存储在 JVM 中。如果 RDD 无法容纳在内存中，请将无法容纳的分区存储在磁盘上，并在需要时从那里读取它们。

MEMORY_AND_DISK_SER(Java and Scala)： 将 RDD 存储为序列化Java 对象（每个分区一个字节数组）。这通常比反序列化对象更节省空间，特别是在使用 快速序列化器时，但读取时更需要 CPU 密集型。

DISK_ONLY： 仅将 RDD 分区存储在磁盘上。

OFF_HEAP (experimental)：   与MEMORY_ONLY_SER类似，但将数据存储在 堆外内存中。这需要启用堆外内存。

Spark 还会自动在 shuffle 操作中保留一些中间数据（例如reduceByKey），即使用户没有调用persist. 这样做是为了避免在Suffle期间节点失败时重新计算整个输入。如果用户打算重用生成的 RDD，我们仍然建议persist他们调用它。

存储级别

Spark 的存储级别旨在提供内存使用和 CPU 效率之间的不同权衡。我们建议通过以下过程来选择一个：

如果您的 RDD 适合默认存储级别 ( MEMORY_ONLY)，请保持这种状态。这是 CPU 效率最高的选项，允许 RDD 上的操作尽可能快地运行。

如果没有，请尝试使用MEMORY_ONLY_SER并选择一个快速序列化库，以使对象更加节省空间，但访问速度仍然相当快。（Java 和 Scala）

不要溢出到磁盘，除非计算数据集的函数很昂贵，或者它们过滤大量数据。否则，重新计算分区可能与从磁盘读取分区一样快。

如果您想要快速故障恢复（例如，如果使用 Spark 来处理来自 Web 应用程序的请求），请使用复制存储级别。所有存储级别都通过重新计算丢失的数据来提供完整的容错能力，但复制存储级别允许您继续在 RDD 上运行任务，而无需等待重新计算丢失的分区。

Spark 自动监控每个节点上的缓存使用情况，并以最近最少使用 (LRU) 方式删除旧数据分区。如果您想手动删除 RDD，而不是等待它从缓存中删除，请使用该RDD.unpersist()方法。

通常，当传递给 Spark 操作（例如map或reduce）的函数在远程集群节点上执行时，它会作用于该函数中使用的所有变量的单独副本。这些变量被复制到每台机器，并且远程机器上的变量更新不会传播回驱动程序。支持跨任务的通用读写共享变量效率很低。然而，Spark 确实为两种常见的使用模式提供了两种有限类型的共享变量：广播变量和累加器。

累加器是仅通过关联和交换运算“添加”的变量，因此可以有效地支持并行。它们可用于实现计数器（如在 MapReduce 中）或求和。Spark 原生支持数字类型的累加器，程序员可以添加对新类型的支持。作为用户，您可以创建命名或未命名的累加器。如下图所示，一个命名的累加器（在本例中counter）将显示在修改该累加器的阶段的 Web UI 中。Spark 在“任务”表中显示由任务修改的每个累加器的值。在 UI 中跟踪累加器对于了解运行阶段的进度很有用（注意：Python 尚不支持）。

可以通过调用SparkContext.longAccumulator()或SparkContext.doubleAccumulator() 分别累加 Long 或 Double 类型的值来创建数字累加器。然后可以使用该方法将在集群上运行的任务添加到集群中add。然而，他们无法读取其value。只有驱动程序可以使用累加器的方法读取累加器的值value。下面的代码显示了一个用于将数组元素相加的累加器：scala> val accum = sc.longAccumulator(\"My Accumulator\

 application submission guide介绍了如何向集群提交申请.简而言之，一旦您将应用程序打包到 JAR（对于 Java/Scala）或一组.py或.zip文件（对于 Python）中，该bin/spark-submit脚本就可以让您将其提交给任何支持的集群管理器。

org.apache.spark.launcher 包提供了使用简单的 Java API 将 Spark 作业作为子进程启动的类。

Spark 适合使用任何流行的单元测试框架进行单元测试。只需在您的测试中创建一个，SparkContext并将主 URL 设置为local，运行您的操作，然后调用SparkContext.stop()将其拆除。确保在finally块或测试框架tearDown方法内停止Context，因为 Spark 不支持在同一程序中同时运行两个Context。

您可以在 Spark 网站上查看一些示例 Spark 程序。此外，Spark 在examples目录中包含多个示例（Scala、 Java、 Python、 R）。您可以通过将类名传递给 Spark 的bin/run-example脚本来运行 Java 和 Scala 示例；例如：./bin/run-example SparkPi对于 Python 示例，请spark-submit改用：./bin/spark-submit examples/src/main/python/pi.py对于 R 示例，请spark-submit改用：./bin/spark-submit examples/src/main/r/dataframe.R为了帮助优化您的程序，配置和 调整指南提供了有关最佳实践的信息。它们对于确保您的数据以有效的格式存储在内存中尤其重要。为了帮助部署，集群模式概述描述了分布式操作中涉及的组件和支持的集群管理器。最后，完整的 API 文档提供了 Scala、Java、Python和R版本。

1-3 RDD 编程指南