Spark编程指南

怼烎@ 2022-07-28 05:04 140阅读 0赞

# 1、概述 #

在高层的角度上看，每一个Spark应用都有一个驱动程序（driver program）。驱动程序就是运行用户的main主程序并在集群上执行各种并行操作的程序。Spark中的一个主要的抽象概念就是弹性分布数据集（resilient distributed dataset,RDD），RDD是分布在多个节点构成的集群上的元素的集合，并支持并行操作。RDD可以由Hadoop的分布式文件系统（或其他支持Hadoop分布式系统的文件系统）中的文件创建，也可以通过在驱动程序中的Scala集合创建，同时，对一个RDD进行转换操作（transform）也可以创建一个新的RDD。用户也可以将RDD存储在内存中，这样RDD就可以在后序的并行操作中高效地重复使用。最后，RDD能够从某个节点的实效中进行恢复。

Spark的第二个抽象概念就是共享变量（shared variables）。共享变量应用于并行操作中。默认情况下，当Spark在几个节点构成的集群上并行执行一系列任务时，Spark会携带函数中使用的每一个变量到每一个任务中。有时，一些变量需要在几个任务间共享，或者在任务和驱动程序间共享。Spark支持两种共享变量：将数据缓存中所有节点中的广播变量（broadcast variables），和只能增加的累加器（accumulators），比如计数器和总和（sums）。

这里我只使用了Scala的版本。Spark还支持Java和Python。如果打开Spark的交互式脚本，很容易理解这个知道的内容。

# 2、连接Spark #

Spark 1.6.1使用Scala 2.10版本。如果使用Scala编写Spark应用，应该使用兼容的版本（比如2.10.x）。

在[Intellij Idea搭建Spark开发环境][Intellij Idea_Spark]中介绍了使用Idea+Maven搭建Spark开发环境。如果编写Spark应用，应该添加Spark的依赖，具体的信息如下：

groupId = org.apache.spark
    artifactId = spark-core_2.10
    version = 1.6.1

同样，如果使用HDFS的分布式文件系统，也要添加hadoop-client的依赖：

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

最后，需要在Scala中添加如下import语句：

import org.apache.spark.SparkContext
    import org.apache.spark.SparkConf

这里要注意的是，在Spark 1.3.0版本以前，Scala中需要显示添加import org.apache.spark.SparkContext.\_来使用重要的隐式转换。不过1.3.0以后的版本就不需要了。

# 3、初始化Spark #

编写Spark的第一件事就是创建SparkContext对象，来告诉Spark如何使用一个集群。创建SparkContext对象需要使用SparkConf对象，这个对象包含一些关于应用程序的信息。

注意，每一个JVM上只能有一个活跃的SparkContext对象。所以，必须调用stop()来终止SparkContext对象才能创建另一个新的SparkContext对象。

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

其中，appName参数是应用程序在集群中的名字。master参数指定主节点的位置，它可以是一个Spark，Mesos或者YARN集群的URL地址，也可以使用本地模式的“local”。一般来说，当应用程序运行在集群上时，在代码上硬编码master并不方便，而是在使用spark-submti提交应用的时候使用参数指定master。对于本地测试来说，可以使用“local”来运行Spark程序。

## 3.1、使用交互式shell ##

Spark提供了交互式的shell，在这个交互式shell中，已经创建了一个SparkContext对象，这个变量就是sc，不用创建直接使用即可，自己创建的反而不能用。可以使用--master参数指定sc链接到哪个master，还有很多参数可以选择，这里给出几个例子。

下面使用本地模式4个核：

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

下面使用--jar指定了要运行的jar包：

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

下面使用--packages添加了依赖：

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

上面仅仅是一些例子，Spark还有很多启动参数，可以运行spark-shell --help获得更多的信息。下面是master可选的值：

<table>  Master URLs  
 <tbody> 
  <tr> 
   <td>Master URL</td> 
   <td>含义</td> 
  </tr> 
  <tr> 
   <td>local</td> 
   <td>在本地运行Spark并且只有一个worker线程（也就是说没有并行）</td> 
  </tr> 
  <tr> 
   <td>local[K]</td> 
   <td>在本地运行Spark并使用K个worker线程（基本上设置K值为本地机器的核心数）</td> 
  </tr> 
  <tr> 
   <td>local[*]</td> 
   <td>在本地运行Spark，并且使用本地机器尽可能多的核心数</td> 
  </tr> 
  <tr> 
   <td>spark：//HOST:PORT</td> 
   <td>连接到给定的Spark standalone集群上。端口号必须是主节点设置使用的端口号，默认使用7077</td> 
  </tr> 
  <tr> 
   <td>mesos：//HOST:PORT</td> 
   <td>连接到给定的Mesos集群。端口号默认使用5050</td> 
  </tr> 
  <tr> 
   <td>yarn</td> 
   <td>连接到yarn集群，可以通过--deploy-mode参数设置使用client或cluster两种模式</td> 
  </tr> 
  <tr> 
   <td>yarn-client</td> 
   <td>和使用--deploy-mode client连接到yarn等价</td> 
  </tr> 
  <tr> 
   <td>yarn-cluster</td> 
   <td>和使用--deploy-mode cluster连接到yarn等价</td> 
  </tr> 
 </tbody> 
</table>

# 4、弹性分布数据集（RDD） #

Spark的一个核心抽象概念就是弹性分布数据集（resilient distributed dataset，RDD）。RDD是一个可以并行操作的可容错的元素集合。有两种方法可以创建一个RDD：将驱动程序中已存在的集合进行并行化操作，或者从外部存储系统中创建RDD。事实上，还可以通过对已有的RDD进行转化操作创建一个新的RDD。

## 4.1、对集合序列化 ##

可以通过调用SparkContext对象的parallelize方法把一个在驱动程序中已存在的集合序列化为RDD。集合中的元素会被复制为分布式数据集来支持并行计算。下面的例子将一个1到5的数组序列化操作为一个RDD：

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

RDD一旦创建，就可以并行操作。例如，可以使用下面的操作计算所有元素的和：

val sum=distData.reduce((a,b)=>a+b)

稍后会介绍RDD的一些操作。

在将一个集合序列化时一个重要的参数就是partitions，也就是说要将这个集合分成几个部分。Spark会对每个部分执行一个任务来达到并行操作的效果。一般来说，集群中的每个CPU分配2到4个部分。通常，Spark会基于集群的配置自动设置这个值。然而，用户也可以自己设置这个值：

val distData=sc.parallelize(data,10)

这样，就把data分为10个部分。

## 4.2、外部数据集 ##

Spark可以通过任何支持Hadoop的存储系统创建分布式数据集，包括本地文件系统，HDFS，Cassandra，HBase，Amazon S3等等。Spark支持文本文件，SequenceFiles，和任何实现了Hadoop InputFormat接口的文件。

文本文件的RDD可以通过SparkContext对象的textFile方法创建。这个方法传递一个URI参数来指定文件，URI参数可以使本地文件路径或者hdfs://等，然后将文件读取为行的集合。例子如下：

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

一旦创建，distFile就可以进行数据集操作。例如，下面的代码使用map和reduce操作计算所有行的大小：

distFile.map(s => s.length).reduce((a, b) => a + b)

使用Spark读取文件时的一些注意事项：

*  如果传递本地文件的路径，那么这个文件也必须在集群的所有worker节点中的相同路径上。或者将这个文件复制到所有worker上的相同路径上；
 *  Spark中所有关于文件输入的方法，包括textFile，都支持目录，压缩文件和通配符。比如`textFile("/my/directory")`,`textFile("/my/directory/*.txt")`, 和 `textFile("/my/directory/*.gz")；`
 *  textFile方法也可以通过第二个参数来指定这个文件被分为几个部分。默认情况下，文件基于块来进行划分（在HDFS中块默认为64MB或128MB，我这里是128MB）。不过可以指定的分区数比块数多，但不能比块数少；

除了文本文件，Spark的Scala API还支持另外的数据格式：

*  SparkContext.wholeTextFiles允许读取一个目录下所有的小文本文件，然后返回（文件名，内容）键值对。这和textFile返回文件中的所有行不同；
 *  对于SequenceFiles，可以使用SparkContext对象的sequenceFile\[K,V\]方法，其中K和V是文件中键和值的类型。这些文件必须是实现Hadoop的Writable接口的类，比如IntWritable和Text。而且，Spark也允许使用基本类型，比如sequenceFile\[Int,String\]就会自动读取IntWritable和Text；
 *  对于其它的Hadoop输入格式，可以使用SparkContext.hadoopRDD方法；
 *  RDD.saveAsObjectFile和SparkContext.objectFiles方法支持将一个RDD中的Java对象序列化为对象文件。不过这个方法并不像Avro那样高效；

## 4.3、RDD操作 ##

RDD支持两种类型的操作：转换操作（transformations）和行动操作（actions）。转换操作从一个已有的RDD创建一个新的RDD；行动操作对这个RDD数据集进行一系列运算后返回驱动程序一个结果。区分两种操作的办法就是看返回结果的类型，如果返回的是一个RDD，那么就是转换操作，否则就是行动操作。比如，map就是一个转换操作，它把数据集中的每一个元素都调用一个函数，将结果作为新的RDD的元素。而reduce就是一个行动操作，它对数据集的所有元素调用一个聚合函数，然后把最终结果返回驱动程序（尽管还存在一个并行的方法reduceByKey返回一个分布数据集）。

Spark中所有的转换操作都是惰性求值的。所谓惰性求值，是说并不马上计算结果，而仅仅记住对这个RDD进行的转换操作序列。只有当对这个RDD调用一个需要返回给驱动程序一个结果的行动操作时才计算结果。Spark的这个设计使得程序运行得更有效。比如，使用map方法创建的一个RDD很可能调用reduce来计算结果并将结果返回个驱动程序，惰性求值使得只需给驱动程序返回reduce计算的结果，否则要返回一个map操作创建的RDD，显然这个代价太大了。

默认情况下，对通过转换操作形成的RDD执行行动操作时都会重新计算这个RDD。这种情况下，可以将RDD通过persist或者cache方法存储在内存中，Spark会把数据集中的元素存在集群中的所有节点上，这样下次计算的时候就可以快速得到结果。同样，Spark也可以把RDD存储在磁盘上，或者在多个节点间进行复制。

### 4.3.1、基础 ###

下面的程序给出了RDD的基础操作：

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

第一行通过SparkContext对象的textFile读取文件创建了一个RDD变量lines，lines数据集并不会加载到内存中或采取其它行动，它仅仅是一个指向文件的指针。第二行定义了一个变量lineLengths保存map操作的结果，这是一个转换操作，因此lineLengths也是一个RDD，map方法需要一个函数参数，对象Spark传递一个参数后面会介绍，这里的函数参数将每一行映射为一个数值，这个数值就是这一行的长度。注意，由于Spark的惰性求值程序进行到这里并没有计算lineLengths。最终，对lineLengths调用reduce方法，这是一个行动操作。这时，Spark将计算分解为多个任务运行在集群的多个机器上，然后每个机器执行自己那部分的map和reduce操作，计算完后将自己这部分的结果返回给驱动程序。

如果在后序的操作中还会用到lineLengths，就可以将它存储在内存中：

lineLengths.persist()

这样，下一次调用reduce的时候就不用再计算了。

### 4.3.2、给Spark传递函数 ###

Spark的API严重依赖从驱动程序传递函数给集群。有两种方式来传递函数：

*  匿名函数语法，可以减少代码；
 *  在一个单独的object中定义一个函数。比如，可以定义一个object MyFunctions，然后传递MyFunctions：

object MyFunctions {
      def func1(s: String): String = { ... }
    }
    
    myRdd.map(MyFunctions.func1)

注意还可以传递一个类实例的方法引用（和object相反），不过这需要把包含这个类的对象同这个方法传递过去。比如，考虑下面的代码：

class MyClass {
      def func1(s: String): String = { ... }
      def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
    }

这里，如果我们创建一个新的MyClass实例然后调用doStuff方法，里面的map引用这个MyClass实例里面的func1方法，所以整个对象需要被传递到集群中。这和  `rdd.map(x => this.func1(x))`相似。

同样，如果访问外部属性也需要传递整个对象：

class MyClass {
      val field = "Hello"
      def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
    }

这和 rdd.map(x => this.field + x)相似。为了避免这样，可以将属性复制到方法内部而不是在外部访问：

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

### 4.3.3、理解闭包 ###

Spark的一个难点就是理解在集群执行代码时变量和方法的作用域和声明周期。对在RDD作用域外部的变量进行修改操作很让人迷惑。在下面的例子中，我们考察使用foreach()方法增加的代码，不过同样的事情也会发生在其它的操作中。

#### 4.3.3.1、一个例子 ####

考虑下面的RDD元素相加的例子，这个例子是否运行在同一个JVM上会产生不同的结果。比如，运行在本地模式（--master=local\[n\]）和将Spark应用部署到集群上（spark-submit）：

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

####    4.3.3.2、本地模式与集群模式 ####

上面代码的行为是不确定的，可能不会得到预期结果。执行工作（job）时，Spark会把RDD的操作分解为多个任务（task），每一个任务由一个执行者（executor）执行。在执行之前，Spark会计算任务的闭包（closure）。闭包就是执行者在对RDD进行自己那部分计算时需要可见的变量和方法（在这个例子中是foreach）。这些闭包会被序列化并传递到每个执行者上。

传递到每个执行者上的闭包中的变量现在被赋值，现在foreach方法中的counter不再是驱动程序中的counter。在驱动节点的内存中仍然有一个counter，但这个counter并不能被执行者访问到，执行者只能访问到在闭包的序列化中复制到执行者上的counter。因此，counter的最终结果仍然是0，因为对counter的所有操作都是在闭包序列化中的counter值。

在本地模式，一些情况下foreach方法会像驱动程序一样运行在同一个JVM上，因此会引用原始的counter，并正确更新它的值。

在这种情况下为了写出良好行为的代码，应该使用累加器（accumulator）。Spark中的累加器在特殊情况下使用，它提供一种在集群中执行被分解的情况下安全更新变量值的机制。累加器的细节会在这个指导的累加器部分讨论。

一般的，闭包，不应该被用来改变全局状态。Spark不定义也不保证对从闭包外部引用的对象的做出改变的行为。一些代码可能在本地模式下能够运行，不过这仅仅是巧合，在分布式模式下就不会得到期望的结果。如果需要全局的聚合操作就使用累加器。

#### 4.3.3.3、打印RDD中的元素 ####

另一个常见用法就是使用rdd.foreach(println)或rdd.map(println)方法打印RDD中的元素。在一台机器上，这个操作会得到预期的结果，打印出RDD中的所有元素。然而在集群模式下，执行者调用的打印方法会在执行者的标准输出（stdout）上打印结果，而不是驱动程序的标准输出上，因此驱动程序的标准输出上并没有结果。如果要将结果打印在驱动程序的标准输出上，需要使用collect()方法先使执行者将各自的RDD部分返回给驱动程序，然后调用foreach或map，即：

rdd.collect().map(println)

不过，这样可能会耗尽驱动程序的内存，因为大多数情况下RDD很大。如果只想查看RDD中的部分元素 ，可以使用take()方法：

rdd.take(100).foreach(println)

这样只取RDD中的100个元素。

### 4.3.4、键值对操作 ###

尽管Spark中的大多数操作都支持任何类型的RDD，不过Spark为包含键值对类型的RDD提供了一些特殊的操作。最常见的操作就是“混洗”（shuffle），比如根据键来对元素进行分组或聚合。

在Scala中，这些操作自动支持包含Tuple2（Scala中内置的元组类型，可以使用（a，b）来创建）对象的RDD。

比如，下面的代码使用reduceByKey方法操作键值对RDD来计算每一行出现的次数：

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

第一行，使用textFile方法从文本文件创建一个RDD变量lines，第二行使用map操作，将lines中的每一个元素（即一行）映射为一个元组（line，1），构成一个键值对RDD，即pairs，第三行调用行动操作reduceByKey，根据键进行reduce操作，将键相同的值相加，就得到了每一行元素出现的次数。

我们也可以使用counts.sortByKey()来对结果进行排序，还可以使用counts.collect()将结果以数组的形式返回给驱动程序。

注意：当在键值对中的键使用自定义类型时，必须保证这个自定义类型的equals方法和hashCode方法匹配。也就是说，如果两个自定义类型的变量a，b的hashCode方法的返回值相同，那么a.equals(b)也一定返回true。

### 4.3.5、转换操作 ###

下表列出了Spark支持的转换操作。具体的细节可以查看RDD的API文档（包括Scala，Java，Python，R），和键值对RDD的函数文档（Scala和R）：

<table> 
 <tbody> 
  <tr> 
   <td>转换操作</td> 
   <td>含义</td> 
  </tr> 
  <tr> 
   <td>map(func)</td> 
   <td>对RDD中的每个元素调用func函数，然后返回结果构成新的RDD</td> 
  </tr> 
  <tr> 
   <td>filter(func)</td> 
   <td>返回一个由通过传给filter的函数的元素组成的RDD</td> 
  </tr> 
  <tr> 
   <td>flatMap(func)</td> 
   <td>将函数应用于RDD中的每一个元素，将返回的迭代器的所有内容构成新的RDD</td> 
  </tr> 
  <tr> 
   <td>mapPartitions(func)</td> 
   <td>和map类似，不过运行在RDD的不同分块上，因此func的类型必须是Iterator&lt;T&gt;=&gt;Iterator&lt;U&gt;</td> 
  </tr> 
  <tr> 
   <td>mapPartitionsWithIndex(func)</td> 
   <td>和mapPartitions类似，不过func函数提供一个整数值表示分块的下标，所以函数的类型是(Int,Iterator&lt;T&gt;=&gt;Iterator&lt;U&gt;)</td> 
  </tr> 
  <tr> 
   <td>sample(withReplacement,fraction,seed)</td> 
   <td>对RDD采样，以及是否替换</td> 
  </tr> 
  <tr> 
   <td>union(otherDataset)</td> 
   <td>生成一个包含两个RDD中所有元素的RDD</td> 
  </tr> 
  <tr> 
   <td>intersection(otherDataset)</td> 
   <td>返回由两个RDD中共同元素组成的RDD</td> 
  </tr> 
  <tr> 
   <td>distinct([numTasks])</td> 
   <td>返回去除原RDD中重复元素的新的RDD</td> 
  </tr> 
  <tr> 
   <td>groupByKey([numTasks])</td> 
   <td>对具有相同键的值进行分组。注意如果仅仅是为了聚合，使用reduceByKey或aggregateByKey性能更好</td> 
  </tr> 
  <tr> 
   <td>reduceByKey(func,[numTasks])</td> 
   <td>合并既有相同键的值</td> 
  </tr> 
  <tr> 
   <td>aggregateByKey(zeroValue)(seqOp,combOp,[numTasks])</td> 
   <td>和reduceByKey类似，不过需要提供一个初始值</td> 
  </tr> 
  <tr> 
   <td>sortByKey([ascending],[numTasks])</td> 
   <td>返回一个根据键排序的RDD</td> 
  </tr> 
  <tr> 
   <td>join(otherDataset,[numTasks])</td> 
   <td>对两个RDD进行内连接。其它的连接操作还有leftOuterJoin，rightOuterJoin和fullOuterJoin</td> 
  </tr> 
  <tr> 
   <td>cogroup(otherDataset,[numTasks])</td> 
   <td>也叫groupWith，对类型(K,V)和(K,W)的RDD进行操作，返回(K,(Iterable&lt;V&gt;,Iterable&lt;W&gt;))类型的RDD</td> 
  </tr> 
  <tr> 
   <td>cartesian(otherDataset)</td> 
   <td>对类型T和U的RDD进行操作，返回(T,U)类型的RDD</td> 
  </tr> 
  <tr> 
   <td>pipe(command,[envVars])</td> 
   <td>将RDD的每个分区通过管道传给一个shell脚本</td> 
  </tr> 
  <tr> 
   <td>coalesce(numPartitions)</td> 
   <td>减少RDD的分区数量。当对一个大的RDD执行filter操作后使用会有效</td> 
  </tr> 
  <tr> 
   <td>repartition(numPartitions)</td> 
   <td>对RDD重新分区</td> 
  </tr> 
  <tr> 
   <td>repartitionAndSortWithinPartitions(partitioner)</td> 
   <td>根据给定的partitioner对RDD重新分区，在每个分区再根据键排序</td> 
  </tr> 
 </tbody> 
</table>

### 4.3.6、行动操作 ###

下面列出了RDD的行动操作：

<table> 
 <tbody> 
  <tr> 
   <td>行动操作</td> 
   <td>含义</td> 
  </tr> 
  <tr> 
   <td>reduce(func)</td> 
   <td>使用func函数并行整合RDD中的所有元素</td> 
  </tr> 
  <tr> 
   <td>collect()</td> 
   <td>返回RDD中的所有元素</td> 
  </tr> 
  <tr> 
   <td>count()</td> 
   <td>返回RDD中的元素个数</td> 
  </tr> 
  <tr> 
   <td>first()</td> 
   <td>返回RDD中的第一个元素</td> 
  </tr> 
  <tr> 
   <td>take(n)</td> 
   <td>返回RDD中的n个元素</td> 
  </tr> 
  <tr> 
   <td>takeSample(withReplacement,num,[seed])</td> 
   <td>从RDD中返回任意一些元素，结果不确定</td> 
  </tr> 
  <tr> 
   <td>takeOrdered(n,[ordering])</td> 
   <td>从RDD中按照提供的顺序返回最前面的n个元素</td> 
  </tr> 
  <tr> 
   <td>saveAsTextFile(path)</td> 
   <td>将RDD中的元素写入文本文件，Spark会调用元素的toString方法</td> 
  </tr> 
  <tr> 
   <td>saveAsSequenceFile(path)</td> 
   <td>将RDD中的元素保存为Hadoop的SequenceFile文件</td> 
  </tr> 
  <tr> 
   <td>saveAsObjectFile(path)</td> 
   <td>将RDD中的元素使用Java中的序列化保存为对象文件，可以使用SparkContext.objectFile()读取</td> 
  </tr> 
  <tr> 
   <td>countByKey()</td> 
   <td>操作键值对RDD，根据键值分别计数</td> 
  </tr> 
  <tr> 
   <td>foreach(func)</td> 
   <td>对RDD中的每个元素调用给定的函数func</td> 
  </tr> 
 </tbody> 
</table>

### 4.3.7、混洗操作（Shuffle） ###

在Spark中，一些操作会触发一个叫做混洗（shuffle）的事件。混洗是Spark的机制，通过对数据进行重新分组使得同一组的在同一个分区。这通常会导致在执行者和机器之间数据的复制与传递，因此混洗操作是一个复杂并消耗性能的操作。

#### 4.3.7.1、背景 ####

我们以reduceByKey操作来理解混洗操作期间发生了什么。reduceByKey操作会生成一个新的RDD，原键值对RDD中的所有元素根据键的不同分组后使用reduce操作得到一个结果，由所有的键和这个结果构成的键值对元素构成了这个新的RDD。问题在于并不是每所有键相同的元素都在同一个分区上，甚至不在同一个机器上，但为了计算结果，它们必须重新存储到同一个位置。

在Spark中，数据一般不会为了某个操作而具体地根据需要进行分组存储。在计算过程中，每一个执行者对自己的分区进行计算，因此，为了给执行者组织对应的分区，Spark需要执行一个满射操作来重新组织数据。Spark必须读取所有的分区来得到所有键，然后对每个键将键相同的元素组织到一起执行reduce操作来计算结果。这就是混洗。

尽管经过混洗后每个分区的元素集合分区本身都是确定的，但是元素的顺序不确定。如果要是数据具有确定的顺序，可以使用下面的混洗方法：

*  使用mapPartitions来对分区排序；
 *  使用reparttitionAndSortWithinPartitions高效的在重分区的同时排序分区；
 *  使用sortBy排序一个RDD；

可以导致混洗的操作有重分区操作比如reparation和coalesce，ByKey操作（除了计数counting）比如groupByKey和reduceByKey，还有连接操作（join）比如cogroup和join。

#### 4.3.7.2、性能影响 ####

由于涉及到磁盘I/O，数据序列化和网络I/O，所以混洗操作性能消耗较大。为了重新组织数据，Spark会产生一些map任务来组织数据，一些reduce任务来进行聚合。这一名称来自于MapRedece，但并不直接和Spark的map和reduce操作相关。

本质上，map任务的结果会存在内存中直到存不下为止。然后，这些结果根据所在的分区进行排序，写入单一的文件中。在reduce阶段，任务会读取这些排好序的相关分块。

一些混洗操作会消耗大量的堆空间，因为它们在转化记录之前或之后会以内存数据结构组织记录。具体来说，reduceByKey和aggregateByKey在map阶段构造这些数据结构，然后ByKey系列操作在reduce阶段生成数据。当内存中存不下这些数据时，Spark会将这些数据存到磁盘中，导致额外的磁盘I/O并增加垃圾收集。

混洗也会在磁盘上产生大量的中间数据。在Spark 1.3中，这些数据会一直保存到相关的RDD不会再次使用，然后被当做垃圾收集。这对于操作谱系还会重新计算的时候是有益的。如果应用经常使用这些RDD或者垃圾收集机制没能经常收集，垃圾收集会经过很长一段时间才发生。这意味着长时间运行的应用会占用大量的磁盘空间。在创建SparkContext时可以使用spark.local.dir来指定这个临时存储路径。

通过调整各种参数配置可以设置混洗的行为。这会在Spark配置里介绍。

## 4.4、RDD的持久化（缓存） ##

Spark中的一个重要特征就是在操作过程中将数据集缓存在内存中。当缓存一个RDD后，计算RDD的节点会分别保存它们所求出的分区数据，然后在随后的操作中重复使用。这使得后序的操作执行的更快（通常快10几倍）。持久化是迭代式算法和快速交互式使用的关键。

可以使用persist或cache方法持久化一个RDD。当对一个RDD第一次执行行动操作时，RDD会保存在节点的内存中。Spark的缓存是可容错的，意味着如果某个分区丢失了，RDD会自动根据在创建这个RDD时的转换操作重新计算。

而且，每一个持久化的RDD可以使用不同的持久化级别，允许将数据持久化到磁盘，作为Java序列化对象存储在内存，在节点中备份，或者存在堆外空间上。这些持久化级别可以通过传递一个StorageLevel对象给persist方法。cache方法只能使用默认的StorageLevel.MEMORY\_ONLY这一个持久化级别。下面是所有的级别：

<table> 
 <tbody> 
  <tr> 
   <td>持久化级别</td> 
   <td>含义</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY</td> 
   <td>在JVM上存储非序列化的Java对象。如果内存不够，一些分区不会存储，直到需要的时候重新计算。这是默认的级别。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK</td> 
   <td>如果内存不够，会把剩余的分区存储在磁盘上。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_SER</td> 
   <td>存储序列化的Java对象。这一般比序列化空间效率高，不过读取的时候消耗CPU较多。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK_SER</td> 
   <td>和上一个相似，不过如果内存不够会存储到磁盘上，内存中存放序列化后的数据。</td> 
  </tr> 
  <tr> 
   <td>DISK_ONLY</td> 
   <td>仅存储在磁盘上。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_2，MEMORY_AND_DISK_2</td> 
   <td>和上一个相似，只不过会在集群中的两个节点上备份。</td> 
  </tr> 
  <tr> 
   <td>OFF_HEAP（实验中）</td> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

注意，在Python中，会始终序列化要存储的数据，所以持久化级别默认值就是以序列化后的对象存储在JVM堆空间中。

Spark会在混洗操作（比如reduceByKey）中自动持久化一些中间数据，尽管用户并没有调用persist。这就避免了在混洗过程中如果某个节点发生故障而重新计算整个数据集。我们仍然建议如果打算重复使用RDD就使用persist对其进行持久化。

### 4.4.1、选择哪个持久化级别呢？ ###

Spark的持久化级别是为了提供在内存使用和CPU效率间不同的平衡选择。我们建议通过以下步骤进行级别选择：

*  如果RDD适合默认的级别（MEMORY\_ONLY），那么就使用默认值。这是CPU效率最高的选项，使得对RDD的操作尽可能的快；
 *  如果RDD不适合MEMORY\_ONLY，尝试使用MEMORY\_ONLY\_SER，然后选择一个快的序列化库对数据进行序列化来高效使用空间，不过访问还是很快；
 *  除非计算数据集的函数非常耗时，或者这些函数过滤掉大多数的数据，否则不要将数据持久化到磁盘上。不然，重新计算数据可能会和从磁盘中读取一样快；
 *  如果想出错时尽快恢复，就是用备份。所有的级别都使用重新计算保证容错性，但备份级别可以保证程序继续执行而不用等待重新计算丢失的分区；
 *  在有大量内存空间和多应用程序的实验中，实验中的OFF\_HEAP模式有如下的优点：允许多个执行者能共享Tachyon的内存池；有效的降低了垃圾收集的消耗；如果单个执行者发生故障缓存的数据不会丢失；

### 4.4.2、删除数据 ###

Spark会自动跟踪每个节点的缓存使用情况，并且会根据最近最少使用原则（LRU）将最老的分区从内存中删除。如果想手动删除数据，使用unpersist方法。

# 5、共享变量 #

通常，当一个传递给一个Spark操作（比如map或reduce）的函数执行在远程的集群节点上时，它是对函数中使用的变量的另一份副本进行操作的。这些变量会被复制到每一个机器上，并且所有对这些变量的更新不会返回到驱动程序那里。在任务间支持通用的、读写共享的变量并不有效。然而，Spark提供两种常用形式的有限类型的共享变量：广播变量（broadcast）和累加器（accumulators）。

## 5.1、广播变量 ##

广播变量允许开发者在每个机器上缓存一个只读变量而不是把它在任务间复制。它们给每一个节点一个大规模数据的一个副本，并通过高效的方式完成。Spark也会试图使用更好的广播算法来分布式存储广播变量来减少网络流量。

Spark的行动操作在一系列阶段（stage）执行，通过混洗操作进行分割。Spark会自动广播每个阶段任务都需要的数据。这些数据以序列化的形式缓存然后再每个任务使用之前反序列化。这意味着只有当任务需要在跨多个阶段执行过程中使用同一个数据时，或者以反序列化缓存数据是重要的时候，显示广播数据才有用。

可以通过使用SparkContext.broadcast()方法来对变量v创建一个广播变量。广播变量将v包裹起来，可以通过调用value方法获取v。下面的代码演示了广播变量的用法：

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在广播之后不应该改变，这样所有的节点都会获得广播变量中相同的值。

## 5.2、累加器 ##

累加器就是在相应操作中只能增加的变量，因此能更有效的支持并行操作。累加器可以用来作为计数器和求和。Spark支持数字类型的累加器，开发者可以增加新类型的支持。如果累加器在创建时设置了一个名字，那么名字就会在Spark的UI中显示。这对于理解阶段的执行过程有一定的帮助。

可以通过调用SparkContext.accumulator(v)对变量v创建一个累加器。之后集群上的任务就可以通过add方法或+=操作符增加累加器。然而，任务并不能读取累加器的值。只用驱动程序才可以通过value方法读取累加器的值。

下面的代码演示了使用累加器计算数组的和：

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

上面的代码使用内置支持的Int类型创建了累加器，开发者可以通过实现AccumulatorParam接口来对自己的类型增加累加器支持。AccumulatorParam接口有两个方法：zero方法提供一个自己的类型的零值，addInPlace方法定义两个值相加的操作。比如，假设我们有一个可以代表数学上向量的类型Vector，可以这样写：

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)

Scala中，Spark也支持更常用的A吃醋姆拉不了接口来累加那些结果类型和元素类型不同的数据（比如通过收集数据构成一个list列表），还有一个SparkContext.accumulableCollection方法累加常见的Scala集合类型。

对累加器来说，更新总是在行动操作中执行，Spark保证每一个任务对累加器的更新只有一次，比如重启的任务不会更新这个值。在转换操作中，用户应该意识到如果任务或者job阶段重复执行，那每个任务的更新操作可能执行多次。

累加器并没有改变Spark的惰性求值策略。如果一个RDD的一个累加器被更新了，RDD在行动操作中累加器的值只更新一次。因此，当在惰性的转换操作比如map中，累加器的更新并不能保证会执行。下面的代码片段演示了这个属性：

val accum = sc.accumulator(0)
    data.map { x => accum += x; f(x) }
    // Here, accum is still 0 because no actions have caused the map to be computed.

# 6、部署到集群上 #

在应用提交指导中介绍如何提交应用到集群上。简单来说，一旦应用打包成jar文件，spark-submit可以将你的应用部署到任何集群上。

# 7、更多 #

在Spark的网站上有一些Spark应用的例子。而且，在Spark的examples目录下也有一些程序实例。你可以通过run-example脚本运行例子：

./bin/run-example SparkPi

[Intellij Idea_Spark]: http://blog.csdn.net/u012877472/article/details/51000690