Part 1

1. Spark计算模型

1.1 Spark程序模型

首先通过一个简单的实例了解Spark的程序模型。

1）SparkContext中的textFile函数从HDFS读取日志文件，输出变量file。

valfile=sc.textFile(“hdfs://xxx”)

2）RDD中的filter函数过滤带“ERROR”的行，输出errors（errors也是一个RDD）。

valerrors=file.filter(line=>line.contains(“ERROR”)

3）RDD的count函数返回“ERROR”的行数：

errors.count()。

其示意图如下：

从图中可以看到，RDD由很多partition组成，一个partition对应到物理模块上算是一个block，由block

manager统一管理。

Spark程序模型的主要思想是RDD（Resilient

Distributed Dataset），把所有计算的数据保存在分布式的内存中。在迭代计算中，通常情况下，都是对同一的数据集做反复的迭代计算，数据保存在内存中，将大大提高性能。RDD就是数据partition方式保存在cluster的内存中。操作有两种：transformation和action，transform就是把一种RDD转换为另一个RDD，和Hadoop的map操作很类似，只是定义operator比较丰富（map，join，filter，groupByKey等操作），action就类似于hadoop的reduce，其输出是一个aggregation函数的值如count，或者是一个集合（collection）。

1.2 弹性分布式数据集（RDD）

RDD是Spark的核心数据结构，通过RDD的依赖关系形成Spark的调度顺序。通过对RDD的操作形成整个Spark程序。

1.2.1 RDD的四种创建方式

1）从Hadoop文件系统（或与Hadoop兼容的其他持久化存储系统，如Hive、Cassandra、Hbase）输入（如HDFS）创建。

2）从父RDD转换得到新的RDD。

3）调用SparkContext方法的parallelize，将Driver上的数据集并行化，转化为分布式的RDD。

4）更改RDD的持久性（persistence），例如cache()函数，默认RDD计算后会在内存中清除。通过cache函数将计算后的RDD缓存在内存中。

1.2.2 RDD的两种操作算子

对于RDD可以有两种计算操作算子：Transformation（变换）与Action（行动）。

1）Transformation（变换）。Transformation操作是延迟计算的，也就是说从一个RDD转换生成另一个RDD的转换操作不是马上执行，需要等到有Actions操作时，才真正触发运算。

2）Action（行动）Action算子会触发Spark提交作业（Job），并将数据输出到Spark系统。

1.2.3 RDD的重要内部属性

1）分区列表。

2）计算每个分片的函数。

3）对父RDD的依赖列表。

4）对Key-Value对数据类型RDD的分区器，控制分区策略和分区数。

5）每个数据分区的地址列表（如HDFS上的数据块的地址）。

1.2.4 RDD与DSM的对比

通过上述表格可以看到，RDD有更好的容错性，采用血统机制后，可以不用回滚程序实现容错。

RDD和DSM对比主要有如下两个优势：

1）对于RDD中的批量操作，运行时将根据数据存放的位置来调度任务，从而提高性能。

2）对于扫描类型操作，如果内存不足以缓存整个RDD，就进行部分缓存，将内存容纳不下的分区存储到磁盘上。

1.3 Spark的数据存储

Spark数据存储的核心是弹性分布式数据集（RDD）。RDD可以被抽象地理解为一个大的数组（Array），但是这个数组是分布在集群上的。逻辑上RDD的每个分区叫一个Partition。

在Spark的执行过程中，RDD经历一个个的Transfomation算子之后，最后通过Action算子进行触发操作。

RDD之间通过Lineage产生依赖关系，这个关系在容错中有很重要的作用。

RDD会被划分成很多的分区分布到集群的多个节点中。分区是个逻辑概念，变换前后的新旧分区在物理上可能是同一块内存存储。这是很重要的优化，以防止函数式数据不变性导致的内存需求无限扩张。

在上图中，在物理上，RDD对象实质上是一个元数据结构，存储着Block、Node等的映射关系，以及其他的元数据信息。

每个Block中存储着RDD所有数据项的一个子集。

1.4 Spark的算子作用与分类

1.4.1 算子的作用

算子是RDD中定义的函数，可以对RDD中的数据进行转换和操作。

1）输入：在Spark程序运行中，数据从外部数据空间（如分布式存储：textFile读取HDFS等，parallelize方法输入Scala集合或数据）输入Spark，数据进入Spark运行时数据空间，转化为Spark中的数据块，通过BlockManager进行管理。

2）运行：在Spark数据输入形成RDD后便可以通过变换算子，如fliter等，对数据进行操作并将RDD转化为新的RDD，通过Action算子，触发Spark提交作业。如果数据需要复用，可以通过Cache算子，将数据缓存到内存。

3）输出：程序运行结束数据会输出Spark运行时空间，存储到分布式存储中（如saveAsTextFile输出到HDFS），或Scala数据或集合中（collect输出到Scala集合，count返回Scala int型数据）。

1.4.2 算子的分类

大致可以分为三大类算子。

1）Value数据类型的Transformation算子，这种变换并不触发提交作业，针对处理的数据项是Value型的数据。

2）Key-Value数据类型的Transfromation算子，这种变换并不触发提交作业，针对处理的数据项是Key-Value型的数据对。

3）Action算子，这类算子会触发SparkContext提交Job作业。

1.4.3 Value型的Transformation算子

这种算子可以根据RDD变换算子的输入分区与输出分区关系分为以下几种类型：

1）输入分区与输出分区一对一型。

2）输入分区与输出分区多对一型。

3）输入分区与输出分区多对多型。

4）输出分区为输入分区子集型。

5）还有一种特殊的输入与输出分区一对一的算子类型：Cache型。Cache算子对RDD分区进行缓存。

1.4.3.1 输入分区与输出分区一对一型：

(1)Map

源码中的map算子相当于初始化一个RDD，新RDD叫作

MappedRDD(this, sc.clean(f))。

(2)flatMap

将原来RDD中的每个元素通过函数f转换为新的元素，并将生成的RDD的每个集合中的元素合并为一个集合。内部创建FlatMappedRDD(this, sc.clean(f))。

(3)mapPartitions

mapPartitions函数获取到每个分区的迭代器，在函数中通过这个分区整体的迭代器对整个分区的元素进行操作。内部实现是生成MapPartitionsRDD。

(4)glom

glom函数将每个分区形成一个数组，内部实现是返回的GlommedRDD。

图3-7中的每个方框代表一个RDD分区。

1.4.3.2 输入分区与输出分区多对一型：

(1)union

使用union函数时需要保证两个RDD元素的数据类型相同，返回的RDD数据类型和被合并的RDD元素数据类型相同，并不进行去重操作，保存所有元素。如果想去重，可以使用distinct()。++符号相当于union函数操作。

(2)cartesian

对两个RDD内的所有元素进行笛卡尔积操作。操作后，内部实现返回CartesianRDD。

1.4.3.3 输入分区与输出分区多对多型：

groupBy：

将元素通过函数生成相应的Key，数据就转化为Key-Value格式，之后将Key相同的元素分为一组。

实现：

②sc.clean( )函数将用户函数预处理：valcleanF = sc.clean(f)

②对数据map进行函数操作，最后再对groupByKey进行分组操作。this.map(t => (cleanF(t), t)).groupByKey(p)

1.4.3.4 输出分区为输入分区子集型：

(1)filter

filter的功能是对元素进行过滤，对每个元素应用f函数，返回值为true的元素在RDD中保留，返回为false的将过滤掉。内部实现相当于生成FilteredRDD(this，sc.clean(f))。

deffilter(f:T=>Boolean):RDD[T]=new FilteredRDD(this,sc.clean(f))

(2)distinct

distinct将RDD中的元素进行去重操作。

(3)subtract

subtract相当于进行集合的差操作，RDD 1去除RDD 1和RDD 2交集中的所有元素。

(4)sample

sample将RDD这个集合内的元素进行采样，获取所有元素的子集。用户可以设定是否有放回的抽样、百分比、随机种子，进而决定采样方式。内部实现是生成

SampledRDD(withReplacement,fraction, seed)

函数参数设置如下:

withReplacement=true，表示有放回的抽样

withReplacement=false，表示无放回的抽样

(5)takesample

takeSample()函数和上面的sample函数是一个原理，但是不使用相对比例采样，而是按设定的采样个数进行采样，同时返回结果不再是RDD，而是相当于对采样后的数据进行Collect()，返回结果的集合为单机的数组。

1.4.3.5 Cache型:

(1)cache

cache将RDD元素从磁盘缓存到内存，相当于persist(MEMORY_ONLY)函数的功能。

(2)persist

persist函数对RDD进行缓存操作。数据缓存在哪里由StorageLevel枚举类型确定。

persist(newLevel:StorageLevel)

part 2

2. Spark I/O机制

2.1 序列化

2.1.1 序列化的含义和目的

含义：序列化是将对象转换为字节流，本质上可以理解为将链表存储的非连续空间的数据存储转化为连续空间存储的数组中。这样就可以将数据进行流式传输或者块存储。相反，反序列化就是将字节流转化为对象。

目的：进程间通信：不同节点之间进行数据传输。

数据持久化存储到磁盘：本地节点将对象写入磁盘。

2.1.2 两种序列化方式对比

Java序列化：在默认情况下，Spark采用Java的ObjectOutputStream序列化一个对象。该方式适用于所有实现了java.io.Serializable的类。Java序列化非常灵活，但是速度较慢，在某些情况下序列化的结果也比较大。

Kryo序列化：Spark也能使用Kryo（版本2）序列化对象。Kryo不但速度极快，而且产生的结果更为紧凑（通常能提高10倍）。Kryo的缺点是不支持所有类型，为了更好的性能，你需要提前注册程序中所使用的类（class）。

2.2 压缩

当大片连续区域进行数据存储并且存储区域中数据重复性高的状况下，数据适合进行压缩。

序列化后的数据可以压缩，使数据紧缩，减少空间开销。

2.2.1 两种压缩方式对比

Snappy的目标是在合理的压缩量的情况下，提高压缩速度，因此压缩比并不是很高。根据数据集的不同，压缩比能达到20%~100%。Snappy通常在达到相当压缩的情况下，要比同类的LZO、LZF、FastLZ和QuickLZ等快速的压缩算法快，LZF提供了更高的压缩比。

2.2.2 序列化与压缩

在分布式计算中，序列化和压缩是两个重要的手段。Spark通过序列化将链式分布的数据转化为连续分布的数据，这样就能够进行分布式的进程间数据通信，或者在内存进行数据压缩等操作，提升Spark的应用性能。通过压缩，能够减少数据的内存占用，以及IO和网络数据传输开销。

2.3 Spark块管理

整体的I/O管理分为以下两个层次：

1）通信层：I/O模块也是采用Master-Slave结构来实现通信层的架构，Master和Slave之间传输控制信息、状态信息。

2）存储层：Spark的块数据需要存储到内存或者磁盘，有可能还需传输到远端机器，这些是由存储层完成的。

2.3.1 实体与类

如图中所示，在Storage模块中，根据层次划分有如下模块：

（1）管理和接口

BlockManager：当其他模块要和storage模块进行交互时，storage模块提供了统一的操作类BlockManager，外部类与storage模块打交道都需要调用BlockManager相应接口来实现。

（2）通信层

·BlockManagerMasterActor：从主节点创建，从节点通过这个Actor的引用向主节点传递消息和状态。

·BlockManagerSlaveActor：在从节点创建，主节点通过这个Actor的引用向从节点传递命令，控制从节点的块读写。

·BlockManagerMaster：对Actor通信进行管理。

（3）数据读写层

·DiskStore：提供Block在磁盘上以文件形式读写的功能。

·MemoryStore：提供Block在内存中的Block读写功能。

·ConnectionManager：提供本地机器和远端节点进行网络传输Block的功能。

·BlockManagerWorker：对远端数据的异步传输进行管理。

整体的数据存储通信仍相当于Master-Slave模型，节点之间传递消息和状态，Master节点负责总体控制，Slave节点接收命令、汇报状态。

2.3.2 读写流程

（1）数据写入

1）RDD调用compute（）方法进行指定分区的写入。

2）CacheManager中调用BlockManager判断数据是否已经写入，如果未写则写入。

3）BlockManager中数据与其他节点同步。

4）BlockManager根据存储级别写入指定的存储层。

5）BlockManager向主节点汇报存储状态

（2）数据读取

在RDD类中，通过compute方法调用iterator读写某个分区（Partition），作为数据读取的入口。分区是逻辑概念，在物理上是一个数据块（block）。

（3）读取逻辑

通过BlockManager读取代码进入读取逻辑

1）本地读取。

在本地同步读取数据块，首先看能否在内存读取数据块，如果不能读取，则看能否从Tachyon读取数据块，如果仍不能读取，则看能否从磁盘读取数据块。

2）远程读取。

远程获取调用路径，然后getRemote调用doGetRemote，通过BlockManagerWorker.syncGetBlock从远程获取数据。

其中Tachyon是一个分布式内存文件系统，可以在集群里以访问内存的速度来访问存在tachyon里的文件。把Tachyon是架构在最底层的分布式文件存储和上层的各种计算框架之间的一种中间件。主要职责是将那些不需要落地到DFS里的文件，落地到分布式内存文件系统中，来达到共享内存，从而提高效率。同时可以减少内存冗余，GC时间等。

在BlockManagerWorker中调用syncGetBlock获取远端数据块，这里使用了Future模型。Future本身是一种被广泛运用的并发设计模式，可在很大程度上简化需要数据流同步的并发应用开发。

该模型是将异步请求和代理模式联合的模型产物。

客户端发送一个长时间的请求，服务端不需等待该数据处理完成便立即返回一个伪造的代理数据（相当于商品订单，不是商品本身），用户也无需等待，先去执行其他的若干操作后，再去调用服务器已经完成组装的真实数据。该模型充分利用了等待的时间片段。

2.3.3 数据块读写管理

数据块的读写，如果在本地内存存在所需数据块，则先从本地内存读取，如果不存在，则看本地的磁盘是否有数据，如果仍不存在，再看网络中其他节点上是否有数据，即数据有3个类别的读写来源。

（1）MemoryStore内存块读写

进行块读写是线程间同步的。通过entries.synchronized控制多线程并发读写，防止出现异常。

PutBlock对象用来确保只有一个线程写入数据块。这样确保数据读写且线程安全的。示例代码如下：

Private val putLock = new Object()

内存Block块管理是通过链表来实现的

（2）DiskStore磁盘块写入

在DiskStore中，一个Block对应一个文件。在diskManager中，存储blockId和一个文件路径映射。数据块的读写入相当于读写文件流。