Spark 编程指南

• 概述
• Spark 依赖
• 初始化 Spark
  • 使用 Shell
• 弹性分布式数据集 (RDDs)
  • 并行集合
  • 外部 Datasets（数据集）
  • RDD 操作
    • 基础
    • 传递 Functions（函数）给 Spark
    • 理解闭包
      • 示例
      • Local（本地）vs. cluster（集群）模式
      • 打印 RDD 的 elements
    • 与 Key-Value Pairs 一起使用
    • Transformations（转换）
    • Actions（动作）
    • Shuffle 操作
      • Background（幕后）
      • 性能影响
  • RDD Persistence（持久化）
    • 如何选择存储级别 ?
    • 删除数据
• 共享变量
  • 广播变量
  • Accumulators（累加器）
• 部署应用到集群中
• 从 Java / Scala 启动 Spark jobs
• 单元测试
• 快速链接

概述

在一个较高的概念上来说，每一个 Spark 应用程序由一个在集群上运行着用户的 main 函数和执行各种并行操作的 driver program（驱动程序）组成。Spark 提供的主要抽象是一个弹性分布式数据集（RDD），它是可以执行并行操作且跨集群节点的元素的集合。RDD 可以从一个 Hadoop 文件系统（或者任何其它 Hadoop 支持的文件系统），或者一个在 driver program（驱动程序）中已存在的 Scala 集合，以及通过 transforming（转换）来创建一个 RDD。用户为了让它在整个并行操作中更高效的重用，也许会让 Spark persist（持久化）一个 RDD 到内存中。最后，RDD 会自动的从节点故障中恢复。

在 Spark 中的第二个抽象是能够用于并行操作的 shared variables（共享变量），默认情况下，当 Spark 的一个函数作为一组不同节点上的任务运行时，它将每一个变量的副本应用到每一个任务的函数中去。有时候，一个变量需要在整个任务中，或者在任务和 driver program（驱动程序）之间来共享。Spark 支持两种类型的共享变量 : broadcast variables（广播变量），它可以用于在所有节点上缓存一个值，和 accumulators（累加器），他是一个只能被 “added（增加）” 的变量，例如 counters 和 sums。

本指南介绍了每一种 Spark 所支持的语言的特性。如果您启动 Spark 的交互式 shell - 针对 Scala shell 使用 bin/spark-shell 或者针对 Python 使用 bin/pyspark 是很容易来学习的。

Spark 依赖

• Scala
• Java
• Python

groupId = org.apache.spark

artifactId = spark-core_2.11

version = 2.2.0

groupId = org.apache.hadoop

artifactId = hadoop-client

version = <your-hdfs-version>

import org.apache.spark.SparkContext

import org.apache.spark.SparkConf

初始化 Spark

• Scala
• Java
• Python

val conf = new SparkConf().setAppName(appName).setMaster(master)

new SparkContext(conf)

这个 appName 参数是一个在集群 UI 上展示应用程序的名称。 master 是一个 Spark, Mesos 或 YARN 的 cluster URL，或者指定为在 local mode（本地模式）中运行的 “local” 字符串。在实际工作中，当在集群上运行时，您不希望在程序中将 master 给硬编码，而是用 使用 spark-submit 启动应用并且接收它。然而，对于本地测试和单元测试，您可以通过 “local” 来运行 Spark 进程。

使用 Shell

• Scala
• Python

$ ./bin/spark-shell --master local[4]

$ ./bin/spark-shell --master local[4] --jars code.jar

$ ./bin/spark-shell --master local[4] --packages "org.example:example:0.1"

弹性分布式数据集 (RDDs)

Spark 主要以一个 弹性分布式数据集（RDD）的概念为中心，它是一个容错且可以执行并行操作的元素的集合。有两种方法可以创建 RDD : 在你的 driver program（驱动程序）中 parallelizing 一个已存在的集合，或者在外部存储系统中引用一个数据集，例如，一个共享文件系统，HDFS，HBase，或者提供 Hadoop InputFormat 的任何数据源。

并行集合

• Scala
• Java
• Python

val data = Array(1, 2, 3, 4, 5)

val distData = sc.parallelize(data)

并行集合中一个很重要参数是 partitions（分区）的数量，它可用来切割 dataset（数据集）。Spark 将在集群中的每一个分区上运行一个任务。通常您希望群集中的每一个 CPU 计算 2-4 个分区。一般情况下，Spark 会尝试根据您的群集情况来自动的设置的分区的数量。当然，您也可以将分区数作为第二个参数传递到 parallelize (e.g. sc.parallelize(data, 10)) 方法中来手动的设置它。注意: 代码中的一些地方会使用 term slices (a synonym for partitions) 以保持向后兼容.

外部 Datasets（数据集）

• Scala
• Java
• Python

scala> val distFile = sc.textFile("data.txt")

distFile: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[10] at textFile at <console>:26

RDD 操作

RDDs support 两种类型的操作: transformations（转换）, 它会在一个已存在的 dataset 上创建一个新的 dataset, 和 actions（动作）, 将在 dataset 上运行的计算后返回到 driver 程序. 例如, map 是一个通过让每个数据集元素都执行一个函数，并返回的新 RDD 结果的 transformation, reducereduce 通过执行一些函数，聚合 RDD 中所有元素，并将最终结果给返回驱动程序（虽然也有一个并行 reduceByKey 返回一个分布式数据集）的 action.

Spark 中所有的 transformations 都是 lazy（懒加载的）, 因此它不会立刻计算出结果. 相反, 他们只记得应用于一些基本数据集的转换 (例如. 文件). 只有当需要返回结果给驱动程序时，transformations 才开始计算. 这种设计使 Spark 的运行更高效. 例如, 我们可以了解到，map 所创建的数据集将被用在 reduce 中，并且只有 reduce 的计算结果返回给驱动程序，而不是映射一个更大的数据集.

默认情况下，每次你在 RDD 运行一个 action 的时， 每个 transformed RDD 都会被重新计算。但是，您也可用 persist (或 cache) 方法将 RDD persist（持久化）到内存中；在这种情况下，Spark 为了下次查询时可以更快地访问，会把数据保存在集群上。此外，还支持持续持久化 RDDs 到磁盘，或复制到多个结点。

基础

• Scala
• Java
• Python

val lines = sc.textFile("data.txt")

val lineLengths = lines.map(s => s.length)

val totalLength = lineLengths.reduce((a, b) => a + b)

lineLengths.persist()

传递 Functions（函数）给 Spark

• Scala
• Java
• Python

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) }

}

class MyClass {

  val field = "Hello"

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

}

def doStuff(rdd: RDD[String]): RDD[String] = {

  val field_ = this.field

  rdd.map(x => field_ + x)

}

理解闭包

在集群中执行代码时，一个关于 Spark 更难的事情是理解变量和方法的范围和生命周期. 修改其范围之外的变量 RDD 操作可以混淆的常见原因。在下面的例子中，我们将看一下使用的 foreach() 代码递增累加计数器，但类似的问题，也可能会出现其他操作上.

示例

考虑一个简单的 RDD 元素求和，以下行为可能不同，具体取决于是否在同一个 JVM 中执行. 一个常见的例子是当 Spark 运行在 local 本地模式（--master = local[n]）时，与部署 Spark 应用到群集（例如，通过 spark-submit 到 YARN）:

• Scala
• Java
• Python

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（集群）模式

上面的代码行为是不确定的，并且可能无法按预期正常工作。执行作业时，Spark 会分解 RDD 操作到每个 executor 中的 task 里。在执行之前，Spark 计算任务的 closure（闭包）。而闭包是在 RDD 上的 executor 必须能够访问的变量和方法（在此情况下的 foreach()）。闭包被序列化并被发送到每个执行器。

闭包的变量副本发给每个 counter ，当 counter 被 foreach 函数引用的时候，它已经不再是 driver node 的 counter 了。虽然在 driver node 仍然有一个 counter 在内存中，但是对 executors 已经不可见。executor 看到的只是序列化的闭包一个副本。所以 counter 最终的值还是 0，因为对 counter 所有的操作均引用序列化的 closure 内的值。

在 local 本地模式，在某些情况下的 foreach 功能实际上是同一 JVM 上的驱动程序中执行，并会引用同一个原始的 counter 计数器，实际上可能更新.

为了确保这些类型的场景明确的行为应该使用的 Accumulator 累加器。当一个执行的任务分配到集群中的各个 worker 结点时，Spark 的累加器是专门提供安全更新变量的机制。本指南的累加器的部分会更详细地讨论这些。

在一般情况下，closures - constructs 像循环或本地定义的方法，不应该被用于改动一些全局状态。Spark 没有规定或保证突变的行为，以从封闭件的外侧引用的对象。一些代码，这可能以本地模式运行，但是这只是偶然和这样的代码如预期在分布式模式下不会表现。如果需要一些全局的聚合功能，应使用 Accumulator（累加器）。

打印 RDD 的 elements

另一种常见的语法用于打印 RDD 的所有元素使用 rdd.foreach(println) 或 rdd.map(println)。在一台机器上，这将产生预期的输出和打印 RDD 的所有元素。然而，在集群 cluster 模式下，stdout 输出正在被执行写操作 executors 的 stdout 代替，而不是在一个驱动程序上，因此 stdout 的 driver 程序不会显示这些！要打印 driver 程序的所有元素，可以使用的 collect() 方法首先把 RDD 放到 driver 程序节点上: rdd.collect().foreach(println)。这可能会导致 driver 程序耗尽内存，虽说，因为 collect() 获取整个 RDD 到一台机器; 如果你只需要打印 RDD 的几个元素，一个更安全的方法是使用 take(): rdd.take(100).foreach(println)。

与 Key-Value Pairs 一起使用

• Scala
• Java
• Python

val lines = sc.textFile("data.txt")

val pairs = lines.map(s => (s, 1))

val counts = pairs.reduceByKey((a, b) => a + b)

Transformations（转换）

下表列出了一些 Spark 常用的 transformations（转换）. 详情请参考 RDD API 文档 (Scala, Java, Python, R) 和 pair RDD 函数文档 (Scala, Java).

Actions（动作）

下表列出了一些 Spark 常用的 actions 操作。详细请参考 RDD API 文档 (Scala, Java, Python, R)

和 pair RDD 函数文档 (Scala, Java).

该 Spark RDD API 还暴露了一些 actions（操作）的异步版本，例如针对 foreach 的 foreachAsync，它们会立即返回一个FutureAction 到调用者，而不是在完成 action 时阻塞。 这可以用于管理或等待 action 的异步执行。.

Shuffle 操作

Spark 里的某些操作会触发 shuffle。shuffle 是spark 重新分配数据的一种机制，使得这些数据可以跨不同的区域进行分组。这通常涉及在 executors 和 机器之间拷贝数据，这使得 shuffle 成为一个复杂的、代价高的操作。

Background（幕后）

为了明白 reduceByKey 操作的过程，我们以 reduceByKey 为例。reduceBykey 操作产生一个新的 RDD，其中 key 所有相同的的值组合成为一个 tuple - key 以及与 key 相关联的所有值在 reduce 函数上的执行结果。面临的挑战是，一个 key 的所有值不一定都在一个同一个 paritition 分区里，甚至是不一定在同一台机器里，但是它们必须共同被计算。

在 spark 里，特定的操作需要数据不跨分区分布。在计算期间，一个任务在一个分区上执行，为了所有数据都在单个 reduceByKey 的 reduce 任务上运行，我们需要执行一个 all-to-all 操作。它必须从所有分区读取所有的 key 和 key对应的所有的值，并且跨分区聚集去计算每个 key 的结果 - 这个过程就叫做 shuffle.。

Although the set of elements in each partition of newly shuffled data will be deterministic, and so is the ordering of partitions themselves, the ordering of these elements is not. If one desires predictably ordered data following shuffle then it’s possible to use:

尽管每个分区新 shuffle 的数据集将是确定的，分区本身的顺序也是这样，但是这些数据的顺序是不确定的。如果希望 shuffle 后的数据是有序的，可以使用:

• mapPartitions 对每个 partition 分区进行排序，例如, .sorted
• repartitionAndSortWithinPartitions 在分区的同时对分区进行高效的排序.
• sortBy 对 RDD 进行全局的排序

触发的 shuffle 操作包括 repartition 操作，如 repartition 和 coalesce, ‘ByKey 操作 (除了 counting 之外) 像 groupByKey 和 reduceByKey, 和 join操作, 像 cogroup 和 join.

性能影响

该 Shuffle 是一个代价比较高的操作，它涉及磁盘 I/O、数据序列化、网络 I/O。为了准备 shuffle 操作的数据，Spark 启动了一系列的任务，map 任务组织数据，reduce 完成数据的聚合。这些术语来自 MapReduce，跟 Spark 的 map 操作和 reduce 操作没有关系。

在内部，一个 map 任务的所有结果数据会保存在内存，直到内存不能全部存储为止。然后，这些数据将基于目标分区进行排序并写入一个单独的文件中。在 reduce 时，任务将读取相关的已排序的数据块。

某些 shuffle 操作会大量消耗堆内存空间，因为 shuffle 操作在数据转换前后，需要在使用内存中的数据结构对数据进行组织。需要特别说明的是，reduceByKey 和 aggregateByKey 在 map 时会创建这些数据结构，'ByKey 操作在 reduce 时创建这些数据结构。当内存满的时候，Spark 会把溢出的数据存到磁盘上，这将导致额外的磁盘 I/O 开销和垃圾回收开销的增加。

shuffle 操作还会在磁盘上生成大量的中间文件。在 Spark 1.3 中，这些文件将会保留至对应的 RDD 不在使用并被垃圾回收为止。这么做的好处是，如果在 Spark 重新计算 RDD 的血统关系（lineage）时，shuffle 操作产生的这些中间文件不需要重新创建。如果 Spark 应用长期保持对 RDD 的引用，或者垃圾回收不频繁，这将导致垃圾回收的周期比较长。这意味着，长期运行 Spark 任务可能会消耗大量的磁盘空间。临时数据存储路径可以通过 SparkContext 中设置参数 spark.local.dir 进行配置。

shuffle 操作的行为可以通过调节多个参数进行设置。详细的说明请看 Spark 配置指南 中的 “Shuffle 行为” 部分。

RDD Persistence（持久化）

Spark 中一个很重要的能力是将数据 persisting 持久化（或称为 caching 缓存），在多个操作间都可以访问这些持久化的数据。当持久化一个 RDD 时，每个节点的其它分区都可以使用 RDD 在内存中进行计算，在该数据上的其他 action 操作将直接使用内存中的数据。这样会让以后的 action 操作计算速度加快（通常运行速度会加速 10 倍）。缓存是迭代算法和快速的交互式使用的重要工具。

RDD 可以使用 persist() 方法或 cache() 方法进行持久化。数据将会在第一次 action 操作时进行计算，并缓存在节点的内存中。Spark 的缓存具有容错机制，如果一个缓存的 RDD 的某个分区丢失了，Spark 将按照原来的计算过程，自动重新计算并进行缓存。

另外，每个持久化的 RDD 可以使用不同的 storage level 存储级别进行缓存，例如，持久化到磁盘、已序列化的 Java 对象形式持久化到内存（可以节省空间）、跨节点间复制、以 off-heap 的方式存储在 Tachyon。这些存储级别通过传递一个 StorageLevel 对象 (Scala, Java, Python) 给 persist() 方法进行设置。cache() 方法是使用默认存储级别的快捷设置方法，默认的存储级别是 StorageLevel.MEMORY_ONLY（将反序列化的对象存储到内存中）。详细的存储级别介绍如下:

Note: 在 Python 中, stored objects will 总是使用 Pickle library 来序列化对象, 所以无论你选择序列化级别都没关系. 在 Python 中可用的存储级别有 MEMORY_ONLY, MEMORY_ONLY_2, MEMORY_AND_DISK, MEMORY_AND_DISK_2, DISK_ONLY, 和 DISK_ONLY_2.

在 shuffle 操作中（例如 reduceByKey），即便是用户没有调用 persist 方法，Spark 也会自动缓存部分中间数据.这么做的目的是，在 shuffle 的过程中某个节点运行失败时，不需要重新计算所有的输入数据。如果用户想多次使用某个 RDD，强烈推荐在该 RDD 上调用 persist 方法.

如何选择存储级别 ?

Spark 的存储级别的选择，核心问题是在 memory 内存使用率和 CPU 效率之间进行权衡。建议按下面的过程进行存储级别的选择:

• 如果您的 RDD 适合于默认存储级别 (MEMORY_ONLY), leave them that way. 这是CPU效率最高的选项，允许RDD上的操作尽可能快地运行.

• 如果不是, 试着使用 MEMORY_ONLY_SER 和 selecting a fast serialization library 以使对象更加节省空间，但仍然能够快速访问。 (Java和Scala)

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

• 如果需要快速故障恢复，请使用复制的存储级别 (e.g. 如果使用Spark来服务 来自网络应用程序的请求). All 存储级别通过重新计算丢失的数据来提供完整的容错能力，但复制的数据可让您继续在 RDD 上运行任务，而无需等待重新计算一个丢失的分区.

删除数据

Spark 会自动监视每个节点上的缓存使用情况，并使用 least-recently-used（LRU）的方式来丢弃旧数据分区。 如果您想手动删除 RDD 而不是等待它掉出缓存，使用 RDD.unpersist() 方法。

共享变量

通常情况下，一个传递给 Spark 操作（例如 map 或 reduce）的函数 func 是在远程的集群节点上执行的。该函数 func 在多个节点执行过程中使用的变量，是同一个变量的多个副本。这些变量的以副本的方式拷贝到每个机器上，并且各个远程机器上变量的更新并不会传播回 driver program（驱动程序）。通用且支持 read-write（读-写） 的共享变量在任务间是不能胜任的。所以，Spark 提供了两种特定类型的共享变量 : broadcast variables（广播变量）和 accumulators（累加器）。

广播变量

Broadcast variables（广播变量）允许程序员将一个 read-only（只读的）变量缓存到每台机器上，而不是给任务传递一个副本。它们是如何来使用呢，例如，广播变量可以用一种高效的方式给每个节点传递一份比较大的 input dataset（输入数据集）副本。在使用广播变量时，Spark 也尝试使用高效广播算法分发 broadcast variables（广播变量）以降低通信成本。

Spark 的 action（动作）操作是通过一系列的 stage（阶段）进行执行的，这些 stage（阶段）是通过分布式的 “shuffle” 操作进行拆分的。Spark 会自动广播出每个 stage（阶段）内任务所需要的公共数据。这种情况下广播的数据使用序列化的形式进行缓存，并在每个任务运行前进行反序列化。这也就意味着，只有在跨越多个 stage（阶段）的多个任务会使用相同的数据，或者在使用反序列化形式的数据特别重要的情况下，使用广播变量会有比较好的效果。

广播变量通过在一个变量 v 上调用 SparkContext.broadcast(v) 方法来进行创建。广播变量是 v 的一个 wrapper（包装器），可以通过调用 value方法来访问它的值。代码示例如下:

• Scala
• Java
• Python

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))

broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

scala> broadcastVar.value

res0: Array[Int] = Array(1, 2, 3)

在创建广播变量之后，在集群上执行的所有的函数中，应该使用该广播变量代替原来的 v 值，所以节点上的 v 最多分发一次。另外，对象 v 在广播后不应该再被修改，以保证分发到所有的节点上的广播变量具有同样的值（例如，如果以后该变量会被运到一个新的节点）。

Accumulators（累加器）

Accumulators（累加器）是一个仅可以执行 “added”（添加）的变量来通过一个关联和交换操作，因此可以高效地执行支持并行。累加器可以用于实现 counter（ 计数，类似在 MapReduce 中那样）或者 sums（求和）。原生 Spark 支持数值型的累加器，并且程序员可以添加新的支持类型。

作为一个用户，您可以创建 accumulators（累加器）并且重命名. 如下图所示, 一个命名的 accumulator 累加器（在这个例子中是 counter）将显示在 web UI 中，用于修改该累加器的阶段。 Spark 在 “Tasks” 任务表中显示由任务修改的每个累加器的值.

在 UI 中跟踪累加器可以有助于了解运行阶段的进度（注: 这在 Python 中尚不支持）.

• Scala
• Java
• Python

scala> val accum = sc.longAccumulator("My Accumulator")

accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0)

scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))

...

10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

scala> accum.value

res2: Long = 10

class VectorAccumulatorV2 extends AccumulatorV2[MyVector, MyVector] {

  private val myVector: MyVector = MyVector.createZeroVector

  def reset(): Unit = {

    myVector.reset()

  }

  def add(v: MyVector): Unit = {

    myVector.add(v)

  }

  ...

}

// Then, create an Accumulator of this type:

val myVectorAcc = new VectorAccumulatorV2

// Then, register it into spark context:

sc.register(myVectorAcc, "MyVectorAcc1")

累加器的更新只发生在 action 操作中，Spark 保证每个任务只更新累加器一次，例如，重启任务不会更新值。在 transformations（转换）中， 用户需要注意的是，如果 task（任务）或 job stages（阶段）重新执行，每个任务的更新操作可能会执行多次。

累加器不会改变 Spark lazy evaluation（懒加载）的模式。如果累加器在 RDD 中的一个操作中进行更新，它们的值仅被更新一次，RDD 被作为 action 的一部分来计算。因此，在一个像 map() 这样的 transformation（转换）时，累加器的更新并没有执行。下面的代码片段证明了这个特性:

• Scala
• Java
• Python

val accum = sc.longAccumulator

data.map { x => accum.add(x); x }

// Here, accum is still 0 because no actions have caused the map operation to be computed.

部署应用到集群中

该 应用提交指南 描述了如何将应用提交到集群中. 简单的说, 在您将应用打包成一个JAR(针对 Java/Scala) 或者一组 .py 或 .zip 文件 (针对Python), 该 bin/spark-submit 脚本可以让你提交它到任何所支持的 cluster manager 上去.

从 Java / Scala 启动 Spark jobs

该 org.apache.spark.launcher package 提供了 classes 用于使用简单的 Java API 来作为一个子进程启动 Spark jobs.

单元测试

Spark 可以友好的使用流行的单元测试框架进行单元测试。在将 master URL 设置为 local 来测试时会简单的创建一个 SparkContext，运行您的操作，然后调用 SparkContext.stop() 将该作业停止。因为 Spark 不支持在同一个程序中并行的运行两个 contexts，所以需要确保使用 finally 块或者测试框架的 tearDown 方法将 context 停止。

快速链接

您可以在 Spark 网站上看一下 Spark 程序示例. 此外, Spark 在 examples 目录中包含了许多示例 (Scala, Java, Python, R). 您可以通过传递 class name 到 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

针对应用程序的优化, 该 配置 和 优化 指南一些最佳实践的信息. 这些优化建议在确保你的数据以高效的格式存储在内存中尤其重要. 针对部署参考, 该 集群模式概述 描述了分布式操作和支持的 cluster managers 集群管理器的组件.

最后,所有的 API 文档可在 Scala, Java, Python and R 中获取.

我们一直在努力

apachecn/spark-doc-zh