3.2 弹性分布式数据集

3.2 弹性分布式数据集

本节简单介绍RDD，并介绍RDD与分布式共享内存的异同。

3.2.1 RDD简介

在集群背后，有一个非常重要的分布式数据架构，即弹性分布式数据集（resilient distributed dataset，RDD），它是逻辑集中的实体，在集群中的多台机器上进行了数据分区。通过对多台机器上不同RDD分区的控制，就能够减少机器之间的数据重排（data shuffling）。Spark提供了“partitionBy”运算符，能够通过集群中多台机器之间对原始RDD进行数据再分配来创建一个新的RDD。RDD是Spark的核心数据结构，通过RDD的依赖关系形成Spark的调度顺序。通过对RDD的操作形成整个Spark程序。

（1）RDD的两种创建方式

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

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

（2）RDD的两种操作算子

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

1）Transformation（变换）。

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

2）Action（行动）

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

（3）RDD的重要内部属性

1）分区列表。

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

3）对父RDD的依赖列表。

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

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

3.2.2 RDD与分布式共享内存的异同

RDD是一种分布式的内存抽象，表3-1列出了RDD与分布式共享内存（Distributed Shared Memory，DSM）的对比。在DSM系统[插图]中，应用可以向全局地址空间的任意位置进行读写操作。DSM是一种通用的内存数据抽象，但这种通用性同时也使其在商用集群上实现有效的容错性和一致性更加困难。

RDD与DSM主要区别在于[插图]，不仅可以通过批量转换创建（即“写”）RDD，还可以对任意内存位置读写。RDD限制应用执行批量写操作，这样有利于实现有效的容错。特别是，由于RDD可以使用Lineage（血统）来恢复分区，基本没有检查点开销。失效时只需要重新计算丢失的那些RDD分区，就可以在不同节点上并行执行，而不需要回滚（Roll Back）整个程序。

表3-1 RDD与DSM的对比

[插图]

通过备份任务的复制，RDD还可以处理落后任务（即运行很慢的节点），这点与MapReduce类似，DSM则难以实现备份任务，因为任务及其副本均需读写同一个内存位置的数据。

与DSM相比，RDD模型有两个优势。第一，对于RDD中的批量操作，运行时将根据数据存放的位置来调度任务，从而提高性能。第二，对于扫描类型操作，如果内存不足以缓存整个RDD，就进行部分缓存，将内存容纳不下的分区存储到磁盘上。

另外，RDD支持粗粒度和细粒度的读操作。RDD上的很多函数操作（如count和collect等）都是批量读操作，即扫描整个数据集，可以将任务分配到距离数据最近的节点上。同时，RDD也支持细粒度操作，即在哈希或范围分区的RDD上执行关键字查找。

后续将算子从两个维度结合在3.3节对RDD算子进行详细介绍。

1）Transformations（变换）和Action（行动）算子维度。

2）在Transformations算子中再将数据类型维度细分为：Value数据类型和Key-Value对数据类型的Transformations算子。Value型数据的算子封装在RDD类中可以直接使用，Key-Value 对数据类型的算子封装于PairRDDFunctions类中，用户需要引入import org.apache.spark.SparkContext._才能够使用。进行这样的细分是由于不同的数据类型处理思想不太一样，同时有些算子是不同的。

3.2.3 Spark的数据存储

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

在Spark的执行过程中，RDD经历一个个的Transfomation算子之后，最后通过Action算子进行触发操作。逻辑上每经历一次变换，就会将RDD转换为一个新的RDD，RDD之间通过Lineage产生依赖关系，这个关系在容错中有很重要的作用。变换的输入和输出都是RDD。RDD会被划分成很多的分区分布到集群的多个节点中。分区是个逻辑概念，变换前后的新旧分区在物理上可能是同一块内存存储。这是很重要的优化，以防止函数式数据不变性（immutable）导致的内存需求无限扩张。有些RDD是计算的中间结果，其分区并不一定有相应的内存或磁盘数据与之对应，如果要迭代使用数据，可以调cache()函数缓存数据。

图3-2为RDD的数据存储模型。

[插图]

图3-2 RDD数据管理模型

图3-2中的RDD_1含有5个分区（p1、p2、p3、p4、p5），分别存储在4个节点（Node1、node2、Node3、Node4）中。RDD_2含有3个分区（p1、p2、p3），分布在3个节点（Node1、Node2、Node3）中。

在物理上，RDD对象实质上是一个元数据结构，存储着Block、Node等的映射关系，以及其他的元数据信息。一个RDD就是一组分区，在物理数据存储上，RDD的每个分区对应的就是一个Block，Block可以存储在内存，当内存不够时可以存储到磁盘上。

每个Block中存储着RDD所有数据项的一个子集，暴露给用户的可以是一个Block的迭代器（例如，用户可以通过mapPartitions获得分区迭代器进行操作），也可以就是一个数据项（例如，通过map函数对每个数据项并行计算）。本书会在后面章节具体介绍数据管理的底层实现细节。

如果是从HDFS等外部存储作为输入数据源，数据按照HDFS中的数据分布策略进行数据分区，HDFS中的一个Block对应Spark的一个分区。同时Spark支持重分区，数据通过Spark默认的或者用户自定义的分区器决定数据块分布在哪些节点。例如，支持Hash分区（按照数据项的Key值取Hash值，Hash值相同的元素放入同一个分区之内）和Range分区（将属于同一数据范围的数据放入同一分区）等分区策略。

下面具体介绍这些算子的功能。

3.3 Spark算子分类及功能

本节将主要介绍Spark算子的作用，以及算子的分类。

1.Saprk算子的作用

图3-3描述了Spark的输入、运行转换、输出。在运行转换中通过算子对RDD进行转换。算子是RDD中定义的函数，可以对RDD中的数据进行转换和操作。

[插图]

图3-3 Spark算子和数据空间

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型数据）。

Spark的核心数据模型是RDD，但RDD是个抽象类，具体由各子类实现，如MappedRDD、ShuffledRDD等子类。Spark将常用的大数据操作都转化成为RDD的子类。

2.算子的分类

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

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

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

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

下面分别对这3类算子进行详细介绍。

3.3.1 Value型Transformation算子

处理数据类型为Value型的Transformation算子可以根据RDD变换算子的输入分区与输出分区关系分为以下几种类型。

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

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

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

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

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

1.输入分区与输出分区一对一型

（1）map

将原来RDD的每个数据项通过map中的用户自定义函数f映射转变为一个新的元素。源码中的map算子相当于初始化一个RDD，新RDD叫作MappedRDD(this,sc.clean(f))。

图3-4中的每个方框表示一个RDD分区，左侧的分区经过用户自定义函数f:T-＞U映射为右侧的新的RDD分区。但是实际只有等到Action算子触发后，这个f函数才会和其他函数在一个Stage中对数据进行运算。V1输入f转换输出V’1。

[插图]

图3-4 map算子对RDD转换

（2）flatMap

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

图3-5中小方框表示RDD的一个分区，对分区进行flatMap函数操作，flatMap中传入的函数为f:T-＞U，T和U可以是任意的数据类型。将分区中的数据通过用户自定义函数f转换为新的数据。外部大方框可以认为是一个RDD分区，小方框代表一个集合。V1、V2、V3在一个集合作为RDD的一个数据项，转换为V’1、V’2、V’3后，将结合拆散，形成为RDD中的数据项。

[插图]

图3-5 flapMap算子对RDD转换

（3）mapPartitions

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

图3-6中，用户通过函数f(iter)=＞iter.filter(_＞=3)对分区中的所有数据进行过滤，＞=3的数据保留。一个方块代表一个RDD分区，含有1、2、3的分区过滤只剩下元素3。

[插图]

图3-6 mapPartitions算子对RDD转换

（4）glom

glom函数将每个分区形成一个数组，内部实现是返回的GlommedRDD。图3-7中的每个方框代表一个RDD分区。

图3-7中的方框代表一个分区。该图表示含有V1、V2、V3的分区通过函数glom形成一个数组Array[(V1),(V2),(V3)]。

[插图]

图3-7 glom算子对RDD转换

2.输入分区与输出分区多对一型

（1）union

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

图3-8中左侧的大方框代表两个RDD，大方框内的小方框代表RDD的分区。右侧大方框代表合并后的RDD，大方框内的小方框代表分区。含有V1，V2…U4的RDD和含有V1，V8…U8的RDD合并所有元素形成一个RDD。V1、V1、V2、V8形成一个分区，其他元素同理进行合并。

（2）cartesian

对两个RDD内的所有元素进行笛卡尔积操作。操作后，内部实现返回CartesianRDD。图3-9中左侧的大方框代表两个RDD，大方框内的小方框代表RDD的分区。右侧大方框代表合并后的RDD，大方框内的小方框代表分区。

[插图]

图3-8 union算子对RDD转换

图3-9中的大方框代表RDD，大方框中的小方框代表RDD分区。例如，V1和另一个RDD中的W1、W2、Q5进行笛卡尔积运算形成(V1,W1)、(V1,W2)、(V1,Q5)。

[插图]

图3-9 cartesian算子对RDD转换

3.输入分区与输出分区多对多型

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

函数实现如下。

①sc.clean()函数将用户函数预处理：

val cleanF=sc.clean(f)

②对数据map进行函数操作，最后再对groupByKey进行分组操作。

this.map(t=＞(cleanF(t),t)).groupByKey(p)

其中，p中确定了分区个数和分区函数，也就决定了并行化的程度。图3-10中的方框代表RDD分区。

图3-10中的方框代表一个RDD分区，相同key的元素合并到一个组。例如，V1，V2合并为一个Key-Value对，其中key为“V”，Value为“V1,V2”，形成V,Seq(V1,V2)。

[插图]

图3-10 groupBy算子对RDD转换

4.输出分区为输入分区子集型

（1）filter

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

下面代码为函数的本质实现。

def filter(f:T=＞Boolean):RDD[T]=new FilteredRDD(this,sc.clean(f))

图3-11中的每个方框代表一个RDD分区。T可以是任意的类型。通过用户自定义的过滤函数f，对每个数据项进行操作，将满足条件，返回结果为true的数据项保留。例如，过滤掉V2、V3保留了V1，将区分命名为V1'。

[插图]

图3-11 filter算子对RDD转换

（2）distinct

distinct将RDD中的元素进行去重操作。图3-12中的方框代表RDD分区。

图3-12中的每个方框代表一个分区，通过distinct函数，将数据去重。例如，重复数据V1、V1去重后只保留一份V1。

[插图]

图3-12 distinct算子对RDD转换

（3）subtract

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

图3-13中左侧的大方框代表两个RDD，大方框内的小方框代表RDD的分区。右侧大方框代表合并后的RDD，大方框内的小方框代表分区。V1在两个RDD中均有，根据差集运算规则，新RDD不保留，V2在第一个RDD有，第二个RDD没有，则在新RDD元素中包含V2。

[插图]

图3-13 subtract算子对RDD转换

（4）sample

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

内部实现是生成SampledRDD(withReplacement,fraction,seed)。

函数参数设置如下。

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

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

图3-14中的每个方框是一个RDD分区。通过sample函数，采样50%的数据。V1、V2、U1、U2、U3、U4采样出数据V1和U1、U2，形成新的RDD。

（5）takeSample

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

图3-15中左侧的方框代表分布式的各个节点上的分区，右侧方框代表单机上返回的结果数组。通过takeSample对数据采样，设置为采样一份数据，返回结果为V1。

[插图]

图3-14 sample算子对RDD转换

5.Cache型

（1）cache

cache将RDD元素从磁盘缓存到内存，相当于persist(MEMORY_ONLY)函数的功能。图3-14中的方框代表RDD分区。

图3-16中的每个方框代表一个RDD分区，左侧相当于数据分区都存储在磁盘，通过cache算子将数据缓存在内存。

[插图]

图3-15 takeSample算子对RDD转换

[插图]

图3-16 cache算子对RDD转换

（2）persist

persist函数对RDD进行缓存操作。数据缓存在哪里由StorageLevel枚举类型确定。有以下几种类型的组合（见图3-15），DISK代表磁盘，MEMORY代表内存，SER代表数据是否进行序列化存储。

下面为函数定义，StorageLevel是枚举类型，代表存储模式，用户可以通过图3-17按需选择。

persist(newLevel:Stor ageLevel)

图3-17中列出persist函数可以缓存的模式。例如，MEMORY_AND_DISK_SER代表数据可以存储在内存和磁盘，并且以序列化的方式存储。其他同理。

[插图]

图3-17 persist算子对RDD转换

图3-18中的方框代表RDD分区。disk代表存储在磁盘，mem代表存储在内存。数据最初全部存储在磁盘，通过persist(MEMORY_AND_DISK)将数据缓存到内存，但是有的分区无法容纳在内存，例如：图3-18中将含有V1,V2,V3的RDD存储到磁盘，将含有U1，U2的RDD仍旧存储在内存。

[插图]

图3-18 Persist算子对RDD转换

3.3.2 Key-Value型Transformation算子

Transformation处理的数据为Key-Value形式的算子，大致可以分为3种类型：输入分区与输出分区一对一、聚集、连接操作。

1.输入分区与输出分区一对一

mapValues：针对（Key,Value）型数据中的 Value进行Map操作，而不对Key进行处理。

图3-19中的方框代表RDD分区。a=＞a+2代表只对(V1,1)数据中的1进行加2操作，返回结果为3。

[插图]

图3-19 mapValues算子RDD对转换

2.对单个RDD或两个RDD聚集

（1）单个RDD聚集

1）combineByKey。

定义combineByKey算子的代码如下。

combineByKey[C](createCombiner:(V)⇒ C，

mergeValue:(C，V)⇒ C，

mergeCombiners:(C，C)⇒ C，

partitioner:Partitioner

mapSideCombine:Boolean=true，

serializer:Serializer=null):RDD[(K,C)]

说明：

□createCombiner：V=＞C，在C不存在的情况下，如通过V创建seq C。

□mergeValue：(C,V)=＞C，当C已经存在的情况下，需要merge，如把item V加到seq C中，或者叠加。

□mergeCombiners：(C,C)=＞C，合并两个C。

□partitioner：Partitioner（分区器），Shuffle时需要通过Partitioner的分区策略进行分区。

□mapSideCombine：Boolean=true，为了减小传输量，很多combine可以在map端先做。例如，叠加可以先在一个partition中把所有相同的Key的Value叠加，再shuffle。

□serializerClass:String=null，传输需要序列化，用户可以自定义序列化类。

例如，相当于将元素为(Int，Int)的RDD转变为了（Int,Seq[Int]）类型元素的RDD。

图3-20中的方框代表RDD分区。通过combineByKey，将（V1,2）、（V1,1）数据合并为（V1,Seq（2,1））。

[插图]

图3-20 comBineByKey算子对RDD转换

2）reduceByKey。

reduceByKey是更简单的一种情况，只是两个值合并成一个值，所以createCombiner很简单，就是直接返回v，而mergeValue和mergeCombiners的逻辑相同，没有区别。

函数实现代码如下。

def reduceByKey(partitioner:Partitioner，func:(V，V)=＞V):RDD[(K，V)]={

combineByKey[V]((v:V)=＞v，func，func，partitioner)

}

图3-21中的方框代表RDD分区。通过用户自定义函数（A,B）=＞（A+B），将相同Key的数据（V1,2）、（V1,1）的value相加，结果为（V1,3）。

[插图]

图3-21 reduceByKey算子对RDD转换

3）partitionBy。

partitionBy函数对RDD进行分区操作。

函数定义如下。

partitionBy(partitioner:Partitioner)

如果原有RDD的分区器和现有分区器（partitioner）一致，则不重分区，如果不一致，则相当于根据分区器生成一个新的ShuffledRDD。

图3-22中的方框代表RDD分区。通过新的分区策略将原来在不同分区的V1、V2数据都合并到了一个分区。

[插图]

图3-22 partitionBy算子对RDD转换

（2）对两个RDD进行聚集

cogroup函数将两个RDD进行协同划分，cogroup函数的定义如下。

cogroup[W](other:RDD[(K,W)],numPartitions:Int):RDD[(K,(Iterable[V],Iterable[W]))]

对在两个RDD中的Key-Value类型的元素，每个RDD相同Key的元素分别聚合为一个集合，并且返回两个RDD中对应Key的元素集合的迭代器。

(K,(Iterable[V],Iterable[W]))

其中，Key和Value，Value是两个RDD下相同Key的两个数据集合的迭代器所构成的元组。

图3-23中的大方框代表RDD，大方框内的小方框代表RDD中的分区。将RDD1中的数据（U1,1）、（U1,2）和RDD2中的数据（U1,2）合并为（U1,（（1,2）,（2）））。

[插图]

图3-23 Cogroup算子对RDD转换

3.连接

（1）join

□oin对两个需要连接的RDD进行cogroup函数操作，cogroup原理请见上文。cogroup操作之后形成的新RDD，对每个key下的元素进行笛卡尔积操作，返回的结果再展平，对应Key下的所有元组形成一个集合，最后返回RDD[(K,(V,W))]

下面代码为join的函数实现，本质是通过cogroup算子先进行协同划分，再通过flatMapValues将合并的数据打散。

this.cogroup(other,partitioner).flatMapValues { case(vs,ws)=＞

for(v ＜- vs;w ＜- ws)yield(v,w)}

图3-24是对两个RDD的join操作示意图。大方框代表RDD，小方框代表RDD中的分区。函数对拥有相同Key的元素（例如V1）为Key，以做连接后的数据结果为(V1,(1,1))和(V1,(1,2))。

[插图]

图3-24 join算子对RDD转换

（2）leftOutJoin和rightOutJoin

LeftOutJoin（左外连接）和RightOutJoin（右外连接）相当于在join的基础上先判断一侧的RDD元素是否为空，如果为空，则填充为空。如果不为空，则将数据进行连接运算，并返回结果。

下面代码是leftOutJoin的实现。

if(ws.isEmpty){

vs.map(v=＞(v,None))

}else {

for(v ＜- vs;w ＜- ws)yield(v,Some(w))

}

3.3.3 Actions算子

本质上在Actions算子中通过SparkContext执行提交作业的runJob操作，触发了RDD DAG的执行。

例如，Actions算子collect函数的代码如下，感兴趣的读者可以顺着这个入口进行源码剖析。

/*返回这个RDD的所有数据，结果以数组形式存储*/

def collect():Array[T]={

/*提交Job*/

val results=sc.runJob(this,(iter:Iterator[T])=＞iter.toArray)

Array.concat(results:_*)

}

下面根据Action算子的输出空间将Action算子进行分类：无输出、HDFS、Scala集合和数据类型。

1.无输出

（1）foreach

对RDD中的每个元素都应用f函数操作，不返回RDD和Array，而是返回Uint。

图3-25表示foreach算子通过用户自定义函数对每个数据项进行操作。本例中自定义函数为println()，控制台打印所有数据项。

2.HDFS

（1）saveAsTextFile

函数将数据输出，存储到HDFS的指定目录。

下面为函数的内部实现。

this.map(x=＞(NullWritable.get(),new Text(x.toString)))

.saveAsHadoopFile[TextOutputFormat[NullWritable,Text]](path)

将RDD中的每个元素映射转变为(Null,x.toString)，然后再将其写入HDFS。

图3-26中左侧的方框代表RDD分区，右侧方框代表HDFS的Block。通过函数将RDD的每个分区存储为HDFS中的一个Block。

[插图]

图3-25 foreach算子对RDD转换

[插图]

图3-26 saveAsHadoopFile算子对RDD转换

（2）saveAsObjectFile

saveAsObjectFile将分区中的每10个元素组成一个Array，然后将这个Array序列化，映射为(Null,BytesWritable(Y))的元素，写入HDFS为SequenceFile的格式。

下面代码为函数内部实现。

map(x=＞(NullWritable.get(),new BytesWritable(Utils.serialize(x))))

图3-27中的左侧方框代表RDD分区，右侧方框代表HDFS的Block。通过函数将RDD的每个分区存储为HDFS上的一个Block。

[插图]

图3-27 saveAsObjectFile算子对RDD转换

3.Scala集合和数据类型

（1）collect

collect相当于toArray，toArray已经过时不推荐使用，collect将分布式的RDD返回为一个单机的scala Array数组。在这个数组上运用scala的函数式操作。

图3-28中的左侧方框代表RDD分区，右侧方框代表单机内存中的数组。通过函数操作，将结果返回到Driver程序所在的节点，以数组形式存储。

（2）collectAsMap

collectAsMap对(K,V)型的RDD数据返回一个单机HashMap。对于重复K的RDD元素，后面的元素覆盖前面的元素。

图3-29中的左侧方框代表RDD分区，右侧方框代表单机数组。数据通过collectAsMap函数返回给Driver程序计算结果，结果以HashMap形式存储。

[插图]

图3-28 Collect算子对RDD转换

[插图]

图3-29 collectAsMap算子对RDD转换

（3）reduceByKeyLocally

实现的是先reduce再collectAsMap的功能，先对RDD的整体进行reduce操作，然

后再收集所有结果返回为一个HashMap。

（4）lookup

下面代码为lookup的声明。

lookup(key:K):Seq[V]

Lookup函数对(Key,Value)型的RDD操作，返回指定Key对应的元素形成的Seq。这个函数处理优化的部分在于，如果这个RDD包含分区器，则只会对应处理K所在的分区，然后返回由(K,V)形成的Seq。如果RDD不包含分区器，则需要对全RDD元素进行暴力扫描处理，搜索指定K对应的元素。

图3-30中的左侧方框代表RDD分区，右侧方框代表Seq，最后结果返回到Driver所在节点的应用中。

（5）count

count返回整个RDD的元素个数。内部函数实现如下。

Def count():Long=sc.runJob(this,Utils.getIteratorSize_).sum

在图3-31中，返回数据的个数为5。一个方块代表一个RDD分区。

[插图]

图3-30 lookup对RDD转换

[插图]

图3-31 count对RDD转换

（6）top

top可返回最大的k个元素。函数定义如下。

top(num:Int)(implicit ord:Ordering[T]):Array[T]

相近函数说明如下。

□top返回最大的k个元素。

□take返回最小的k个元素。

□takeOrdered返回最小的k个元素，并且在返回的数组中保持元素的顺序。

□first相当于top(1)返回整个RDD中的前k个元素，可以定义排序的方式Ordering[T]。返回的是一个含前k个元素的数组。

（7）reduce

reduce函数相当于对RDD中的元素进行reduceLeft函数的操作。函数实现如下。

Some(iter.reduceLeft(cleanF))

reduceLeft先对两个元素＜K，V＞进行reduce函数操作，然后将结果和迭代器取出的下一个元素＜k，V＞进行reduce函数操作，直到迭代器遍历完所有元素，得到最后结果。

在RDD中，先对每个分区中的所有元素＜K,V＞的集合分别进行reduceLeft。每个分区形成的结果相当于一个元素＜K,V＞，再对这个结果集合进行reduceleft操作。

例如:用户自定义函数如下。

f:(A，B)=＞(A._1+"@"+B._1，A._2+B._2)

图3-32中的方框代表一个RDD分区，通过用户自定函数f将数据进行reduce运算。示例最后的返回结果为V1@[插图]V2U!@U2@U3@U4,12。

[插图]

图3-32 reduce算子对RDD转换

（8）fold

fold和reduce的原理相同，但是与reduce不同，相当于每个reduce时，迭代器取的第一个元素是zeroValue。

图3-33中通过下面的用户自定义函数进行fold运算，图中的一个方框代表一个RDD分区。读者可以参照(7)reduce函数理解。

fold(("V0@"，2))((A，B)=＞(A._1+"@"+B._1，A._2+B._2))

[插图]

图3-33 fold算子对RDD转换

（9）aggregate

aggregate先对每个分区的所有元素进行aggregate操作，再对分区的结果进行fold操作。

aggreagate与fold和reduce的不同之处在于，aggregate相当于采用归并的方式进行数据聚集，这种聚集是并行化的。而在fold和reduce函数的运算过程中，每个分区中需要进行串行处理，每个分区串行计算完结果，结果再按之前的方式进行聚集，并返回最终聚集结果。

函数的定义如下。

aggregate[B](z:B)(seqop:(B,A)⇒ B,combop:(B,B)⇒ B):B

图3-34通过用户自定义函数对RDD 进行aggregate的聚集操作，图中的每个方框代表一个RDD分区。

rdd.aggregate("V0@",2)((A，B)=＞(A._1+"@"+B._1，A._2+B._2)),

(A,B)=＞(A._1+"@"+B_1,A._@+B_.2))

最后，介绍两个计算模型中的两个特殊变量。

广播（broadcast）变量：其广泛用于广播Map Side Join中的小表，以及广播大变量等场景。这些数据集合在单节点内存能够容纳，不需要像RDD那样在节点之间打散存储。Spark运行时把广播变量数据发到各个节点，并保存下来，后续计算可以复用。相比Hadoop的distributed cache，广播的内容可以跨作业共享。Broadcast的底层实现采用了BT机制。有兴趣的读者可以参考论文[插图]。

[插图]

图3-34 aggregate算子对RDD转换

㈡代表V。㈢代表U。

accumulator变量：允许做全局累加操作，如accumulator变量广泛使用在应用中记录当前的运行指标的情景。

3.4 本章小结

本章主要介绍了Spark的计算模型，Spark将应用程序整体翻译为一个有向无环图进行调度和执行。相比MapReduce，Spark提供了更加优化和复杂的执行流。

读者还可以深入了解Spark的运行机制与Spark算子，这样能更加直观地了解API的使用。Spark提供了更加丰富的函数式算子，这样就为Spark上层组件的开发奠定了坚实的基础。

通过阅读本章，读者可以对Spark计算模型进行更为宏观的把握。相信读者还想对Spark内部执行机制进行更深入的了解，下面章节就对Spark的内核进行更深入的剖析。