弹性分布式数据集（RDDs）
Spark围绕着弹性分布式数据集（RDD）这个概念，RDD是具有容错机制的元素集合，可以并行操作。有两种方式创建RDDs：并行化驱动程序中已存在的集合，或者引用外部存储系统中的数据集，例如一个共享文件系统，HDFS，HBase，或者任何支持Hadoop输入格式的数据源。
并行集合
在驱动程序中已存在的集合上调用SparkContext的parallelize方法可创建并行集合。通过拷贝已存在集合中的元素来生成可并行操作的分布式数据集。下面是创建并行集合的例子：
val data = Array(1, 2, 3, 4, 5)val distData = sc.parallelize(data)
一旦创建完成，分布式数据集(distData)就可以被并行操作。例如，调用distData.reduce((a, b) => a + b)可以累加数组的元素。稍后介绍分布式数据集上的操作。
并行集合的一个重要参数是分区（partitions）的数量，用于切分数据集。Spark会为集群上的每个分区运行一个任务。通常你会想要为集群上的每个CPU分配2-4个分区。一般情况下，Spark会尝试根据集群自动设置分区的数量。当然，你可能想手动设置，通过传递parallelize方法的第二个参数（如sc.parallelize(data, 10)）可以实现。注意：有些代码为了向下兼容使用了术语slices（和partitions一个意思）。
外部数据集
Spark可以通过任何Hadoop支持的存储源创建分布式数据集，包括你本地的文件系统，HDFS， Cassandra，HBase，Amazon S3等等。Spark支持文本文件，SequenceFiles和任意其它Hadoop输入格式。
文本文件RDDs可使用SparkContext的textFile方法创建。这个方法需要文件的URI（一个本地机器上的路径，或者一个hdfs://, s3n://等），文件内容读取后是所有行的集合。下面是一个示例：
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节点上可以访问。可以拷贝本地文件到所有woker节点，也可以使用网络共享文件系统。
Spark中所有基于文件的输入方法，包括
textFile，都支持在目录，压缩文件和通配符。例如，可以使用textFile("/my/directory")，textFile("/my/directory/*.txt")和textFile("/my/directory/*.gz")。textFile方法还包含一个可选参数，用于控制文件的分区数量。默认情况下，Spark会为文件的每个block创建一个分区（HDFS默认的block大小是128MB），不过你可以通过可选参数设置更大的分区值。需要注意的是不能比blocks的数量还少。
除了文本文件，Spark的Scala API还支持多种数据格式：
SparkContext.wholeTextFiles可以读取包含多个小文本文件的目录，并且把每个文件作为(filename, content)返回。相比之下，textFile会把每个文件的每一行作为一条记录返回。对于SequenceFiles，使用
SparkContext的sequenceFile[K, V]方法，K和V是文件中的键值类型。K和V应该是Hadoop的Writable接口的子类，如IntWritable和Text。此外，Spark允许为一些常见的Writables指定原生类型；例如，sequenceFile[Int, String]会自动读取IntWritables和Texts。对于其它Hadoop输入格式，可以使用
SparkContext.hadoopRDD方法，以任意JobConf和输入格式类，key类和value类作为参数。和使用输入源设置Hadoop作业一样设置上述参数。对于基于新MapReduce API(org.apache.hadoop.mapreduce)的输入格式，可以使用SparkContext.newAPIHadoopRDD。RDD.saveAsObjectFile和SparkContext.objectFile支持将RDD保存到由序列化的Java对象组成的简单格式。虽然这不如专业格式Avro效率高，却提供了一种保存RDD的简单方式。
RDD操作
RDD支持两种操作：transformations（从一个已存在的数据集创建新的数据集）和actions（在数据集上进行计算并将结果返回给驱动程序）。例如，map是一个transformation，用于将数据集中的每个元素传递给一个函数并且返回一个新的RDD作为结果。reduce是一个action，它会用某个函数将RDD的所有元素聚合然后将最终结果返回给驱动程序（有一个reduceByKey返回分布式数据集）。
Spark的所有transformation都是lazy的，不会立刻计算结果。相反，只是记录应用到基础数据集（如文件）的transformation。只有action要返回结果到驱动程序时才会计算transformation。这样设计是为了Spark更高效。例如，map创建的数据集会在reduce中使用，之后会返回reduce的结果给驱动程序，而map创建的那个更大的数据集。
默认情况下，转换得到的RDD，每次要对其执行action的时候重新计算。可以使用persist（或cache）方法将RDD保存到内存中，Spark会将RDD的元素存储到集群中，下次查询的时候会更快。Spark也支持将RDD存储到磁盘上，或者跨多个节点复制。
基础
RDD的基本操作，如下：
val lines = sc.textFile("data.txt")val lineLengths = lines.map(s => s.length)val totalLength = lineLengths.reduce((a, b) => a + b)第一行从外部文件定义了一个基本RDD。这个数据集没有加载到内存，也没有做其它操作，lines仅仅是指向文件的指针。第二行定义了lineLengths作为map transformation的结果。lineLengths不是立即计算的，因为是懒加载的。最后，执行reduce action。这时候Spark会将计算分解成多个任务运行在独立的机器上，每个机器会执行一部分map和局部的reduce，然后返回结果到驱动程序。
如果之后还想用lineLengths，在reduce之前执行以下语句：
lineLengths.persist()
这样可以在第一次计算时将lineLengths保存到内存中。
函数传递到Spark
Spark的API很依赖给驱动程序传递函数来在集群上运行。有两种推荐方式：
匿名函数，用于短代码。
在全局单例对象中的静态方法。例如，定义了
object 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，其中的map会引用实例的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)写法等价。为了避免这个问题，最简单的方法是拷贝field到本地变量，不进行外部访问：
def doStuff(rdd: RDD[String]): RDD[String] = { val field_ = this.field
rdd.map(x => field_ + x)
}理解闭包
在集群上执行代码时，理解变量和方法的作用范围和生命周期是Spark的一个难点。RDD操作在作用范围之外修改变量是经常出现的问题。下面是一个foreach()增加计数的例子，相同的问题也会出现在其它操作当中。
示例
看下面RDD元素求和的示例，代码的行为会根据是否在同一个JVM上执行而有所不同。常见的例子是在local模式（--master = local[n]）下运行与部署在集群（spark-submit提交给YARN）上运行进行对比。
var counter = 0var rdd = sc.parallelize(data)// Wrong: Don't do this!!rdd.foreach(x => counter += x)
println("Counter value: " + counter)本地模式 vs. 集群模式
上面代码的行为是未定义的，可能无法按照预期工作。为执行作业，Spark会把RDD操作分拆成任务，每个任务由一个executor执行。执行之前，Spark会计算任务的闭包。闭包就是变量和方法（上面例子里是foreach()），它们对于在RDD上执行计算的executor是可以见的。这个闭包会被序列化并发送到每个executor。
发送到每个executor的闭包变量是一个拷贝，在foreach函数中引用counter时，已经不是驱动节点上的counter了。驱动节点的内存中仍然有counter，但是对于所有executor已经不可见了！所有executor只能看到序列化闭包中的拷贝。counter最后的值还是0，因为所有的操作都在序列化闭包内的counter上执行。
本地模式中，在某些情况下，foreach函数会和驱动程序在同一个JVM上执行，这样可以引用到原始的counter`并更新这个变量。
要保证有明确定义的行为，需要使用Accumulator
。当对变量的操作跨集群中的多个工作节点时，Accumulator提供一种安全更新变量的机制。后面介绍Accumulator时会详细说明。
通常来说，闭包—构建像循环或局部定义的方法，不应该用于改变全局状态。对于改变闭包外对象的行为，Spark没有定义也不提供保证。有些代码用本地模式执行，但那不是最常用的，这样的代码在分布式模式中不会按照预期执行。如果需要全局聚合，使用Accumulator。
打印RDD的元素
另外一个问题是用rdd.foreach(println)或rdd.map(println)打印RDD的元素。在一台机器上，可以保证正常打印所有RDD的元素。但是在集群模式中，executor使用的是自己的stdout，不是驱动节点上的，所以在驱动节点的stdout是看不到打印结果的！要在驱动节点上打印所有元素，可以使用collect()方法将所有元素放到驱动节点：rdd.collect().foreach(println)。这样做可能会导致驱动节点内存耗尽，因为collect()方法将整个RDD都放到一台机器上了；如果只想打印部分元素，使用take()是一种安全的方式：rdd.take(100).foreach(println)。
操作键值对
RDD的大部分操作都可以处理任意对象类型，不过有几个特殊操作只能操作键值对类型的RDD。最常用的就是分布式"shuffle"操作，如针对key进行分组或聚合元素。
在Scala中，这些操作对于包含Tuple2对象（语言内置的元组，可用(a, b)创建）的RDD自动可用。键值对操作在PairRDDFunctions类中可用，能够自动处理包含元组的RDD。
例如，下面代码在键值对上使用reduceByKey操作，计算每一行在文件中出现的次数：
val lines = sc.textFile("data.txt")val pairs = lines.map(s => (s, 1))val counts = pairs.reduceByKey((a, b) => a + b)也可使用counts.sortByKey()，按字母序排序，使用counts.collect()将结果当做对象数据放到驱动程序中。
注意：当在键值对操作中使用自定义对象作为key时，必须保证有自定义的equals()方法以及与之匹配的hashCode()方法。更多细节，请参见Object.hashCode() documentation。
Transformations
下面列出了常用的transformations。更多细节，请参见RDD API doc（Scala）和pair RDD functions doc（Scala）。
| Transformation | 描述 |
|---|---|
| map(func) | 通过将源数据的每个元素传递给func生成新的分布式数据集。 |
| filter(func) | 选择func返回true的源数据元素生成新的分布式数据集。 |
| flatMap(func) | 和map类似，但是每个输入项可对应0或多个输出项（func应该返回Seq而不是单一项）。 |
| 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) | 抽样一小部分数据，withReplacement可选，使用给定的随机数种子。 |
| union(otherDataset) | 返回新的数据集，包含源数据集和参数数据集元素的union。 |
| intersection(otherDataset) | 返回新的数据集，包含源数据集和参数数据集元素的intersection。 |
| distinct([numTasks])) | 返回新的数据集，包含源数据集中的不同元素。 |
| groupByKey([numTasks]) | 当在(K, V)数据集上调用时，返回(K, Iterable<V>)数据集。 注意：如果是为了执行聚合（如求和或求均值）进行分组，使用 reduceByKey和aggregateByKey会获得更好的性能。注意：默认地，输出的并行度取决于父RDD的分区数量。可以传递可选参数 numTasks设置不同的任务数量。 |
| reduceByKey(func, [numTasks]) | 当在(K, V)数据集上调用时，返回(K, V)数据集，每个key的值使用给定的函数func进行聚合，func必须是(V,V) => V类型。像groupByKey一样，reduce任务数量可通过第二个可选参数进行配置。 |
| aggregateByKey(zeroValue)(seqOp, combOp, [numTasks]) | 当在(K, V)数据集上调用时，返回(K, U)数据集，每个key的值使用给定的combine函数和"zero"值进行聚合。允许聚合值类型与输入值类型不同，同时避免不必要的分配。像groupByKey一样，任务数量可通过第二个可选参数进行配置。 |
| sortByKey([ascending], [numTasks]) | 当在(K, V)数据集上调用时，其中K是可排序的，返回按照key排序的(K, V)数据集，布尔参数ascending可指定升序或降序。 |
| join(otherDataset, [numTasks]) | 当在(K, V)和(K, W)数据集上调用时，返回(K, (V, W))数据集。支持外连接，leftOuterJoin， rightOuterJoin和fullOuterJoin。 |
| cogroup(otherDataset, [numTasks]) | 当在(K, V)和(K, W)数据集上调用时，返回(K, (Iterable<V>, Iterable<W>)) 数据集。这个操作也叫做groupWith。 |
| cartesian(otherDataset) | 当在类型T和U的数据集上调用时，返回(T, U)数据集。用于过滤大数据集后更有效地执行操作。 |
| pipe(command, [envVars]) | 通过shell命令将RDD以管道的方式处理每个分区，如Perl或bash脚本。RDD元素会被写入进程的stdin并且作为字符串类型的RDD返回，按行输出到stdout。 |
| coalesce(numPartitions) | 将RDD的分区数量减少到numPartitions。 |
| repartition(numPartitions) | 随机Reshuffle RDD中的数据来创建更多或更少的分区并进行平衡。总是在网络上shuffle所有数据。 |
| repartitionAndSortWithinPartitions(partitioner) | 根据给定的partitioner对RDD重新分区，在每个结果分区中，根据key进行排序。这个方法比repartition更高效，并且可以对每个分区进行排序，因为它会将排序放到shuffle machinery。 |
Actions
下面的表列出了常用的actions。更多细节，请参见RDD API doc（Scala）和pair RDD functions doc（Scala）。
| Action | 描述 |
|---|---|
| reduce(func) | 用func函数（需要两个参数，返回一个值）聚合数据集的元素。func函数是可交换可结合的，这样可以正确进行并行计算。 |
| collect() | 将数据集元素作为数据返回到驱动程序中。这个方法通常用于过滤器或其它操作返回的足够小的数据子集。 |
| count() | 返回数据集中元素的数量。 |
| first() | 返回数据集中的第一个元素。（take(1)类似） |
| take(n) | 返回一个数组，包含数据集中的前n个元素。 |
| takeSample(withReplacement, num, [seed]) | 返回num个随机抽样的元素组成的数组，withReplacement可选，可指定随机数生成器的种子。 |
| takeOrdered(n, [ordering] | 返回RDD的前n个元素，使用自然顺序或者自定义比较器。 |
| saveAsTextFile(path) | 将数据集作为文本文件（或文本文件集合）写入到本地文件系统的指定目录，HDFS，或者任何其它Hadoop支持的文件系统。Spark会对每个元素调用toString将其转换成文件中的一行文本。 |
| saveAsSequenceFile(path) (Java and Scala) | 将数据集作为Hadoop SequenceFile写入到本地文件系统的指定目录，HDFS，或者任何其它Hadoop支持的文件系统。在实现了Hadoop Writable接口的键值对类型的RDD上可用。在Scala中，对于可隐式转换为Writable的类型也可用（Spark包含对基本类似的转换，如Int，Double，String等） |
| saveAsObjectFile(path) (Java and Scala) | 使用Java序列化将数据集的元素写入一种简单格式，可使用SparkContext.objectFile()加载。 |
| countByKey() | 只在(K, V)类型的RDD上可用。返回hashmap (K, Int)，Int只每个key的数量。 |
| foreach(func) | 在数据集的每个元素上执行func函数。这个方法通常用于更新Accumulator或者与外部存储系统交互。 注意：修改 foreach()外部Accumulator以外的变量可能会导致未定义的行为。前面闭包里面说过。 |
Spark RDD API也暴露了一些action的异步版本，如foreachAsync，立刻返回FutureAction给调用者，不会阻塞在action的计算上。这类方法用于管理或等待action的异步执行。通常需要在executor和机器之间拷贝数据，shuffle是一个复杂耗时的操作。
Shuffle操作
Spark中的一些操作会触发被称为shuffle的事件。shuffle是Spark重新分配数据的机制，让数据在分区间有不同的分组。
背景
想要理解shuffle的细节，可参见reduceByKey操作。reduceByKey操作生成了一个新RDD，单个key的所有值都放到了元组（包含了key和这个key相关的所有值执行reduce函数后的结果）中。面临的问题是，单个key的所有值不是一定放在同一个分区或者同一台机器中，但是这些值需要一起计算结果。
在Spark中，数据通常不会为特定操作跨分区分布在需要的位置上。在计算时。单个任务在单个分区上执行—这样，为了组织单个reduceByKey的reduce任务需要的所有数据，Spark需要执行一个all-to-all操作。需要从所有分区上找出所有key的所有值，然后将每个key的值跨分区集合起来结算处最后的结果—这就是shuffle。
虽然shuffle之后每个分区的元素集合是确定的，分区的顺序也是确定，但是元素是无序的。如果想要在shuffle之后得到有序数据，可使用：
mapPartitions，使用.sorted给每个分区排序repartitionAndSortWithinPartitions，在重新分区的同时高效地排序sortBy，生成全局排序的RDD
会进行shuffle的操作包括repartition操作，如repartition和coalesce，'ByKey操作（除了计数），如groupByKey和reduceByKey，join操作，如cogroup和join。
性能影响
Shuffle是非常耗时的操作，因为需要磁盘I/O，数据序列化，以及网络I/O。为了shuffle组织数据，Spark生成了任务集合，map任务集合负责祖师数据，reduce任务集合负责聚合数据。这个命名方式来自MapReduce，和Spark的map和reduce操作没有直接关系。
单个map任务的结果一直放在内存中直到放不下为止。然后，这些结果会根据目标分区进行排序并写到单个文件中。reduce任务会读取相关的已排序的块。
一些shuffle操作会消耗大量堆内存，因为它们在转换前后使用内存数据结构来组织记录。特别地，reduceByKey和aggregateByKey在map时创建这些数据结构，'ByKey操作在reduce时生成这些结构。当数据不适合放在内存中时，Spark会将这些表拆分到磁盘中，这样会导致额外的磁盘I/O开销以及增加GC。
shuffle也会在磁盘上生成大量中间文件。Spark 1.3，这些文件会保留到对应的RDD不再使用并且已经被回收。这样做的话，在重新计算时shuffle文件不需要重新创建。如果应用程序一直保留这些RDD的引用或者GC不频繁，那么shuffle文件可能会很长时间之后才会回收。这就意味着长时间运行的Spark作业可能会消耗大量磁盘空间。在配置Spark Context时，spark.local.dir配置参数用于指定临时存储目录。
shuffle的行为可通过很多配置参数进行调整。具体参见Spark Configuration Guide中的‘Shuffle Behavior’。
RDD持久化
Spark最重要的功能之一就是在内存中跨操作持久化（或缓存）数据集。当持久化RDD时，每个节点会将其要计算的分区存储到内存中，并且在数据集上进行其它action操作时重用内存中的数据。这样之后的action操作可以执行得更快（通常超过10x）。缓存是迭代算法和快速交互的重要工具。
可使用persist()或cache()方法将RDD标记为持久化。第一在action中进行计算时，持久化的RDD会保存到节点内存中。Spark的缓存是具有容错机制的—如果RDD的任意分区丢失了，会使用最初创建它的transformations自动重新计算。
另外，每个持久化的RDD可使用不同的存储级别进行存储，例如，持久化数据集到磁盘，作为序列化的Java对象持久化到内存，跨节点复制。这些等级通过给persist()传递StorageLevel对象（Scala）进行设置。cache()方法使用默认存储等级，即torageLevel.MEMORY_ONLY（在内存中存储反序列化对象）。存储等级如下：
| 存储等级 | 描述 |
|---|---|
| MEMORY_ONLY | 在JVM中以反序列化的Java对象存储RDD。如果RDD无法完整存储到内存，一些分区就不会缓存，每次需要的时候重新计算。这是默认级别。 |
| MEMORY_AND_DISK | 在JVM中以反序列化的Java对象存储RDD。如果RDD无法完整存储到内存，无法存储到内存的分区会放到磁盘上，需要的时候从磁盘读取。 |
| MEMORY_ONLY_SER (Java and Scala) | 以序列化的Java对象存储RDD。通常这种方式比反序列化对象更节省空间，尤其是使用fast serializer，但是在读取时需要消耗更多CPU。 |
| MEMORY_AND_DISK_SER (Java and Scala) | 和MEMORY_ONLY_SER类似，但是无法存到内存的分区会放到磁盘，不会在需要时从新计算。 |
| DISK_ONLY | 只将RDD分区存储到磁盘。 |
| MEMORY_ONLY_2, MEMORY_AND_DISK_2等 | 和前面的等级一样，不过每个分区会复制到两个集群节点上。 |
| OFF_HEAP (experimental) | MEMORY_ONLY_SER类似，但是将数据存储到off-heap memory。需要启用off-heap内存。 |
注意：在Python中，使用Pickle库保存的对象永远都是序列化的，所以是否选择序列化等级都没关系。Python可用的存储等级包括MEMORY_ONLY，MEMORY_ONLY_2，MEMORY_AND_DISK，MEMORY_AND_DISK_2，DISK_ONLY和DISK_ONLY_2。
Spark会自动持久化shuffle操作的中间数据（如reduceByKey），甚至不需要用户调用persist。这样做是为了防止shuffle期间如果有节点出错了需要重新计算整个输入。如果想要重用RDD，建议用户手动调用persist。
如何选择存储等级
Spark的存储等级提供了在内存使用和CPU效率之间的不同权衡方案。推荐按照下面方法选择存储等级：
如果RDD内存适合使用默认存储等级（
MEMORY_ONLY），那就选择默认存储等级。这种方式是CPU效率最高的，能够让RDD上的操作尽可能快递执行。如果不适合，使用
MEMORY_ONLY_SER并且选择一个快的序列化库让对象存储更节省空间，但是仍然可以合理地快速访问。除非计算数据集非常耗时，或者它们过滤掉了大量数据，否则不要将数据集放到磁盘。不然的话，重新计算分区可能和从磁盘读取是一样快的。
如果想要快速进行错误恢复，使用复制存储等级（例如，如果使用Spark服务web应用请求）。所有存储等级通过重新计算丢失数据提供全容错机制，但是复制存储等级可以让任务继续在RDD上执行，不需要等着丢失分区重新计算完成。
删除数据
Spark会自动监控每个节点上的缓存使用情况，使用LRU将老数据分区清理掉。如果想要手动删除RDD，使用RDD.unpersist()方法。
作者：steanxy
链接：https://www.jianshu.com/p/92c31d9b562d