Apache Spark - Programming Guide(Spark编程指南)

Apache Spark 编程指南

Overview
Linking with Spark
Initializing Spark
Master URLs
Deploying Code on a Cluster
Resilient Distributed Datasets (RDDs)
Parallelized Collections
Hadoop Datasets
RDD Operations
Transformations
Actions
RDD Persistence
Which Storage Level to Choose?
Shared Variables
Broadcast Variables
Accumulators
Where to Go from Here

概览

在较高的抽象级别上看, 每个Spark应用程序都是由驱动程序(driver program)一个构成,它负责运行用户main函数和在集群上执行各种并行操作.Spark提出的最主要的抽象概念是RDD(resilient distributed dataset ,弹性分布式数据集), 它是一个被划分到集群各个节点上的元素集合,可以进行并行操作. RDDs可以由HDFS(或者其他支持Hadoop的文件系统)上的一个文件创建, 或是从驱动程序中的一个现有Scala集合转换而来. 用户可以让Spark在内存中持久化RDD,
使其在并行操作中有效复用. 最后, RDDs能从节点故障中自动恢复.

Spark中的第二个抽象概念是可以在并行操作中使用的共享变量(shared variables). 默认情况下, 当Spark在不同节点上作为任务集运行一个并行函数的时候,它会将函数中使用到的变量复制到每个任务中. 有时, 一个变量需要在任务间或是任务与驱动程序间共享. Spark支持两种类型的共享变量: 广播变量(broadcast variables)和累加器( accumulators).
广播变量可以在内存所有节点上缓存变量,而累加器只能用于作诸如计数求和之类的加法变量.

本指南将展示这些特性并执行一些样例辅助说明. It assumes some familiarity with Scala, 尤其是闭包closures的语法. 你也可以通过

bin/spark-shell

脚本交互式的运行Spark.
我们强烈建议你在接下来的步骤中这样做!

连接Spark

Spark 0.9.0-incubating 需要Scala 2.10环境. 如果你用Scala编写应用, 你需要使用匹配的Scala版本(e.g. 2.10.X) – 更新的大版本可能不兼容.

要写一个Spark应用, 首先需要引入Spark依赖. 如果你使用SBT或Maven, Spark可以通过Maven Central获得:

groupId = org.apache.spark
artifactId = spark-core_2.10
version = 0.9.0-incubating

更进一步的话, 如果你希望存储HDFS集群, 需要添加合适版本的

hadoop-client

的依赖来引入HDFS:

groupId = org.apache.hadoop
artifactId = hadoop-client
version = <your-hdfs-version>

在其他构建系统中, 可以使用

sbt/sbt assembly

把Spark和他的依赖包一起打包 (

assembly/target/scala-2.10/spark-assembly-0.9.0-incubating-hadoop*.jar

),
然后加入CLASSPATH. 并设置HDFS的版本作为描述(here).

最后, 需要在程序中引入一些Spark类和隐式转换.加入如下行:

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

Spark初始化

Spark必须做的第一件事是创建一个

SparkContext

对象, 它将告诉你Spark如何访问一个集群. 通常使用如下构造器:

new SparkContext(master, appName, [sparkHome], [jars])

或者通过

new SparkContext(conf)

定义,它使用了 SparkConf 对象进行高级配置.

master

参数是一个用于指定连接到的 Spark
or Mesos cluster URL 的字符串, 或是如下描述的一个用于在local模式下运行的特殊“local”字符串.

appName

是应用的名称, 将会在集群的web UI中显示. 如果部署到集群,在分布式模式下运行的话,最后两个参数是必须的, 稍后会详述.

在Spark shell中, 一个特殊的解释器感知的SparkContext已经为你创建好了, 变量名叫

sc

. 这同时使得自定义的SparkContext无效. 你可以使用master环境变量设置sc连接到master, 也可以用

ADD_JARS

变量把JARs加入到classpath.
例如, 在四核CPU上使用

bin/spark-shell

：

$ MASTER=local[4] ./bin/spark-shell

或者加入

code.jar

到classpath, 如下:

$ MASTER=local[4] ADD_JARS=code.jar ./bin/spark-shell

Master URLs

传递给Spark的master URL可以是下列任意一种格式:

Master URL	Meaning含义
local	使用一个工作线程本地化运行Spark.(完全不并行)
local[K]	使用K个工作线程本地化运行Spark.(比较理想的是,K值根据运行机器的CPU核数设定)
spark://HOST:PORT	连接到指定的Spark单机版集群( Spark standalone cluster )master. 必须使用master配置的端口, 默认是 7077 .
mesos://HOST:PORT	连接到指定的Mesos 集群. host参数是Mesos集群的hostname. 必须使用master配置的端口, 默认是 5050 .

如果没有指定 master URL , spark shell 将会默认设置为 “local”.

如果运行于 YARN, Spark 将会在YARN上启动一个standalone部署的集群实例; 详情参见 running on YARN .

在集群上部署代码

如果你要在集群上运行应用，你需要给SparkContext指定两个可选参数，使其能找到你的代码：

sparkHome

: 你的集群机器上Spark的安装路径（所有机器上路径必须一致） .

jars

: 在本地机器上的JAR文件列表,其中包括你应用的代码以及任何的依赖,Spark将会把他们部署到所有的集群结点上.你需要使用你的编译系统将你的应用打包成一系列JAR文件.例如,如果你使用SBT,用 sbt-assembly 插件将你的代码和所有依赖变成一个JAR文件是一个好的办法.

如果你在集群上运行

bin/spark-shell

, 在启动之前你可以通过指定ADD_JAR环境变量将JAR文件们加载在集群上，这个变量需要包括一个用逗号分隔的JAR文件列表. 例如,

ADD_JARS=a.jar,b.jar
./bin/spark-shell

将启动一个在 classpath中带有

a.jar

和

b.jar

的脚本.
此外, 任何在shell脚本中定义的新类都会被自动分发.

Resilient Distributed Datasets (弹性分布式数据集)

Spark的核心概念是弹性分布式数据集(resilient distributed dataset,RDD), 这是一个有容错机制并可以被并行操作的元素集合. 目前有两种类型的RDDs: 并行集合(parallelized collections)和Hadoop数据集合(Hadoop datasets).并行集合可以接收一个已有的Scala 然后进行各种并行运算.Hadoop数据集合可以在一个文件的每条记录上运行函数(只要文件系统是HDFS或Hadoop支持的任意存储系统).
这两种类型的RDD都可以通过相同的方式进行操作.

并行集合

通过调用

SparkContext

的

parallelize

方法,可以将已存在的Scala集合(一个Seq对象)
转化为并行集合. 集合的对象将会被拷贝,创建出一个可以被并行操作的分布式数据集.例如,下面的解释器输出,演示了如何从一个数组创建一个并行集合:

scala> val data = Array(1, 2, 3, 4, 5)
data: Array[Int] = Array(1, 2, 3, 4, 5)

scala> val distData = sc.parallelize(data)
distData: spark.RDD[Int] = spark.ParallelCollection@10d13e3e

一旦创建, 分布式数据集 (

distData

) 就可以执行并行操作了. 例如, 我们可以调用

distData.reduce(_
+ _)

来迭加数组中的元素. 我们会在后续的分布式数据集运算中进一步描述.

并行集合的一个重要参数是slices, 表示数据集切分的份数. Spark将会在集群上为每一份数据起一个任务. 典型情况是集群的每个CPU上分布 2-4个slices. 一般来说，Spark会尝试根据集群的状况，来自动设定slices的数目. 当然,你也可以通过传递给parallelize的第二个参数来进行手动设置

parallelize

(例如

sc.parallelize(data,
10)

).

Hadoop数据集

Spark可以从存储在HDFS或Hadoop支持的其它文件系统(包括本地文件, Amazon S3, Hypertable, HBase等)上的文件创建分布式数据集. Spark可以支持TextFile, SequenceFiles以及其它任何Hadoop输入格式.

Text file 的RDDs 可以通过

SparkContext

的

textFile

方法构建.
此方法接收一个文件的 URI 地址(或机器上的本地路径, 或一个

hdfs://

,

s3n://

,

kfs://

,
其它 URI). 如下例所示:

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

一旦创建,

distFile

可以进行数据集操作. 例如, 我们可以使用如下的

map

和

reduce

操作
:

distFile.map(_.size).reduce(_ + _)

.

textFile

方法也可以通过输入一个可选的第二参数来控制文件的分片数目. 默认情况下, Spark为每一块文件创建一个分片（HDFS默认的块大小为64MB)，但是你也可以通过传入一个更大的值，来指定一个更高的片值。注意，你不能指定一个比块数更小的片值.

对于 SequenceFiles, 使用 SparkContext 的

sequenceFile[K,
V]

方法创建, 其中

K

和

V

是文件中的key和values的类型.
它们应该是Hadoop Writable 接口的子类, 就像 IntWritable 和 Text一样.
此外, Spark允许你特定的原生类型替代通用的Writables类型; 例如,

sequenceFile[Int, String]

将会自动读取 IntWritables 和 Texts.

最后,对其他的 Hadoop 输入格式, 可以使用

SparkContext.hadoopRDD

方法, 它可以接收任意类型的JobConf和输入格式类，键类型和值类型。按照像设置Hadoop作业一样来设置输入源就可以了.

RDD操作

RDDs 支持两种类型的操作:transformations( 转换, 从现有数据集创建一个新的数据集), actions(操作, 在数据及上运行计算后返回一个值给驱动程序). 例如,

map

是一种转换, 它将数据集每个元素都传递给函数,
并返回一个新的分布数据集表示结果. 另一方面,

reduce

是一个操作，它通过一些函数迭加所有元素并将最终结果返回给驱动程序. (还有一个并行的

reduceByKey

可以返回分布式数据集).

Spark中的所有转换都是惰性的, 也就是说不会直接计算结果(...函数式编程特性). 取而代之, 它们记住的是对数据集(例如一个文件)的转换动作. 只有驱动程序要求结果返回时, 这些转换才会发生. 这种设计可以让Spark更有效的运行 – 例如, 我们可以实现通过

map

创建数据集并在

reduce

中使用,
返回

reduce

的结果给驱动, 而不是整个大数据集.

默认情况下, 每个转换后的 RDD 都会在执行一个操作时被重新计算. 不过你也可以使用

persist

(或者

cache

)
方法在内存中持久化一个RDD, 这种情况下Spark将会在集群中保存相关元素以备下次查询. 在磁盘上持久化数据集或在集群之间复制数据集都可以, 这些选项将在本文档的下节进行描述.

如下表格列出了目前支持的转换和操作 (详情参见 RDD API doc ):

Transformations (转换)

转换	含义
map(func)	返回一个新的分布式数据集，由每一个输入元素经过func函数转换后组成
filter(func)	返回一个新的分布式数据集，由经过func函数计算后返回值为true的输入元素组成
flatMap(func)	类似于map，但是每一个输入元素可以被映射为0或多个输出元素（因此func应该返回一个序列，而不是单一元素）
mapPartitions(func)	类似于map，但独立地在RDD的每一个分块上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U].
mapPartitionsWithIndex(func)	类似于mapPartitions, 但func带有一个整数参数表示分块的索引值。因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Iterator[T]) => Iterator[U].
sample(withReplacement,fraction,seed)	根据fraction指定的比例，对数据进行采样，可以选择是否用随机数进行替换，seed用于指定随机数生成器种子.
union(otherDataset)	返回一个新的数据集，新数据集是由源数据集和参数数据集联合而成.
distinct([numTasks]))	返回一个包含源数据集中所有不重复元素的新数据集.
groupByKey([numTasks])	在一个（K,V）对的数据集上调用，返回一个（K，Seq[V])对的数据集. 注意：默认情况下，只有8个并行任务来做操作，但是你可以传入一个可选的numTasks参数来改变它.
reduceByKey(func, [numTasks])	在一个（K，V)对的数据集上调用时，返回一个（K，V）对的数据集，使用指定的reduce函数，将相同key的值聚合到一起。类似groupByKey，reduce任务个数是可以通过第二个可选参数来配置的.
sortByKey([ascending], [numTasks])	在一个（K，V)对的数据集上调用，K必须实现Ordered接口，返回一个按照Key进行排序的（K，V）对数据集。升序或降序由ascending布尔参数决定.
join(otherDataset, [numTasks])	在类型为（K,V)和（K,W)类型的数据集上调用时，返回一个相同key对应的所有元素对在一起的(K, (V, W))数据集.
cogroup(otherDataset, [numTasks])	在类型为（K,V)和（K,W)的数据集上调用，返回一个 (K, Seq[V], Seq[W])元组的数据集。这个操作也可以称之为groupwith.
cartesian(otherDataset)	笛卡尔积，在类型为 T 和 U 类型的数据集上调用时，返回一个 (T, U)对数据集(两两的元素对).

完整的转换列表详见 RDD API doc.

Actions (操作)

操作	含义
reduce(func)	通过函数func（接受两个参数，返回一个参数）聚集数据集中的所有元素。这个功能必须可交换且可关联的，从而可以正确的被并行执行.
collect()	在驱动程序中，以数组的形式，返回数据集的所有元素。这通常会在使用filter或者其它操作并返回一个足够小的数据子集后再使用会比较有用.
count()	返回数据集的元素的个数.
first()	返回数据集的第一个元素(l类似于 take(1)).
take(n)	返回一个由数据集的前n个元素组成的数组. 注意：这个操作目前并非并行执行, 而是由驱动程序计算所有的元素.
takeSample(withReplacement,num, seed)	返回一个数组，在数据集中随机采样num个元素组成，可以选择是否用随机数替换不足的部分，Seed用于指定的随机数生成器种子.
saveAsTextFile(path)	将数据集的元素，以textfile的形式，保存到本地文件系统，HDFS或者任何其它hadoop支持的文件系统。对于每个元素，Spark将会调用toString()方法，将它转换为文件中的文本行.
saveAsSequenceFile(path)	将数据集的元素，以Hadoop sequencefile的格式，保存到指定的目录下，本地系统，HDFS或者任何其它hadoop支持的文件系统。这个只限于由key-value对组成，并实现了Hadoop的Writable接口，或者隐式的可以转换为Writable的RDD。Spark包括了基本类型的转换，例如Int，Double，String等.
countByKey()	对(K,V)类型的RDD有效，返回一个(K，Int)对的Map，表示每一个key对应的元素个数.
foreach(func)	在数据集的每一个元素上，运行函数func进行更新。这通常用于边缘效果，例如更新一个累加器，或者和外部存储系统进行交互，例如HBase.

完整的操作列表详见 RDD API doc.

RDD持久化

Spark最重要的一个功能，就是在不同操作间，持久化或缓存一个数据集在内存中。当你持久化一个RDD，每一个结点都将把它的计算分块结果保存在内存中，并在对此数据集(或者衍生出的数据集)进行的其它操作中重用。这将使得后续操作变得更加迅速(通常快10倍).
缓存是用Spark构建迭代算法的关键.

你可以用persist()或cache()方法来标记一个要被持久化的RDD，然后一旦首次被一个动作（Action）触发计算，它将会被保留在计算结点的内存中并重用. Cache有容错机制，如果RDD的任一分区丢失了，通过使用原先创建它的转换操作，它将会被自动重算（不需要全部重算，只计算丢失的部分）.

此外, 每一个RDD都可以用不同的保存级别进行保存, 从而允许你持久化数据集在硬盘, 或者在内存作为序列化的Java对象(节省空间), 甚至于跨结点复制. 通过传递一个

org.apache.spark.storage.StorageLevel

对象到

persist()

可以实现持久化级别的选择

,

cache()

方法是使用默认存储级别的快捷方法,
也就是

StorageLevel.MEMORY_ONLY

(将反序列化对象存入内存). 完整的可选存储级别如下:

存储级别	含义
MEMORY_ONLY	把RDD 作为Java反序列化对象存储在JVM中. If the RDD does not fit in memory, some partitions will not be cached and will be recomputed on the fly each time they're needed. This is the default level.
MEMORY_AND_DISK	把RDD 作为Java反序列化对象存储在JVM中. 如果RDD不能被与内存装下, 超出的分区将被保存在硬盘上, 并且在需要时被读取 .
MEMORY_ONLY_SER	把RDD 作为序列化的的对象进行存储(每一分区占用一个字节数组). 通常来说, 这比将对象反序列化的空间利用率更高(尤其是 fast serializer), 但是读取时CPU占用率较高.
MEMORY_AND_DISK_SER	同 MEMORY_ONLY_SER相似, 但是把超出内存的分区将存储在硬盘上而不是每次需要的时候重新计算.
DISK_ONLY	只将RDD 分区存储在硬盘上.
MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc.	与上述存储级别一样, 但是将每个分区都复制到两个集群节点上.

如何选择存储级别？

Spark的不同存储级别旨在满足内存使用和CPU效率权衡上的不同需求. 我们建议通过如下步骤选择:

如果你的RDDs 能够很好的适应默认的存储级别(

MEMORY_ONLY

), 请离开这里. 这已经是CPU效率最高的选项,它使得RDDs的操作尽可能的快.
如果不行, 试着使用

MEMORY_ONLY_SER

和选择一个快速序列化的库( selecting
a fast serialization library) 使得对象在有比较高的空间实用率的情况下, 依然可以较快访问.
尽可能不要存储到硬盘上. 除非计算数据集的函数计算量特别大, 或者它们过滤了大量的数据,否则重新计算一个分区的速度和与从硬盘中读取基本差不多快.
如果你想有快速故障恢复能力, 使用复制存储级别(例如：用Spark来响应web应用的请求). 所有的存储级别都有通过重新计算丢失数据恢复错误的容错机制, 但是复制存储级别可以让你在RDD上持续的运行任务, 而不需要等待丢失的分区被重新计算.

如果希望定义自己的存储级别 (比如说, 使用复制因子为 3 替代 2), 然后使用

StorageLevel

单例对象中的apply()方法.

Shared Variables(共享变量)

一般来讲, 当一个函数被传递给在远程集群上运行的Spark操作 (例如

map

或

reduce

)
, 它实际上操作的是这个函数用到的所有变量的独立拷贝. 这些变量会被拷贝到每台机器, 而且远程机器上的变量更新都不会回传到驱动程序. 通常认为在任务之间读写共享变量不够高效，然而Spark依然为两种常见的使用模式提供了不同的共享变量: 广播变量和累加器.

Broadcast Variables(广播变量)

广播变量允许程序员保留一个只读的变量，缓存在每一台机器上，而非每个任务保存一份拷贝。它们可以用来给每个结点高效传输一个大的输入数据集.Spark会尝试使用一种高效的广播算法来传播广播变量，从而减少通信的代价.
广播变量是通过调用

SparkContext.broadcast(v)

方法从变量

v中创建的.广播变量是一个对

v

的封装器,它的值可以通过调用

value

方法获得.
如下所示:

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: spark.Broadcast[Array[Int]] = spark.Broadcast(b5c40191-a864-4c7d-b9bf-d87e1a4e787c)

scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)

广播变量被创建以后, 它应该在集群运行的任何函数中代替

v

被调用，而

v

不需再次传递到这些节点上.
此外对象

v

不能在广播后修改，这样可以保证所有结点的收到的都是一模一样的广播值.

Accumulators(累加器)

累加器是一种只能通过关联操作进行“加”操作的变量,因此可以高效被并行支持.它们可以用来实现计数器和求和器.Spark原生就支持Int和Double类型的累加器,开发者可以自己添加新的支持类型.

一个累加器可以通过调用 SparkContext.accumulator(v) 方法从一个初始值 v中创建 .
运行在集群上的任务,可以通过使用+=来给它加值,但是不能读取这个值。只有驱动程序可以使用value的方法来读取累加器的值.

如下的解释器模块，展示了如何利用累加器，将一个数组里面的所有元素相加:

scala> val accum = sc.accumulator(0)
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

进一步学习

在Spark官网上可以找到 example Spark programs . 而且, Spark在

examples/src/main/scala

目录下提供了一些例子,其中一些既有Spark版本也有local版本
(非并行) , 从中可以发现移植到集群上所需的修改. 你还可以把类名传输给Spark中的

bin/run-example

脚本运行; 例如:

./bin/run-example org.apache.spark.examples.SparkPi

任何样例程序在运行时如果没有提供任何参数，都会打印使用帮助.

如果需要优化程序的帮助, configuration 和tuning 指导提供了最佳实践信息.如果要确保你的数据以高效的格式存储在内存中，这一步至关重要.