Spark Programming Guide (Python) Spark编程指南 (三)

原始版本参见：http://cholerae.com/2015/04/11/-%E7%BF%BB%E8%AF%91-Spark%E7%BC%96%E7%A8%8B%E6%8C%87%E5%8D%97-Python%E7%89%88/

对部分内容有修改，恕本人水平有限，如有错误，在所难免。

1.  打印出RDD中的元素

a)  一个常见做法是试图通过如下语句打印出RDD中每一个元素:

Rdd.foreach(println)
Rdd.map(println)b)  在单机上这将实现预期的结果，打印出每一个元素。然而，在集群模式下executors调用，输出到stdout的结果将是输出到每一个executors的stdout，而不是driver所在的机器上。所以driver所在机器的stdout并不会捕捉到预期结果。为了在driver所在机器上打印出所有元素，你可以使用collect()方法，首先将RDD转移到driver所在节点。
Rdd.collect().foreach(println)
c)  上式将导致driver内存耗尽，因为collect()将整个RDD取到单台机器上。如果你只是需要打印RDD的部分节点，一个安全的做法是使用:

take():rdd.take(100).foreach(println)
2.  使用键值对

a)  虽然大部分Spark的RDD操作都支持所有种类的对象，但是有少部分特殊的操作只能作用于键值对类型的RDD。这类操作中最常见的就是分布的shuffle操作，比如将元素通过键来分组或聚集计算。

b)  在Python中，这类操作一般都会使用Python内建的元组类型，比如(1, 2)。它们会先简单地创建类似这样的元组，然后调用你想要的操作。

c)  比如，一下代码对键值对调用了reduceByKey操作,来统计每一文本行在文本文件中出现的次数:

lines = sc.textFile("data.txt")
pairs = lines.map(lambda s: (s, 1))
counts = pairs.reduceByKey(lambda a,b: a + b)
d)  我们还可以使用counts.sortByKey()，比如，当我们想将这些键值对按照字母表顺序排序，然后调用counts.collect()方法来将结果以对象列表的形式返回。

3.  Transformations操作

a)  下面的表格列出了Spark支持的常用转化操作。欲知细节，请查阅RDD API文档（Scala, Java, Python）和键值对RDD函数文档（Scala, Java）。

建议读者查看原始文档，以免望文生义：

http://spark.apache.org/docs/latest/programming-guide.html#transformations

4.  Actions操作

a)  下面的表格列出了Spark支持的部分常用启动操作:

建议读者查看原始文档，以免望文生义：

http://spark.apache.org/docs/latest/programming-guide.html#actions

5.  RDD持久化

a)  Spark的一个重要功能就是在将数据集持久化（或缓存）到内存中以便在多个操作中重复使用。当我们持久化一个RDD是，每一个节点将这个RDD的每一个分片计算并保存到内存中以便在下次对这个数据集（或者这个数据集衍生的数据集）的计算中可以复用。这使得接下来的计算过程速度能够加快（经常能加快超过十倍的速度）。缓存是加快迭代算法和快速交互过程速度的关键工具。

b)  你可以通过调用persist或cache方法来标记一个想要持久化的RDD。在第一次被计算产生之后，它就会始终停留在节点的内存中。Spark的缓存是具有容错性的——如果RDD的任意一个分片丢失了，Spark就会依照这个RDD产生的转化过程自动重算一遍。

c)  另外，每一个持久化的RDD都有一个可变的存储级别，这个级别使得用户可以改变RDD持久化的储存位置。比如，你可以将数据集持久化到硬盘上，也可以将它以序列化的Java对象形式（节省空间）持久化到内存中，还可以将这个数据集在节点之间复制，或者使用Tachyon将它储存到堆外。这些存储级别都是通过向persist()传递一个StorageLevel对象（Scala, Java, Python）来设置的。存储级别的所有种类请见下表：

http://spark.apache.org/docs/latest/programming-guide.html#rdd-persistence

d)  Spark还会在shuffle操作（比如reduceByKey）中自动储存中间数据，即使用户没有调用persist。这是为了防止在shuffle过程中某个节点出错而导致的全盘重算。不过如果用户打算复用某些结果RDD，我们仍然建议用户对结果RDD手动调用persist，而不是依赖自动持久化机制。

6.  存储级别的选取

a)  如果你的RDD很适合默认的级别（MEMORY_ONLY）,那么久使用默认级别吧。这是CPU最高效运行的选择，能够让RDD上的操作以最快速度运行。

b)  否则，试试MEMORY_ONLY_SER选项并且选择一个快的序列化库来使对象的空间利用率更高，同时尽量保证访问速度足够快。

c)  不要往硬盘上持久化，除非重算数据集的过程代价确实很昂贵，或者这个过程过滤了巨量的数据。否则，重新计算分片有可能跟读硬盘速度一样快。

d)  如果你希望快速的错误恢复（比如用Spark来处理web应用的请求），使用复制级别。所有的存储级别都提供了重算丢失数据的完整容错机制，但是复制一份副本能省去等待重算的时间。

e)  在大内存或多应用的环境中，处于实验中的OFF_HEAP模式有诸多优点：

         i.     这个模式允许多个执行者共享Tachyon中的同一个内存池

       ii.     这个模式显著降低了垃圾回收的花销

      iii.     在某一个执行者个体崩溃之后缓存的数据不会丢失

7.  删除数据

a)  Spark会自动监视每个节点的缓存使用同时使用LRU算法丢弃旧数据分片。如果你想手动删除某个RDD而不是等待它被自动删除，调用RDD.unpersist()方法。

8.  共享变量

a)  通常情况下，当一个函数传递给一个在远程集群节点上运行的Spark操作（比如map和reduce）时，Spark会对涉及到的变量的所有副本执行这个函数。这些变量会被复制到每个机器上，而且这个过程不会被反馈给驱动程序。通常情况下，在任务之间读写共享变量是很低效的。但是，Spark仍然提供了有限的两种共享变量类型用于常见的使用场景：Broadcastvariables和accumulators。

b)  Broadcastvariables允许程序员在每台机器上保持一个只读变量的缓存而不是将一个变量的拷贝传递给各个任务。它们可以被使用，比如，给每一个节点传递一份大输入数据集的拷贝是很低效的。Spark试图使用高效的广播算法来分布广播变量，以此来降低通信花销。

c)  可以通过SparkContext.broadcast(v)来从变量v创建一个广播变量。这个广播变量是v的一个包装，同时它的值可以功过调用value方法
4000
来获得。以下的代码展示了这一点：

>>> broadcastVar = sc.broadcast([1,2, 3])
<pyspark.broadcast.Broadcast object at0x102789f10>

>>> broadcastVar.value
[1, 2, 3]
d)  在广播变量被创建之后，在所有函数中都应当使用它来代替原来的变量v，这样就可以保证v在节点之间只被传递一次。另外，v变量在被广播之后不应该再被修改了，这样可以确保每一个节点上储存的广播变量的一致性（如果这个变量后来又被传输给一个新的节点）。

9.  累加器

a)  累加器是在一个相关过程中只能被”累加”的变量，对这个变量的操作可以有效地被并行化。它们可以被用于实现计数器（就像在MapReduce过程中）或求和运算。Spark原生支持对数字类型的累加器，程序员也可以为其他新的类型添加支持。累加器被以一个名字创建之后，会在Spark的UI中显示出来。这有助于了解计算的累进过程（注意：目前Python中不支持这个特性）。

b)  可以通过SparkContext.accumulator(v)来从变量v创建一个累加器。在集群中运行的任务随后可以使用add方法或+=操作符（在Scala和Python中）来向这个累加器中累加值。但是，他们不能读取累加器中的值。只有驱动程序可以读取累加器中的值，通过累加器的value方法。

c)  以下的代码展示了向一个累加器中累加数组元素的过程：

>>> accum = sc.accumulator(0)
Accumulator<id=0, value=0>

>>> sc.parallelize([1, 2, 3,4]).foreach(lambda x: accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasksfinished in 0.317106 s

scala> accum.value
10
这段代码利用了累加器对int类型的内建支持，程序员可以通过继承AccumulatorParam类来创建自己想要的类型支持。AccumulatorParam的接口提供了两个方法：'zero'用于为你的数据类型提供零值；'addInPlace'用于计算两个值得和。比如，假设我们有一个Vector`类表示数学中的向量，我们可以这样写：

classVectorAccumulatorParam(AccumulatorParam):
def zero(self, initialValue):
return Vector.zeros(initialValue.size)

def addInPlace(self, v1, v2):
v1 += v2
return v1

# Then, create an Accumulator of this type:
vecAccum =sc.accumulator(Vector(...), VectorAccumulatorParam())
累加器的更新操作只会被运行一次，Spark提供了保证，每个任务中对累加器的更新操作都只会被运行一次。比如，重启一个任务不会再次更新累加器。在转化过程中，用户应该留意每个任务的更新操作在任务或作业重新运算时是否被执行了超过一次。

累加器不会该别Spark的惰性求值模型。如果累加器在对RDD的操作中被更新了，它们的值只会在启动操作中作为RDD计算过程中的一部分被更新。所以，在一个懒惰的转化操作中调用累加器的更新，并没法保证会被及时运行。下面的代码段展示了这一点：

accum = sc.accumulator(0)
data.map(lambda x => acc.add(x); f(x))
# Here, acc is still 0 because noactions have cause the `map` to be computed.
10. 单元测试

Spark对单元测试是友好的，可以与任何流行的单元测试框架相容。你只需要在测试中创建一个SparkContext，并如前文所述将master的URL设为local，执行你的程序，最后调用SparkContext.stop()来终止运行。请确保你在finally块或测试框架的tearDown方法中终止了上下文，因为Spark不支持两个上下文在一个程序中同时运行。