Spark编程指南

概述

总体上来说，每个Spark应用都包含一个驱动器（driver）程序，驱动器运行用户的main函数，并在集群上执行各种并行操作。

Spark最重要的一个抽象概念就是弹性分布式数据集（resilient distributed dataset – RDD），RDD是一个可分区的元素集合，其包含的元素可以分布在集群各个节点上，并且可以执行一些分布式并行操作。RDD通常是通过，HDFS（或者其他Hadoop支持的文件系统）上的文件，或者驱动器中的Scala集合对象，来创建或转换得到；其次，用户也可以请求Spark将RDD持久化到内存里，以便在不同的并行操作里复用之；最后，RDD具备容错性，可以从节点失败中自动恢复数据。

Spark第二个重要抽象概念是共享变量，共享变量是一种可以在并行操作之间共享使用的变量。默认情况下，当Spark把一系列任务调度到不同节点上运行时，Spark会同时把每个变量的副本和任务代码一起发送给各个节点。但有时候，我们需要在任务之间，或者任务和驱动器之间共享一些变量。Spark提供了两种类型的共享变量：广播变量和累加器，广播变量可以用来在各个节点上缓存数据，而累加器则是用来执行跨节点的“累加”操作，例如：计数和求和。

本文将会使用Spark所支持的所有语言来展示Spark的特性。如果你能启动Spark的交互式shell动手实验一下，效果会更好（对scala请使用bin/spark-shell，而对于python，请使用bin/pyspark）。

链接Spark

• Scala
• Java
• Python

groupId = org.apache.spark
artifactId = spark-core_2.10
version = 1.6.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 or YARN cluster URL 或者特殊的“local”（本地模式）。实际上，一般不应该将master参数值硬编码到代码中，而是应该用spark-submit脚本的参数来设置。然而，如果是本地测试或单元测试中，你可以直接在代码里给master参数写死一个”local”值。

使用shell

• Scala
• Python

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

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

弹性分布式数据集(RDD)

Spark的核心概念是弹性分布式数据集（RDD），RDD是一个可容错、可并行操作的分布式元素集合。总体上有两种方法可以创建RDD对象：由驱动程序中的集合对象通过并行化操作创建，或者从外部存储系统中数据集加载（如：共享文件系统、HDFS、HBase或者其他Hadoop支持的数据源）。

并行化集合

• Scala
• Java
• Python

val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)

并行化集合的一个重要参数是分区（partition），即这个分布式数据集可以分割为多少片。Spark中每个任务（task）都是基于分区的，每个分区一个对应的任务（task）。典型场景下，一般每个CPU对应2~4个分区。并且一般而言，Spark会基于集群的情况，自动设置这个分区数。当然，你还是可以手动控制这个分区数，只需给parallelize方法再传一个参数即可（如：sc.parallelize(data, 10) ）。注意：Spark代码里有些地方仍然使用分片（slice）这个术语，这只不过是分区的一个别名，主要为了保持向后兼容。

外部数据集

• Scala
• Java
• Python

scala> val distFile = sc.textFile("data.txt")
distFile: RDD[String] = MappedRDD@1d4cee08

RDD算子

RDD支持两种类型的算子（operation）：transformation算子 和 action算子；transformation算子可以将已有RDD转换得到一个新的RDD，而action算子则是基于数据集计算，并将结果返回给驱动器（driver）。例如，map是一个transformation算子，它将数据集中每个元素传给一个指定的函数，并将该函数返回结果构建为一个新的RDD；而 reduce是一个action算子，它可以将RDD中所有元素传给指定的聚合函数，并将最终的聚合结果返回给驱动器（还有一个reduceByKey算子，其返回的聚合结果是一个数据集）。

Spark中所有transformation算子都是懒惰的，也就是说，这些算子并不立即计算结果，而是记录下对基础数据集（如：一个数据文件）的转换操作。只有等到某个action算子需要计算一个结果返回给驱动器的时候，transformation算子所记录的操作才会被计算。这种设计使Spark可以运行得更加高效 – 例如，map算子创建了一个数据集，同时该数据集下一步会调用reduce算子，那么Spark将只会返回reduce的最终聚合结果（单独的一个数据）给驱动器，而不是将map所产生的数据集整个返回给驱动器。

默认情况下，每次调用action算子的时候，每个由transformation转换得到的RDD都会被重新计算。然而，你也可以通过调用persist（或者cache）操作来持久化一个RDD，这意味着Spark将会把RDD的元素都保存在集群中，因此下一次访问这些元素的速度将大大提高。同时，Spark还支持将RDD元素持久化到内存或者磁盘上，甚至可以支持跨节点多副本。

基础

• 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()

将函数传给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里一个比较难的事情就是，理解在整个集群上跨节点执行的变量和方法的作用域以及生命周期。Spark里一个频繁出现的问题就是RDD算子在变量作用域之外修改了其值。下面的例子，我们将会以foreach() 算子为例，来递增一个计数器counter，不过类似的问题在其他算子上也会出现。

示例

考虑如下例子，我们将会计算RDD中原生元素的总和，如果不是在同一个JVM中执行，其表现将有很大不同。例如，这段代码如果使用Spark本地模式（–master=local[n]）运行，和在集群上运行（例如，用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)

本地模式 v.s. 集群模式

上面这段代码其行为是不确定的。在本地模式下运行，所有代码都在运行于单个JVM中，所以RDD的元素都能够被累加并保存到counter变量中，这是因为本地模式下，counter变量和驱动器节点在同一个内存空间中。

然而，在集群模式下，情况会更复杂，以上代码的运行结果就不是所预期的结果了。为了执行这个作业，Spark会将RDD算子的计算过程分割成多个独立的任务（task）- 每个任务分发给不同的执行器（executor）去执行。而执行之前，Spark需要计算闭包。闭包是由执行器执行RDD算子（本例中的foreach()）时所需要的变量和方法组成的。闭包将会被序列化，并发送给每个执行器。由于本地模式下，只有一个执行器，所有任务都共享同样的闭包。而在其他模式下，情况则有所不同，每个执行器都运行于不同的worker节点，并且都拥有独立的闭包副本。

在上面的例子中，闭包中的变量会跟随不同的闭包副本，发送到不同的执行器上，所以等到foreach真正在执行器上运行时，其引用的counter已经不再是驱动器上所定义的那个counter副本了，驱动器内存中仍然会有一个counter变量副本，但是这个副本对执行器是不可见的！执行器只能看到其所收到的序列化闭包中包含的counter副本。因此，最终驱动器上得到的counter将会是0。

为了确保类似这样的场景下，代码能有确定的行为，这里应该使用累加器（Accumulator）。累加器是Spark中专门用于集群跨节点分布式执行计算中，安全地更新同一变量的机制。本指南中专门有一节详细说明累加器。

通常来说，闭包（由循环或本地方法组成），不应该改写全局状态。Spark中改写闭包之外对象的行为是未定义的。这种代码，有可能在本地模式下能正常工作，但这只是偶然情况，同样的代码在分布式模式下其行为很可能不是你想要的。所以，如果需要全局聚合，请记得使用累加器（Accumulator）。

打印RDD中的元素

另一种常见习惯是，试图用 rdd.foreach(println) 或者 rdd.map(println) 来打印RDD中所有的元素。如果是在单机上，这种写法能够如预期一样，打印出RDD所有元素。然后，在集群模式下，这些输出将会被打印到执行器的标准输出（stdout）上，因此驱动器的标准输出（stdout）上神马也看不到！如果真要在驱动器上把所有RDD元素都打印出来，你可以先调用collect算子，把RDD元素先拉倒驱动器上来，代码可能是这样：rdd.collect().foreach(println)。不过如果RDD很大的话，有可能导致驱动器内存溢出，因为collect会把整个RDD都弄到驱动器所在单机上来；如果你只是需要打印一部分元素，那么take是不更安全的选择：rdd.take(100).foreach(println)

使用键值对

• Scala
• Java
• Python

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

转换算子 – transformation

以下是Spark支持的一些常用transformation算子。详细请参考 RDD API doc (Scala, Java, Python, R) 以及 键值对 RDD 函数 (Scala, Java) 。

动作算子 – action

以下是Spark支持的一些常用action算子。详细请参考 RDD API doc (Scala, Java, Python, R) 以及 键值对 RDD 函数 (Scala, Java) 。

混洗操作

有一些Spark算子会触发众所周知的混洗（Shuffle）事件。Spark中的混洗机制是用于将数据重新分布，其结果是所有数据将在各个分区间重新分组。一般情况下，混洗需要跨执行器（executor）或跨机器复制数据，这也是混洗操作一般都比较复杂而且开销大的原因。

背景

为了理解混洗阶段都发生了哪些事，我首先以 reduceByKey 算子为例来看一下。reduceByKey算子会生成一个新的RDD，将源RDD中一个key对应的多个value组合进一个tuple - 然后将这些values输入给reduce函数，得到的result再和key关联放入新的RDD中。这个算子的难点在于对于某一个key来说，并非其对应的所有values都在同一个分区（partition）中，甚至有可能都不在同一台机器上，但是这些values又必须放到一起计算reduce结果。

在Spark中，通常是由于为了进行某种计算操作，而将数据分布到所需要的各个分区当中。而在计算阶段，单个任务（task）只会操作单个分区中的数据 – 因此，为了组织好每个reduceByKey中reduce任务执行时所需的数据，Spark需要执行一个多对多操作。即，Spark需要读取RDD的所有分区，并找到所有key对应的所有values，然后跨分区传输这些values，并将每个key对应的所有values放到同一分区，以便后续计算各个key对应values的reduce结果 – 这个过程就叫做混洗（Shuffle）。

虽然混洗好后，各个分区中的元素和分区自身的顺序都是确定的，但是分区中元素的顺序并非确定的。如果需要混洗后分区内的元素有序，可以参考使用以下混洗操作：

• mapPartitions 使用 .sorted 对每个分区排序 
• repartitionAndSortWithinPartitions 重分区的同时，对分区进行排序，比自行组合repartition和sort更高效
• sortBy 创建一个全局有序的RDD

会导致混洗的算子有：重分区（repartition）类算子，如： repartition 和 coalesce；ByKey 类算子(除了计数类的，如 countByKey) 如：groupByKey 和 reduceByKey；以及Join类算子，如：cogroup 和 join.

性能影响

混洗（Shuffle）之所以开销大，是因为混洗操作需要引入磁盘I/O，数据序列化以及网络I/O等操作。为了组织好混洗数据，Spark需要生成对应的任务集 – 一系列map任务用于组织数据，再用一系列reduce任务来聚合数据。注意这里的map、reduce是来自MapReduce的术语，和Spark的map、reduce算子并没有直接关系。

在Spark内部，单个map任务的输出会尽量保存在内存中，直至放不下为止。然后，这些输出会基于目标分区重新排序，并写到一个文件里。在reduce端，reduce任务只读取与之相关的并已经排序好的blocks。

某些混洗算子会导致非常明显的内存开销增长，因为这些算子需要在数据传输前后，在内存中维护组织数据记录的各种数据结构。特别地，reduceByKey和aggregateByKey都会在map端创建这些数据结构，而ByKey系列算子都会在reduce端创建这些数据结构。如果数据在内存中存不下，Spark会把数据吐到磁盘上，当然这回导致额外的磁盘I/O以及垃圾回收的开销。

混洗还会再磁盘上生成很多临时文件。以Spark-1.3来说，这些临时文件会一直保留到其对应的RDD被垃圾回收才删除。之所以这样做，是因为如果血统信息需要重新计算的时候，这些混洗文件可以不必重新生成。如果程序持续引用这些RDD或者垃圾回收启动频率较低，那么这些垃圾回收可能需要等较长的一段时间。这就意味着，长时间运行的Spark作业可能会消耗大量的磁盘。Spark的临时存储目录，是由spark.local.dir 配置参数指定的。

混洗行为可以由一系列配置参数来调优。参考Spark配置指南（Spark Configuration Guide）中“混洗行为”这一小节。

RDD持久化

Spark的一项关键能力就是它可以持久化（或者缓存）数据集在内存中，从而跨操作复用这些数据集。如果你持久化了一个RDD，那么每个节点上都会存储该RDD的一些分区，这些分区是由对应的节点计算出来并保持在内存中，后续可以在其他施加在该RDD上的action算子中复用（或者从这些数据集派生新的RDD）。这使得后续动作的速度提高很多（通常高于10倍）。因此，缓存对于迭代算法和快速交互式分析是一个很关键的工具。

你可以用persist() 或者 cache() 来标记一下需要持久化的RDD。等到该RDD首次被施加action算子的时候，其对应的数据分区就会被保留在内存里。同时，Spark的缓存具备一定的容错性 – 如果RDD的任何一个分区丢失了，Spark将自动根据其原来的血统信息重新计算这个分区。

另外，每个持久化的RDD可以使用不同的存储级别，比如，你可以把RDD保存在磁盘上，或者以java序列化对象保存到内存里（为了省空间），或者跨节点多副本，或者使用 Tachyon 存到虚拟机以外的内存里。这些存储级别都可以由persist()的参数StorageLevel对象来控制。cache() 方法本身就是一个使用默认存储级别做持久化的快捷方式，默认存储级别是 StorageLevel.MEMORY_ONLY（以java序列化方式存到内存里）。完整的存储级别列表如下：

注意：在Python中存储的对象总是会使用 Pickle 做序列化，所以这时是否选择一个序列化级别已经无关紧要了。

Spark会自动持久化一些混洗操作（如：reduceByKey）的中间数据，即便用户根本没有调用persist。这么做是为了避免一旦有一个节点在混洗过程中失败，就要重算整个输入数据。当然，我们还是建议对需要重复使用的RDD调用其persist算子。

如何选择存储级别？

Spark的存储级别主要可于在内存使用和CPU占用之间做一些权衡。建议根据以下步骤来选择一个合适的存储级别：

• 如果RDD能使用默认存储级别（MEMORY_ONLY），那就尽量使用默认级别。这是CPU效率最高的方式，所有RDD算子都能以最快的速度运行。
• 如果步骤1的答案是否（不适用默认级别），那么可以尝试MEMORY_ONLY_SER级别，并选择一个高效的序列化协议（selecting a fast serialization library），这回大大节省数据对象的存储空间，同时速度也还不错。
• 尽量不要把数据吐到磁盘上，除非：1.你的数据集重新计算的代价很大；2.你的数据集是从一个很大的数据源中过滤得到的结果。否则的话，重算一个分区的速度很可能和从磁盘上读取差不多。
• 如果需要支持容错，可以考虑使用带副本的存储级别（例如：用Spark来服务web请求）。所有的存储级别都能够以重算丢失数据的方式来提供容错性，但是带副本的存储级别可以让你的应用持续的运行，而不必等待重算丢失的分区。
• 在一些需要大量内存或者并行多个应用的场景下，实验性的OFF_HEAP会有以下几个优势：
  • 这个级别下，可以允许多个执行器共享同一个Tachyon中内存池。
  • 可以有效地减少垃圾回收的开销。
  • 即使单个执行器挂了，缓存数据也不会丢失。

删除数据

Spark能够自动监控各个节点上缓存使用率，并且以LRU（最近经常使用）的方式将老数据逐出内存。如果你更喜欢手动控制的话，可以用RDD.unpersist() 方法来删除无用的缓存。

共享变量

一般而言，当我们给Spark算子（如 map 或 reduce）传递一个函数时，这些函数将会在远程的集群节点上运行，并且这些函数所引用的变量都是各个节点上的独立副本。这些变量都会以副本的形式复制到各个机器节点上，如果更新这些变量副本的话，这些更新并不会传回到驱动器（driver）程序。通常来说，支持跨任务的可读写共享变量是比较低效的。不过，Spark还是提供了两种比较通用的共享变量：广播变量和累加器。

广播变量

广播变量提供了一种只读的共享变量，它是把在每个机器节点上保存一个缓存，而不是每个任务保存一份副本。通常可以用来在每个节点上保存一个较大的输入数据集，这要比常规的变量副本更高效（一般的变量是每个任务一个副本，一个节点上可能有多个任务）。Spark还会尝试使用高效的广播算法来分发广播变量，以减少通信开销。

Spark的操作有时会有多个阶段（stage），不同阶段之间的分割线就是混洗操作。Spark会自动广播各个阶段用到的公共数据。这些方式广播的数据都是序列化过的，并且在运行各个任务前需要反序列化。这也意味着，显示地创建广播变量，只有在跨多个阶段（stage）的任务需要同样的数据 或者 缓存数据的序列化和反序列化格式很重要的情况下 才是必须的。

广播变量可以通过一个变量v来创建，只需调用 SparkContext.broadcast(v)即可。这个广播变量是对变量v的一个包装，要访问其值，可以调用广播变量的 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在广播后不应该再被更新，这样才能保证所有节点上拿到同样的值（例如，更新后，广播变量又被同步到另一新节点，新节点有可能得到的值和其他节点不一样）。

累加器

累加器是一种只支持满足结合律的“累加”操作的变量，因此它可以很高效地支持并行计算。利用累加器可以实现计数（类似MapReduce中的计数器）或者求和。Spark原生支持了数字类型的累加器，开发者也可以自定义新的累加器。如果创建累加器的时候给了一个名字，那么这个名字会展示在Spark UI上，这对于了解程序运行处于哪个阶段非常有帮助（注意：Python尚不支持该功能）。

创捷累加器时需要赋一个初始值v，调用 SparkContext.accumulator(v) 可以创建一个累加器。后续集群中运行的任务可以使用 add 方法 或者 += 操作符 （仅Scala和Python支持）来进行累加操作。不过，任务本身并不能读取累加器的值，只有驱动器程序可以用 value 方法访问累加器的值。

以下代码展示了如何使用累加器对一个元素数组求和：

• Scala
• Java
• Python

scala> val accum = sc.accumulator(0, "My Accumulator")
accum: spark.Accumulator[Int] = 0

scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

scala> accum.value
res2: Int = 10

object VectorAccumulatorParam extends AccumulatorParam[Vector] {
  def zero(initialValue: Vector): Vector = {
    Vector.zeros(initialValue.size)
  }
  def addInPlace(v1: Vector, v2: Vector): Vector = {
    v1 += v2
  }
}

// Then, create an Accumulator of this type:
val vecAccum = sc.accumulator(new Vector(...))(VectorAccumulatorParam)

对于在action算子中更新的累加器，Spark保证每个任务对累加器的更新只会被应用一次，例如，某些任务如果重启过，则不会再次更新累加器。而如果在transformation算子中更新累加器，那么用户需要注意，一旦某个任务因为失败被重新执行，那么其对累加器的更新可能会实施多次。

累加器并不会改变Spark懒惰求值的运算模型。如果在RDD算子中更新累加器，那么其值只会在RDD做action算子计算的时候被更新一次。因此，在transformation算子（如：map）中更新累加器，其值并不能保证一定被更新。以下代码片段说明了这一特性：

• Scala
• Java
• Python

val accum = sc.accumulator(0)
data.map { x => accum += x; f(x) }
// 这里，accum任然是0，因为没有action算子，所以map也不会进行实际的计算

部署到集群

应用提交指南（application submission guide）中描述了如何向集群提交应用。换句话说，就是你需要把你的应用打包成 JAR文件（Java/Scala）或者一系列 .py 或 .zip 文件（Python），然后再用 bin/spark-submit 脚本将其提交给Spark所支持的集群管理器。

从Java/Scala中启动Spark作业

org.apache.spark.launcher 包提供了简明的Java API，可以将Spark作业作为子进程启动。

单元测试

Spark对所有常见的单元测试框架提供友好的支持。你只需要在测试中创建一个SparkContext对象，然后吧master URL设为local，运行测试操作，最后调用 SparkContext.stop() 来停止测试。注意，一定要在 finally 代码块或者单元测试框架的 tearDown方法里调用SparkContext.stop()，因为Spark不支持同一程序中有多个SparkContext对象同时运行。

从1.0之前版本迁移过来

• Scala
• Java
• Python

更详细迁移向导请参考这里：Spark Streaming, MLlib 以及 GraphX.

下一步

你可以去Spark的官网上看看示例程序（example Spark programs）。另外，Spark代码目录下也自带了不少例子，见 examples 目录(Scala,Java, Python, R)。你可以把示例中的类名传给 bin/run-example 脚本来运行这些例子；例如：

./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

配置（configuration）和调优（tuning）指南提供了不少最佳实践的信息，可以帮助你优化程序，特别是这些信息可以帮助你确保数据以一种高效的格式保存在内存里。集群模式概览（cluster mode overview）这篇文章描述了分布式操作中相关的组件，以及Spark所支持的各种集群管理器。

最后，完整的API文件见：Scala, Java, Python 以及 R.