Spark详解（三）：Spark编程模型（RDD概述）

1. RDD概述

RDD是Spark的最基本抽象,是对分布式内存的抽象使用，实现了以操作本地集合的方式来操作分布式数据集的抽象实现。RDD是Spark最核心的东西，它表示已被分区，不可变的并能够被并行操作的数据集合，不同的数据集格式对应不同的RDD实现。RDD必须是可序列化的。RDD可以cache到内存中，每次对RDD数据集的操作之后的结果，都可以存放到内存中，下一个操作可以直接从内存中输入，省去了MapReduce大量的磁盘IO操作。这对于迭代运算比较常见的机器学习算法, 交互式数据挖掘来说，效率提升非常大。

RDD 最适合那种在数据集上的所有元素都执行相同操作的批处理式应用。在这种情况下， RDD 只需记录血统中每个转换就能还原丢失的数据分区，而无需记录大量的数据操作日志。所以 RDD 不适合那些需要异步、细粒度更新状态的应用 ，比如 Web 应用的存储系统，或增量式的 Web 爬虫等。对于这些应用，使用具有事务更新日志和数据检查点的数据库系统更为高效。

RDD实际上不存储数据，这里方便理解，暂时理解为存储数据

1.1 RDD简介

RDD实现了多种模型计算，这些模型包括以下几个方面的内容

• 迭代计算
• 交互式SQL查询
• MapReduceRDDD：通过MapReduce的超集，能够高效地执行MapReduce程序
• 流式数据处理：Spark提出了离散数据流（DStream）来解决这样的问题，DStream把流式计算的执行当做一系列短而确定的批量计算的序列，并将状态保存到RDD中。DStream根据相关RDD的依赖关系图进行并行化恢复，可以达到快速恢复故障，避免了数据复制。另外，通过推测执行来对Straggler迁移执行，例如，对于慢任务运行经过推测的备份副本。

1.2 RDD的特征

1. 来源：一种是从持久存储获取数据，另一种是从其他RDD生成
2. 只读：状态不可变，不能修改
3. 分区：支持元素根据 Key 来分区 ( Partitioning ) ，保存到多个结点上，还原时只会重新计算丢失分区的数据，而不会影响整个系统
4. 路径：在 RDD 中叫世族或血统 ( lineage ) ，即 RDD 有充足的信息关于它是如何从其他 RDD 产生而来的
5. 持久化：可以控制存储级别（内存、磁盘等）来进行持久化
6. 操作：丰富的动作 ( Action ) ，如Count、Reduce、Collect和Save 等

1.3 RDD的五大特征

1. RDD是由一系列的partition组成的
2. 函数是作用在每一个partition（split）上的。
3. RDD之间有一系列的依赖关系。
4. 分区器是作用在K,V格式的RDD上。
5. RDD提供一系列最佳的计算位置。

1.4 RDD的类型

Spark编程中开发者需要编写一个驱动程序（DriverProgram）来连接到工作进程（Worker）。驱动程序定义一个或者多个RDD以及相关的操作，驱动程序同时记录RDD的继承关系，即“血统”。而工作进程（Worker）是一种运行的进程，它经过一系列操作后的RDD分区数据保存在内存中。

Spark的操作大致上可以分为以下四类操作

• Create Operation：用于RDD的创建工作
• Transformation Operation：将RDD通过一定的操作转化为新的RDD
• Control Operation：进行RDD的持久化操作，可以将RDD按照不同的存储策略保存在磁盘或者内存中
• Action Operation：能触发Spark运行的操作，例如，对RDD进行collect就是Action操作。

什么是K,V格式的RDD?

如果RDD里面存储的数据都是二元组对象，那么这个RDD我们就叫做K,V格式的RDD。

哪里体现RDD的弹性（容错）？

partition数量，大小没有限制，体现了RDD的弹性。
RDD之间依赖关系，可以基于上一个RDD重新计算出RDD。

哪里体现RDD的分布式？

RDD是由Partition组成，partition是分布在不同节点上的。RDD提供计算最佳位置，体现了数据本地化。体现了大数据中“计算移动数据不移动”的理念。

2. RDD的实现

2.1 作业调度

对于款依赖的操作，在Spark将中间结果物化到父分区的节点上，这里和MapReduce中的物化Map的输出类似，可以简化数据的故障恢复过程。

对于执行失败的任务，只要它对应调度阶段父类信息可以用，该任务就会分散到其他节点上重新执行。如果某些调度阶段不可用（例如，因为Shuffle在map节点的输出丢失了），则重新提交任务，并以并行计算的方式计算丢失的分区。

2.2 解析器集成

Spark提供了一种交互式的Shell（解析器），借助内存数据的低延迟性，可以让用户通过解析器对大数据进行交互式查询。

2.3 内存管理

Spark提供了3种持久化RDD的存储策略：为序列化Java对象存在内存中、序列化的数据存于内存以及存储在磁盘中。

• 第一个选项的性能是最优的，因为可以直接访问在Java虚拟机内存里的RDD对象
• 在空间有限的情况下，第二种方式可以让用户采用比Java对象更加有效的内存组织方式，但是代价是降低了性能。
• 第三种策略使用于RDD太大的情形，每次重新计算该RDD都会带来额外的资源开销（I/O）

对于内存使用LRU回收算法来间来进行管理，当计算得到一个新的RDD分区，但没有足够的空间来存储的时候，系统会从最近最少未使用的RDD中回收其中一个分区的空间。

2.4 检查点支持

虽然“血统”可以用于错误的RDD的恢复，但是对于很长的“血统”的RDD来说，这样恢复的耗时很长，因此需要通过检查点的方式（CheckPoint）保存到外部存储当中。

通常情况下，对于包含宽依赖的长”血统“的RDD设置检查点是非常有效的。

2.5 多用户管理

• 在每个应用程序中，Spark运行多线程同时提交作业，并通过一种等级公平调度器来实现多个作业集群资源的共享，这种调度器和Hadoop Fair Scheduler类似。该算法主要用于创建基于相同内存数据的多用户应用
• Spark的公平调度也使用延迟调度，通过轮询每台机器的数据，在保持公平的情况下给予作业高的本地性。
• 由于任务相对独立，调度器还支持取消作业来为高优先级作业腾出资源

3. RDD编程接口

3.1 RDD分区（partitions）

一个RDD划分成很多的分区（partition）分布在集群的节点中，分区的多少涉及对这个RDD进行并行计算的粒度。在RDD操作中用户可以使用partitions方法获取RDD划分的分区数，当然用户也可以设定分区数目。如果没有指定将使用默认值，而默认数值是该程序所分配到CPU核数，如果是从HDFS文件创建，默认为文件的block数（有一点我们必须要注意，当我们显示的设置分区数时，分区数不允许小于HDFS文件的block数）。

// 使用textFile方法获取指定路径的文件，未设置分区数
val rdd = sc.textFile("/app/spark/workcount.txt")
// 使用partitions方法获取分区数，假设默认的分区数为2，那么将返回2
val partitionSize = rdd.partitions.size
 
// 显示地设置RDD为6个分区
rdd = sc.textFile("/app/spark/wordcount.txt", 6)
// 获取分区数，此时返回6
partitionSize = rdd.partitions.size

3.2 RDD首选位置（preferredlocations）

在Spark形成任务有向无环图（DAG）时，会尽可能地把计算分配到靠近数据的位置，减少数据网络传输。当RDD产生的时候存在首选位置，如HadoopRDD分区的首选位置就是HDFS块所在的节点。当RDD分区被缓存，则计算应该发送到缓存分区所在的节点进行，再不然回溯RDD的血统，一直找到具有首选位置属性的父RDD，并据此决定子RDD的位置。

3.3 RDD依赖关系(dependencies)

dependencies顾名思义就是依赖的意思，由于RDD是粗粒度的操作数据集，每一个转换操作都会生成一个新的RDD，所以RDD之间就会形成类似于流水一样的前后依赖关系，在spark中存在两种类型的依赖，即窄依赖(Narrow dependencies)和宽依赖(wide dependencies)

• 窄依赖：每一个父RDD的分区最多只被子RDD的一个分区所使用
• 宽依赖：多个子RDD的分区会依赖于同一个父RDD的分区，

一：窄依赖可以在集群的一个节点上如流水线一般的执行，可以计算所有父RDD的分区，相反的，宽依赖需要取得父RDD的所有分区上的数据进行计算，将会执行类似于MapReduce一样的shuffle操作，二：对于窄依赖来说，节点计算失败后的恢复会更加有效，只需要重新计算对应的父RDD的分区，而且可以在其他的节点上并行地计算，相反的，在有宽依赖的继承关系中，一个节点的失败将会导致其父RDD的多个分区重新计算，这个代价是非常高的

3.4 RDD分区计算

Spark中RDD的计算操作都是以分区为单位的，而且计算函数都是在对迭代器复合，不需要保存每次计算的结果。分区计算一般是使用mapPartitions等操作来进行的，mapPartitions的输入函数是应用于每个分区，也就是把每个分区的内容作为整体来处理的。

3.5 RDD分区函数（Partitioner）

分区划分对于Shuffle类操作很关键，它决定了该操作的父RDD和子RDD之间的依赖类型。例如Join操作，如果协同划分的话，两个父RDD之间、父RDD和子RDD之间形成一致的分区安排，即同一个Key保证被映射到同一个分区，这样就能形成窄依赖。反正如果不是协同划分，导致宽依赖，这里所说的协同划分是指的是指分区划分器以产生前后一致的分区安排。

在Spark默认提供的两种划分器：哈希分区划分器（HashPartitioner）和范围分区划分器（RangePartitioner），其Partitioner只存在于（K，V）类型的RDD。

4. 创建操作

目前有两种类型的RDD，一种是并行集合（Prallelized Collections），接受到一个已经存在的Scala集合，然后进行并行计算；另外一种是从外部存储创建RDD，外部存储可以是文本文件或Hadoop文件系统HDFS，还可以使从Hadoop接口API创建。

4.1 并行化集合创建操作

例如：val rdd = sc.parallelize(Array(1 to 10)) 根据能启动的executor的数量来进行切分多个slice，每一个slice启动一个Task来进行处理。

val rdd = sc.parallelize(Array(1 to 10), 5) 指定了partition的数量

4.2 外部存储创建RDD

Spark可以将任何Hadoop所支持的存储资源转化成RDD,如本地文件（需要网络文件系统，所有的节点都必须能访问到）、HDFS、Cassandra、HBase、Amazon S3等，Spark支持文本文件、SequenceFiles和任何Hadoop InputFormat格式。

（1）使用textFile()方法可以将本地文件或HDFS文件转换成RDD

支持整个文件目录读取，文件可以是文本或者压缩文件(如gzip等，自动执行解压缩并加载数据)。如textFile（”file:///dfs/data”）

支持通配符读取,例如：

val rdd1 = sc.textFile("file:///root/access_log/access_log*.filter");

val rdd2=rdd1.map(_.split("t")).filter(_.length==6)

rdd2.count()

......

14/08/20 14:44:48 INFO HadoopRDD: Input split: file:/root/access_log/access_log.20080611.decode.filter:134217728+20705903

......

textFile()可选第二个参数slice，默认情况下为每一个block分配一个slice。用户也可以通过slice指定更多的分片，但不能使用少于HDFS block的分片数。

（2）使用wholeTextFiles()读取目录里面的小文件，返回（用户名、内容）对

（3）使用sequenceFileK,V方法可以将SequenceFile转换成RDD。SequenceFile文件是Hadoop用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。

（4）使用SparkContext.hadoopRDD方法可以将其他任何Hadoop输入类型转化成RDD使用方法。一般来说，HadoopRDD中每一个HDFS block都成为一个RDD分区。

此外，通过Transformation可以将HadoopRDD等转换成FilterRDD(依赖一个父RDD产生）和JoinedRDD（依赖所有父RDD）等。

5. 转换操作

转换(Transformations) (如：map, filter, groupBy, join等)，Transformations操作是Lazy的，也就是说从一个RDD转换生成另一个RDD的操作不是马上执行，Spark在遇到Transformations操作时只会记录需要这样的操作，并不会去执行，需要等到有Actions操作的时候才会真正启动计算过程进行计算。

Transformations操作	内容
map(func)	返回一个新的分布式数据集，由每个原元素经过func函数转换后组成
filter(func)	返回一个新的数据集，由经过func函数后返回值为true的原元素组成
flatMap(func)	类似于map，但是每一个输入元素，会被映射为0到多个输出元素（因此，func函数的返回值是一个Seq，而不是单一元素）
sample(withReplacement, frac, seed)	根据给定的随机种子seed，随机抽样出数量为frac的数据
union(otherDataset)	返回一个新的数据集，由原数据集和参数联合而成
groupByKey([numTasks])	在一个由（K,V）对组成的数据集上调用，返回一个（K，Seq[V])对的数据集。注意：默认情况下，使用8个并行任务进行分组，你可以传入numTask可选参数，根据数据量设置不同数目的Task
reduceByKey(func, [numTasks])	在一个（K，V)对的数据集上使用，返回一个（K，V）对的数据集，key相同的值，都被使用指定的reduce函数聚合到一起。和groupbykey类似，任务的个数是可以通过第二个可选参数来配置的。
join(otherDataset, [numTasks])	在类型为（K,V)和（K,W)类型的数据集上调用，返回一个（K,(V,W))对，每个key中的所有元素都在一起的数据集
groupWith(otherDataset, [numTasks])	在类型为（K,V)和(K,W)类型的数据集上调用，返回一个数据集，组成元素为（K, Seq[V], Seq[W]) Tuples。这个操作在其它框架，称为CoGroup
cartesian(otherDataset)	笛卡尔积。但在数据集T和U上调用时，返回一个(T，U）对的数据集，所有元素交互进行笛卡尔积。
flatMap(func)	类似于map，但是每一个输入元素，会被映射为0到多个输出元素（因此，func函数的返回值是一个Seq，而不是单一元素）

6. 行动Action操作

操作(Actions) (如：count, collect, save等)，Actions操作会返回结果或把RDD数据写到存储系统中。Actions是触发Spark启动计算的动因。

Action操作	内容
reduce(func)	通过函数func聚集数据集中的所有元素。Func函数接受2个参数，返回一个值。这个函数必须是关联性的，确保可以被正确的并发执行
count()	返回数据集的元素个数
take(n)	返回一个数组，由数据集的前n个元素组成。注意，这个操作目前并非在多个节点上，并行执行，而是Driver程序所在机器，单机计算所有的元素(Gateway的内存压力会增大，需要谨慎使用）
first()	返回数据集的第一个元素（类似于take（1）
saveAsTextFile(path)	将数据集的元素，以textfile的形式，保存到本地文件系统，hdfs或者任何其它hadoop支持的文件系统。Spark将会调用每个元素的toString方法，并将它转换为文件中的一行文本
saveAsSequenceFile(path)	将数据集的元素，以sequencefile的格式，保存到指定的目录下，本地系统，hdfs或者任何其它hadoop支持的文件系统。RDD的元素必须由key-value对组成，并都实现了Hadoop的Writable接口，或隐式可以转换为Writable（Spark包括了基本类型的转换，例如Int，Double，String等等）
foreach(func)	在数据集的每一个元素上，运行函数func。这通常用于更新一个累加器变量，或者和外部存储系统做交互

7. 控制操作

Spark可以将RDD持久化到内存或者磁盘文件系统中，把RDD持久化到内存中可以极大地提高迭代计算以及计算模型之间的数据共享，一般情况下执行节点的60%内存用于缓存数据，剩下的40%用于运行任务。

cache():RDD[T]
persists():RDD[T]
persist(level:StorageLevel):RDD[T]

Spark中，RDD类可以使用cache() 和 persist() 方法来缓存。cache()是persist()的特例，将该RDD缓存到内存中。而persist可以指定一个StorageLevel。StorageLevel的列表可以在StorageLevel 伴生单例对象中找到：

object StorageLevel {

  val NONE = new StorageLevel(false, false, false, false)

  val DISK_ONLY = new StorageLevel(true, false, false, false)

  val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)

  val MEMORY_ONLY = new StorageLevel(false, true, false, true)

  val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)

  val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)

  val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)

  val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)

  val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)

  val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)

  val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)

  val OFF_HEAP = new StorageLevel(false, false, true, false) // Tachyon

}

// 其中，StorageLevel 类的构造器参数如下：

class StorageLevel private(  private var useDisk_ : Boolean,  private var useMemory_ : Boolean,  private var useOf

Spark的不同StorageLevel ，目的满足内存使用和CPU效率权衡上的不同需求。我们建议通过以下的步骤来进行选择：

• 如果你的RDDs可以很好的与默认的存储级别(MEMORY_ONLY)契合，就不需要做任何修改了。这已经是CPU使用效率最高的选项，它使得RDDs的操作尽可能的快；
• 如果不行，试着使用MEMORY_ONLY_SER并且选择一个快速序列化的库使得对象在有比较高的空间使用率的情况下，依然可以较快被访问；
• 尽可能不要存储到硬盘上，除非计算数据集的函数，计算量特别大，或者它们过滤了大量的数据。否则，重新计算一个分区的速度，和与从硬盘中读取基本差不多快；
• 如果你想有快速故障恢复能力，使用复制存储级别(例如：用Spark来响应web应用的请求)。所有的存储级别都有通过重新计算丢失数据恢复错误的容错机制，但是复制存储级别可以让你在RDD上持续的运行任务，而不需要等待丢失的分区被重新计算；
• 如果你想要定义你自己的存储级别(比如复制因子为3而不是2)，可以使用StorageLevel 单例对象的apply()方法；
• 在不使用cached RDD的时候，及时使用unpersist方法来释放它。