一起学习Spark（三）Spark RDD编程_spark sc.parallelize

一般来说，每个Spark应用程序都有一个Driver程序，Driver运行用户编写的main函数，并在集群上执行各种并行操作。Spark提供的一个主要抽象概念就是RDD(resilient distributed dataset)，RDD是可以并行计算的跨集群节点分区的元素结合，RDD可以通过hdfs(也可以是其他hadoop支持的文件系统)或者现有的Scala集合转换得到，Spark还允许我们将RDD持久化到内存中，在并行操作中高效地重用。RDD还可以从节点故障中自动恢复。

另一个Spark提供的重要抽象概念是可以在并行操作中使用的共享变量(shared variables)。默认情况下，当Spark在不同的节点上并行地运行一组任务时，它会将函数中使用到的每个变量的副本发送到每个任务上。但有些时候，变量可能需要跨任务之间共享，或者在任务与驱动程序之间共享。Spark支持两种类型的共享变量：广播变量(broadcast variable)和累加器(accumulator)，前者可用于在所有节点的内存中缓存一个值，后者是只进行累加的变量，如计数器和sum。

下面是RDD编程的一些步骤及相关说明(我采用的语言是Scala)：

1.独立Spark程序初始化及连接到Spark

当编写Spark的独立应用程序时，需要添加Spark的相关依赖，我使用了Maven来引入了Spark程序的相关依赖，关于maven的pom.xml配置请参见上一篇博客。

Spark程序需要创建一个SparkContext对象，SparkContext告诉Spark如何访问集群。要创建SparkContext，首先需要构建一个包含应用程序信息的SparkConf对象。

每个JVM上只能激活一个SparkContext。在创建新SparkContext之前，必须调用stop()停止当前的SparkContext。下面是创建SparkContext的代码：

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

appName参数是为Spark程序起的名称，这个名称会显示在Spark UI的界面上。master参数可以是一个Spark、Mesos、Yarn等集群的URL或者在本地模式下使用的“local”字符串。实际上，在集群环境中运行时，我们不希望将master的url硬编码到程序里，我们可以在通过spark-submit提交程序时指定master的url然后在程序里接收。但是，对于本地测试和单元测试，可以通过“local”来内含式的运行Spark。

2.使用Spark Shell

在Spark shell中，系统自动创建了SparkContext，可以直接通过sc变量来使用，我们不需要再额外去创建Spark Context。启动Spark Shell时可以通过--master参数来设置shell连接到哪个master，--jar参数可以添加第三方jar包到classpath中，添加多个时用","分隔。添加jar包的另外一种方式是通过--package参数，在参数后指定jar包在maven仓库中的groupId，artifactId以及version，需要添加多个的时候使用","作为分隔符。

例如在本地模式以4个核心来运行spark shell：

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

添加额外的jar包：

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

通过package方式来指定maven仓库中的jar包

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

3.弹性分布式数据集RDD

Spark围绕着RDD的概念展开，RDD是一组可以并行操作的高容错性的元素集合。有两种方法可以创建RDD:通过Driver中的现有集合进行转换或者引用外部存储系统中的数据集，例如共享文件系统、HDFS、HBase或任何提供Hadoop InputFormat的数据源。

a)通过现有集合创建RDD

在Driver中通过现有集合创建RDD需要调用SparkContext的parallelize方法，调用后集合中的元素被复制以形成可并行操作的分布式数据集。下面是从一个包含1到5的集合中创建RDD的示例：

scala> val data=Array(1,2,3,4,5)
data: Array[Int] = Array(1, 2, 3, 4, 5)
scala> val rdd1=sc.parallelize(data)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:26

data这个RDD一旦被创建后，就可以进行并行操作了。例如，我们可以调用data.reduce((a, b) => a + b)来相加数组中的元素。稍后我们将详细描述在分布式数据集上的操作。

RDD一个重要参数是要将数据集切割成的分区的数量，Spark将在集群的每个分区上都启动一个任务。通常集群中的每个CPU需要2-4个分区，Spark会根据集群自动设置分区的数量，我们也可以在调用parallelize方法时手动设置分区数量，例如sc.parallelize(data,5)。注：在代码某些地方分区也被叫做代码片。

b)通过外部数据源创建RDD

Spark可以从Hadoop支持的任何存储源创建RDD，包括本地文件系统、HDFS、Cassandra、HBase、Amazon S3等。Spark支持Text File、SequenceFiles和任何其他Hadoop InputFormat。

文本文件RDD可以由SparkContext的textfile方法创建，该方法需要文件的URI(或者机器上的本地路径，或者hdfs://、s3a://等URI)，并且将文件按行读取为行集合，下面是一个示例：

scala> val distFile = sc.textFile("data.txt")
distFile: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[10] at textFile at <console>:26

一旦创建，就可以通过数据集操作对distFile进行操作。例如，我们可以使用map和reduce来累加所有行的大小:distFile.map(s = > s.length).reduce((a, b) => a + b)。

一些关于使用Spark读取文本文件的注意事项:

①如果使用本地文件系统，那么在所有节点上的文件系统同一位置都必须能访问到相同的文件，可以采用将文件复制到所有的worker或者使用网络文件系统。

②Spark的所有基于文件的输入方法，包括textFile，都支持读取目录、压缩文件和通配符对应的文件。例如，您可以使用textFile("/my/directory")、textFile("/my/directory/*.txt")和textFile("/my/directory/*.gz")。

③textFile方法还接受第二个可选参数，用于控制分区的数量。在默认情况下，Spark为文件的每个块创建一个分区(在HDFS中，块的默认大小为128MB)，但是也可以通过传递更大的值来请求更多的分区。但是，分区的数量不能少于块的数量。

除了文本文件外，Spark的Scala API还支持其他几种数据格式:

①SparkContext.wholeTextFiles方法可以读取包含多个小文本文件的目录，并组成键值对(文件名、内容)返回。这与textFile方法有所区别，textFile是每个文件中每行返回一条记录。分区是由数据局部性(参考https://blog.csdn.net/sunnyyoona/article/details/53888012)决定的，在某些情况下，数据局部性可能导致分区过少。对于这些情况，wholeTextFiles提供了第二个可选参数，用于控制分区的最小数量。

②对于sequenceFile，使用SparkContext的sequenceFile[K, V]方法，其中K和V是文件中键和值的类型。这些应该是Hadoop的Writable接口的子类，比如IntWritable和Text。此外，Spark允许将一些常见的类型指定为基本数据类型;例如，sequenceFile[Int, String]将自动读取IntWritables和text。

③对于其他的Hadoop输入格式，可以使用SparkContext.hadoopRDD方法，hadoopRDD方法参数为jobConf，input format的类，key的类，value的类。设置这些和你设置hadoop job的输入源是一样的。你也可以使用基于新的MapReduce API的SparkContext.newAPIHadoopRDD方法。

④RDD.saveAsObjectFile方法和SparkContext.objectFile方法支持以序列化Java对象的方式来保存RDD。虽然这不如Avro这样的专门格式有效，但它提供了一种简单的方法来保存任何RDD。

4.RDD相关操作

RDD支持两种类型的操作：transformation(从现有的RDD中创建新的RDD)和action(进行RDD的计算，并且将数据返回到Driver)。比如，map是一个transformation操作，它传递每个数据集元素经过函数重新计算返回一个表示结果的新RDD，reduce是一个action操作，reduce聚合RDD的所有元素经过函数计算后将最终结果返回给驱动程序(尽管还有一个并行执行并返回结果的reduceByKey操作)。

Spark中所有的tranformation操作都是lazy(懒加载)的，因为Spark不会立即计算结果，Spark仅仅只记住作用在一个RDD上的转换步骤，只有当某个action操作需要将结果返回到驱动程序时，才会计算转换。这种设计使Spark可以更高效的运行。例如，我们都知道通过map创建的RDD将在reduce操作中使用，并且只需要将reduce的结果返回给驱动程序，而不是更大的map RDD。

默认情况下，每次执行action操作时，被转换的RDD都会重新计算。但是，您也可以使用persist(或cache)方法将RDD持久化到内存中，在这种情况下，Spark将把元素保存在集群中，在下一次查询时可以更快的访问。此外还支持持久化RDD到磁盘中，或者跨多个节点复制RDD。

a)RDD基础知识

为了说明RDD的基础知识，请思考一下下面的简单程序：

val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)

程序的第一行从外部文件中创建了一个RDD，这个RDD并没有加载到内存中或者进行任何操作，lines仅仅只是一个引用指向这个文件。第二行定义了lineLengths引用指向map操作的结果，同样，因为懒加载的存在map方法并没有被立即执行。最后，我们执行了reduce，这是一个action方法，至此Spark将计算分解为在不同机器上运行的任务，每个机器都执行自己的map部分和本地的reduction，并且返回自己的结果给Driver。

如果我们以后还想再次使用lineLengths，我们可以加上:

lineLengths.persist()

这样，在执行reduce之前，在第一次计算lineLengths后将结果保存到内存中。

b)传递函数给Spark

Spark的API严重依赖于在驱动程序中传递函数以在集群上运行，有两种推荐的方法：

①匿名函数，可用于一小段代码

②全局单例对象中的静态方法。例如，您可以定义对象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方法，Spark会将整个对象都发送到集群上。

同样的，访问外部对象的某个字段也将引用整个对象。为了避免这个问题，最简单的方法是将字段复制到本地变量中，而不是从外部访问它。

c)理解闭包

Spark的难点之一是理解变量和方法在集群上运行时候的作用域和生命周期，在变量作用域之外修改变量的RDD操作经常会引起混淆，下面的示例中我们将通过使用foreach()方法来增加一个计数器观察，但是类似的问题也可能发生在其他操作中。

关于例子的说明：

这是一个简单的求数据元素和的方法，但是根据是否在同一个JVM中执行，它的结果可能有所不同。一个常见的例子是在本地模式下运行Spark(--master = local[n])，而不是将Spark应用程序部署到集群(例如，通过Spark-submit到YARN):

var data=Array(1,2,3,4,5)
var counter = 0
var rdd = sc.parallelize(data)
 
// Wrong: Don't do this!!
rdd.foreach(x => counter += x)
println("Counter value: " + counter)

本地模式vs集群模式

实际上，为了执行作业，Spark会将RDD操作的处理分解为任务，每个任务都由一个独立的执行程序来执行。在执行之前，Spark会计算任务的闭包。闭包指的是是执行程序在执行RDD(在本例中为foreach())计算时必须可见的那些变量和方法。这个闭包被序列化并发送到每个执行器。

发送到每个Executor的变量是当前Driver中变量的副本，因此，当executor在foreach函数中引用counter变量时，引用的counter不是Driver的那个。在Driver的内存中仍然有一个counter，但它对executor不可见，executor只能看到序列化闭包的副本。因此，Driver中counter的最终值仍然为0，因为counter的所有操作引用的都是序列化闭包中的值。

例外的是在本地模式的某些情况下，foreach函数将在与Driver相同的JVM中执行，引用相同的原始counter，并实际更新它。

为了确保在这种情况下，我们的计算目的能准确的达到，应该使用Accumulator(累加器，等会会详细介绍)，Spark中的累加器提供了一个跨集群中的多节点安全更新变量的途径。

一般来说，闭包的构造类似于循环或局部定义的方法，不应该被用来改变某些全局状态(例如例子中的更改全局变量的值)。Spark不对闭包外部引用的对象的突变行为进行定义或保证，本地模式下某些代码可能能达到目的，但这只是偶然的，在分布式环境下这些代码就没法达到预期的结果。

打印RDD中的元素

另一个常见的习惯用法是尝试使用RDD.foreach(println)或RDD.map(println)打印出RDD的元素。在一台机器上，这将达到预期的输出并打印RDD中的所有元素。然而，在集群模式下，这些输出打印的结果将会在实际执行的Executor，而不是Driver上，Driver的控制台任何东西都不会打印。要想在Driver上打印RDD的所有元素一个可以使用的方法是使用collect()将数据收集到Driver：RDD.collect().foreach(println)。但是如果RDD中的元素过多，这可能会导致Driver的内存被耗尽。如果只需要打印RDD的几个元素，则更安全的方法是使用take(): RDD.take(100).foreach(println)。

d)操作Key-Value形式的键值对

虽然大多数的Spark操作都可以在包含任何类型对象的rdd上工作，但有一些特殊操作只能在元素是键值对的rdd上可用。最常见的是分布式“shuffle”操作，例如按key对元素进行分组或聚合。

在Scala中，这些操作在包含Tuple2对象的RDDs上自动可用(语言中的内置元组，只需简单地编写(a, b))，键值对操作在PairRDDFunctions类中，它自动包装tuple的RDD。

例如，下面的代码使用键-值对上的reduceByKey操作来计算文件中每行文本出现的次数:

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

我们还可以使用count.sortbykey()按字母顺序对结果进行排序，最后使用count.collect()将它们作为对象数组返回到Driver。

e)Spark中常见的transformation和action操作

下表中列出了一些常见的transformation操作

再列出一些常见的Actions操作

Spark RDD API还公开了一些操作的异步版本，比如foreachAsync for foreach，它会立即向调用者返回一个futuresponse，而不是在操作完成时阻塞。这可用于管理或等待操作的异步执行。 

f)洗牌操作

Spark中的某些操作会触发称为shuffle的事件。shuffle是Spark用于重新分发数据的机制，跨分区对数据进行不同的分组。这通常涉及跨执行程序和机器复制数据，使shuffle成为一项复杂而昂贵的操作。

介绍

我们可以通过reduceByKey操作作为例子来思考shuffle的过程中到底发生了什么，reduceByKey操作会先查找出每个Key对应的所有Value，再对Value进行聚合，得到一个新的RDD。挑战在于，并不是一个Key的所有Value都必须位于相同的分区，甚至是相同的机器上，但是它们最终需要到同一个节点上来计算结果。

在Spark计算期间，单个任务将对单个分区进行操作——因此，Spark需要执行一个all-to-all操作。它必须从所有分区中读取值，然后将所有分区的值放在一起计算每个键的最终结果——这称为shuffle。

会引起shuffle的操作包括重分区操作(如repartition和coalesce)、ByKey操作(除count外)(如groupByKey和reduceByKey)以及join操作(如cogroup和join)。

性能影响

Shuffle是一种昂贵的操作，因为它涉及磁盘I/O、数据序列化和网络I/O，为了组织shuffle的数据，Spark生成一组任务，其中map任务来组织数据，以及一组reduce任务来聚合数据。这个命名法来自MapReduce，并不直接与Spark的map和reduce操作相关。

在内部，单个map任务在内存中进行计算直到计算完成。计算完成后生成一批结果，这些结果往往需要发送不同的分区，Spark会按照要去往的目标分区对他们进行排序，并将其写入到单个文件中。在reduce端，reduce任务负责读取这些已经排序好的文件。（Sort Based Shuffle 具体原理以后的博客再梳理）

某些洗牌操作会消耗大量堆内存，因为它们使用内存来存储传输之前或之后的记录。具体来说，reduceByKey和aggregateByKey在map端创建数据，ByKey操作在reduce端接收数据。当数据过多，内存无法存储时，Spark会将这些数据溢出到磁盘，导致磁盘I/O的额外开销和垃圾收集的增加。

Shuffle还会在磁盘上生成大量的中间文件。从Spark 1.3开始，这些文件一直保存到不再使用相应的rdd并进行垃圾收集为止。这样做是为了在重新计算时不需要重新创建shuffle文件。如果应用程序保留对这些rdd的引用，或者GC不经常启动，那么只有在很长一段时间后才会发生垃圾收集。这意味着长时间运行的Spark作业可能会消耗大量磁盘空间。临时存储目录的位置由spark.local.dir参数指定(配置Spark Context时的配置参数)。

shuffle行为可以通过调整各种配置参数来进行调整。参见Spark配置指南中的“Shuffle Behavior”部分。

g)RDD的持久化

Spark中最重要的功能之一是跨操作在内存中持久化(或缓存)RDD，当对RDD执行持久化操作时，每个节点都会将自己操作的RDD的partition持久化到内存中，并且在之后对该RDD的反复使用中，直接使用内存中缓存的数据。如果使用得当能将计算速度提高10倍。缓存是迭代算法和快速交互使用的关键工具。

使用persist()或cache()方法都可以将RDD进行持久化。第一次在操作中计算它时，它将保存在节点的内存中。Spark的持久化机制是高容错的——如果RDD的任何partition丢失，Spark会自动通过其源RDD，使用transformation操作重新计算该partition。

此外，可以使用不同的存储级别存储每个持久化的RDD。例如，可以将RDD持久化到磁盘上、持久化到内存中，只要在调用persist方法时传递Storage Level参数即可。cache方法默认就是调用persist方法的无参方法。默认的存储级别是内存。如果需要从内存中删除缓存的数据，可以使用unpersist方法。

Spark支持的完整的存储级别如下

 Spark还会在shuffle操作(例如reduceByKey)时自动持久化一些中间数据，甚至不需要调用persist。这样做是为了避免在shuffle过程中某个节点失败时重新计算整个输入。如果用户计划重用RDD，仍然建议对生成的RDD调用persist。

选择存储级别的策略

内存足够的情况下优先MEMORY_ONLY，因为它不需要序列化与反序列化，效率最高。如果需要节省内存建议使用MEMORY_ONLY_SER，如果内存不够的情况下建议使用MEMORY_AND_DISK_SER，因为序列化能显著减少空间占用。

至于存储级别_2，不建议使用，因为创建副本的过程中需要通过网络传输副本数据，本身开销也很大，用到副本的机会也不多，很多时候不如重新再计算。

持久化数据的删除

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

h)变量共享

一般来说，当Spark的操作分布在集群的各个节点执行时，各节点上使用的所有变量都是单独的副本。这些变量被复制到每个节点上，各个节点对变量的更新不会传回Driver。跨节点的变量共享一般来说都比较低效，Spark提供了两种有限类型的变量共享的方式，一种是广播变量，另一种是累加器。

广播变量

当Executor下的task需要使用到某个变量时，默认的做法是每个task都会获得一个该变量的副本，这样在数据量比较大时，将会浪费大量网络带宽资源。而广播变量允许在每个Executor上只缓存一次该变量，而不是在每个task上都放置一个变量的副本。

可以通过SparkContext.broadcast(v)方法从变量v来创建一个广播变量，broadcast变量是变量v的包装类，它的值可以通过调用value()方法来获得。下面是创建广播变量的示例：

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)

 在创建broadcast变量之后，应该在集群上运行的任何函数中使用它而不是v，这样v就不会多次发送到节点。此外，对象v在广播后不应修改，以确保所有节点都获得广播变量的相同值。

 累加器

Spark的累加器是Spark提供的可以让Task并行对值进行累加的变量，可以通过调用SparkContext.longAccumulator()或SparkContext.doubleAccumulator()来分别累积Long或Double类型的值，从而创建一个数字累加器。然后，在集群上运行的Task节点可以使用add方法对值进行累加。但是，他们无法读取它的值。只有Driver才能使用它的value方法读取累加器的值。下面是使用示例：

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))
...
 
scala> accum.value
res2: Long = 10