1.4　弹性分布式数据集

Spark大数据分析实战

1.4　弹性分布式数据集 本节将介绍弹性分布式数据集RDD。Spark是一个分布式计算框架，而RDD是其对分布式内存数据的抽象，可以认为RDD就是Spark分布式算法的数据结构，而RDD之上的操作是Spark分布式算法的核心原语，由数据结构和原语设计上层算法。Spark最终会将算法（RDD上的一连串操作）翻译为DAG形式的工作流进行调度，并进行分布式任务的分发。

1.4.1　RDD简介 在集群背后，有一个非常重要的分布式数据架构，即弹性分布式数据集（Resilient Distributed Dataset，RDD）。它在集群中的多台机器上进行了数据分区，逻辑上可以认为是一个分布式的数组，而数组中每个记录可以是用户自定义的任意数据结构。RDD是Spark的核心数据结构，通过RDD的依赖关系形成Spark的调度顺序，通过对RDD的操作形成整个Spark程序。 （1）RDD创建方式 1）从Hadoop文件系统（或与Hadoop兼容的其他持久化存储系统，如Hive、Cassandra、HBase）输入（例如HDFS）创建。 2）从父RDD转换得到新RDD。 3）通过parallelize或makeRDD将单机数据创建为分布式RDD。 （2）RDD的两种操作算子 对于RDD可以有两种操作算子：转换（Transformation）与行动（Action）。 1）转换（Transformation）：Transformation操作是延迟计算的，也就是说从一个RDD转换生成另一个RDD的转换操作不是马上执行，需要等到有Action操作的时候才会真正触发运算。 2）行动（Action）：Action算子会触发Spark提交作业（Job），并将数据输出Spark系统。 （3）RDD的重要内部属性 通过RDD的内部属性，用户可以获取相应的元数据信息。通过这些信息可以支持更复杂的算法或优化。 1）分区列表：通过分区列表可以找到一个RDD中包含的所有分区及其所在地址。 2）计算每个分片的函数：通过函数可以对每个数据块进行RDD需要进行的用户自定义函数运算。 3）对父RDD的依赖列表：为了能够回溯到父RDD，为容错等提供支持。 4）对key-value pair数据类型RDD的分区器，控制分区策略和分区数。通过分区函数可以确定数据记录在各个分区和节点上的分配，减少分布不平衡。 5）每个数据分区的地址列表（如HDFS上的数据块的地址）。 如果数据有副本，则通过地址列表可以获知单个数据块的所有副本地址，为负载均衡和容错提供支持。 （4）Spark计算工作流 图1-5中描述了Spark的输入、运行转换、输出。在运行转换中通过算子对RDD进行转换。算子是RDD中定义的函数，可以对RDD中的数据进行转换和操作。 ·输入：在Spark程序运行中，数据从外部数据空间（例如，HDFS、Scala集合或数据）输入到Spark，数据就进入了Spark运行时数据空间，会转化为Spark中的数据块，通过BlockManager进行管理。 ·运行：在Spark数据输入形成RDD后，便可以通过变换算子fliter等，对数据操作并将RDD转化为新的RDD，通过行动（Action）算子，触发Spark提交作业。如果数据需要复用，可以通过Cache算子，将数据缓存到内存。 ·输出：程序运行结束数据会输出Spark运行时空间，存储到分布式存储中（如saveAsTextFile输出到HDFS）或Scala数据或集合中（collect输出到Scala集合，count返回Scala Int型数据）。 [插图] 图1-5　Spark算子和数据空间 Spark的核心数据模型是RDD，但RDD是个抽象类，具体由各子类实现，如MappedRDD、ShuffledRDD等子类。Spark将常用的大数据操作都转化成为RDD的子类。

1.4.2　RDD算子分类 本节将主要介绍Spark算子的作用，以及算子的分类。 Spark算子大致可以分为以下两类。 1）Transformation变换算子：这种变换并不触发提交作业，完成作业中间过程处理。 2）Action行动算子：这类算子会触发SparkContext提交Job作业。 下面分别对两类算子进行详细介绍。 1.Transformations算子 下文将介绍常用和较为重要的Transformation算子。 （1）map 将原来RDD的每个数据项通过map中的用户自定义函数f映射转变为一个新的元素。源码中map算子相当于初始化一个RDD，新RDD叫做MappedRDD（this，sc.clean（f））。 图1-7中每个方框表示一个RDD分区，左侧的分区经过用户自定义函数f：T->U映射为右侧的新RDD分区。但是，实际只有等到Action算子触发后这个f函数才会和其他函数在一个stage中对数据进行运算。在图1-6中的第一个分区，数据记录V1输入f，通过f转换输出为转换后的分区中的数据记录V'1。 （2）flatMap 将原来RDD中的每个元素通过函数f转换为新的元素，并将生成的RDD的每个集合中的元素合并为一个集合，内部创建FlatMappedRDD（this，sc.clean（f））。 [插图] 图1-6　map算子对RDD转换 图1-7表示RDD的一个分区进行flatMap函数操作，flatMap中传入的函数为f：T->U，T和U可以是任意的数据类型。将分区中的数据通过用户自定义函数f转换为新的数据。外部大方框可以认为是一个RDD分区，小方框代表一个集合。V1、V2、V3在一个集合作为RDD的一个数据项，可能存储为数组或其他容器，转换为V'1、V'2、V'3后，将原来的数组或容器结合拆散，拆散的数据形成为RDD中的数据项。 [插图] 图1-7　flapMap算子对RDD转换 （3）mapPartitions mapPartitions函数获取到每个分区的迭代器，在函数中通过这个分区整体的迭代器对整个分区的元素进行操作。内部实现是生成MapPartitionsRDD。图1-8中的方框代表一个RDD分区。 图1-8中，用户通过函数f（iter）=>iter.filter（_>=3）对分区中所有数据进行过滤，大于和等于3的数据保留。一个方块代表一个RDD分区，含有1、2、3的分区过滤只剩下元素3。 [插图] 图1-8　mapPartitions算子对RDD转换 （4）union 使用union函数时需要保证两个RDD元素的数据类型相同，返回的RDD数据类型和被合并的RDD元素数据类型相同。并不进行去重操作，保存所有元素，如果想去重可以使用distinct（）。同时Spark还提供更为简洁的使用union的API，通过++符号相当于union函数操作。 图1-9中左侧大方框代表两个RDD，大方框内的小方框代表RDD的分区。右侧大方框代表合并后的RDD，大方框内的小方框代表分区。合并后，V1、V2、V3……V8形成一个分区，其他元素同理进行合并。 （5）cartesian 对两个RDD内的所有元素进行笛卡尔积操作。操作后，内部实现返回CartesianRDD。图1-10中左侧大方框代表两个RDD，大方框内的小方框代表RDD的分区。右侧大方框代表合并后的RDD，大方框内的小方框代表分区。 例如：V1和另一个RDD中的W1、W2、Q5进行笛卡尔积运算形成（V1，W1）、（V1，W2）、（V1，Q5）。 [插图] 图1-9　union算子对RDD转换 [插图] 图1-10　cartesian算子对RDD转换 （6）groupBy groupBy：将元素通过函数生成相应的Key，数据就转化为Key-Value格式，之后将Key相同的元素分为一组。 函数实现如下： 1）将用户函数预处理：

val cleanF = sc.clean(f)

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

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

其中，p确定了分区个数和分区函数，也就决定了并行化的程度。 图1-11中方框代表一个RDD分区，相同key的元素合并到一个组。例如V1和V2合并为V，Value为V1，V2。形成V，Seq（V1，V2）。 [插图] 图1-11　groupBy算子对RDD转换 （7）filter filter函数功能是对元素进行过滤，对每个元素应用f函数，返回值为true的元素在RDD中保留，返回值为false的元素将被过滤掉。内部实现相当于生成FilteredRDD（this，sc.clean（f））。 下面代码为函数的本质实现：

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

图1-12中每个方框代表一个RDD分区，T可以是任意的类型。通过用户自定义的过滤函数f，对每个数据项操作，将满足条件、返回结果为true的数据项保留。例如，过滤掉V2和V3保留了V1，为区分命名为V'1。 （8）sample sample将RDD这个集合内的元素进行采样，获取所有元素的子集。用户可以设定是否有放回的抽样、百分比、随机种子，进而决定采样方式。 内部实现是生成SampledRDD（withReplacement，fraction，seed）。 函数参数设置： ·withReplacement=true，表示有放回的抽样。 ·withReplacement=false，表示无放回的抽样。 图1-13中的每个方框是一个RDD分区。通过sample函数，采样50%的数据。V1、V2、U1、U2……U4采样出数据V1和U1、U2形成新的RDD。 [插图] 图1-12　filter算子对RDD转换 [插图] 图1-13　sample算子对RDD转换 （9）cache cache将RDD元素从磁盘缓存到内存。相当于persist（MEMORY_ONLY）函数的功能。 [插图] 图1-14　Cache算子对RDD转换 图1-14中每个方框代表一个RDD分区，左侧相当于数据分区都存储在磁盘，通过cache算子将数据缓存在内存。 （10）persist persist函数对RDD进行缓存操作。数据缓存在哪里依据StorageLevel这个枚举类型进行确定。有以下几种类型的组合（见图1-14），DISK代表磁盘，MEMORY代表内存，SER代表数据是否进行序列化存储。 下面为函数定义，StorageLevel是枚举类型，代表存储模式，用户可以通过图1-14按需进行选择。

persist(newLevel:StorageLevel)

图1-15中列出persist函数可以进行缓存的模式。例如，MEMORY_AND_DISK_SER代表数据可以存储在内存和磁盘，并且以序列化的方式存储，其他同理。 [插图] 图1-15　persist算子对RDD转换 图1-16中方框代表RDD分区。disk代表存储在磁盘，mem代表存储在内存。数据最初全部存储在磁盘，通过persist（MEMORY_AND_DISK）将数据缓存到内存，但是有的分区无法容纳在内存，将含有V1、V2、V3的分区存储到磁盘。 （11）mapValues mapValues：针对（Key，Value）型数据中的Value进行Map操作，而不对Key进行处理。 图1-17中的方框代表RDD分区。a=>a+2代表对（V1，1）这样的Key Value数据对，数据只对Value中的1进行加2操作，返回结果为3。 [插图] 图1-16　Persist算子对RDD转换 [插图] 图1-17　mapValues算子RDD对转换 （12）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。 图1-18中的方框代表RDD分区。如图，通过combineByKey，将（V1，2），（V1，1）数据合并为（V1，Seq（2，1））。 （13）reduceByKey reduceByKey是比combineByKey更简单的一种情况，只是两个值合并成一个值，（Int，Int V）to（Int，Int C），比如叠加。所以createCombiner reduceBykey很简单，就是直接返回v，而mergeValue和mergeCombiners逻辑是相同的，没有区别。 [插图] 图1-18　comBineByKey算子对RDD转换 函数实现：

def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)] = { combineByKey[V]((v: V) => v, func, func, partitioner) }

图1-19中的方框代表RDD分区。通过用户自定义函数（A，B）=>（A+B）函数，将相同key的数据（V1，2）和（V1，1）的value相加运算，结果为（V1，3）。 [插图] 图1-19　reduceByKey算子对RDD转换

（14）join join对两个需要连接的RDD进行cogroup函数操作，将相同key的数据能够放到一个分区，在cogroup操作之后形成的新RDD对每个key下的元素进行笛卡尔积的操作，返回的结果再展平，对应key下的所有元组形成一个集合。最后返回RDD[（K，（V，W））]。 下面代码为join的函数实现，本质是通过cogroup算子先进行协同划分，再通过flatMapValues将合并的数据打散。

this.cogroup(other,partitioner).f?latMapValues{case(vs,ws)=> for(v<-vs;w>-ws)yield(v,w) }

图1-20是对两个RDD的join操作示意图。大方框代表RDD，小方框代表RDD中的分区。函数对相同key的元素，如V1为key做连接后结果为（V1，（1，1））和（V1，（1，2））。 2.Actions算子 本质上在Action算子中通过SparkContext进行了提交作业的runJob操作，触发了RDD DAG的执行。 [插图] 图1-20　join算子对RDD转换 例如，Action算子collect函数的代码如下，感兴趣的读者可以顺着这个入口进行源码剖析：

• /**
  • Return an array that contains all of the elements in this RDD. / def collect(): Array[T] = { /提交Job/ val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray) Array.concat(results: _) }

下面将介绍常用和较为重要的Action算子。 （1）foreach foreach对RDD中的每个元素都应用f函数操作，不返回RDD和Array，而是返回Uint。 图1-21表示foreach算子通过用户自定义函数对每个数据项进行操作。本例中自定义函数为println（），控制台打印所有数据项。 [插图] 图1-21　foreach算子对RDD转换 （2）saveAsTextFile 函数将数据输出，存储到HDFS的指定目录。 下面为saveAsTextFile函数的内部实现，其内部通过调用saveAsHadoopFile进行实现：

this.map(x => (NullWritable.get(), new Text(x.toString))) .saveAsHadoopFileTextOutputFormat[NullWritable, Text]

将RDD中的每个元素映射转变为（null，x.toString），然后再将其写入HDFS。 图1-22中左侧方框代表RDD分区，右侧方框代表HDFS的Block。通过函数将RDD的每个分区存储为HDFS中的一个Block。 （3）collect collect相当于toArray，toArray已经过时不推荐使用，collect将分布式的RDD返回为一个单机的scala Array数组。在这个数组上运用scala的函数式操作。 图1-23中左侧方框代表RDD分区，右侧方框代表单机内存中的数组。通过函数操作，将结果返回到Driver程序所在的节点，以数组形式存储。

图1-23中左侧方框代表RDD分区，右侧方框代表单机内存中的数组。通过函数操作，将结果返回到Driver程序所在的节点，以数组形式存储。 [插图] 图1-22　saveAsHadoopFile算子对RDD转换 [插图] 图1-23　Collect算子对RDD转换 （4）count count返回整个RDD的元素个数。 内部函数实现为：

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

图1-24中，返回数据的个数为5。一个方块代表一个RDD分区。 [插图] 图1-24　count对RDD算子转换

1.5　本章小结 本章首先介绍了Spark分布式计算平台的基本概念、原理以及Spark生态系统BDAS之上的典型组件。Spark为用户提供了系统底层细节透明、编程接口简洁的分布式计算平台。Spark具有内存计算、实时性高、容错性好等突出特点。同时本章介绍了Spark的计算模型，Spark会将应用程序整体翻译为一个有向无环图进行调度和执行。相比MapReduce，Spark提供了更加优化和复杂的执行流。读者还可以深入了解Spark的运行机制与Spark算子，这样能更加直观地了解API的使用。Spark提供了更加丰富的函数式算子，这样就为Spark上层组件的开发奠定了坚实的基础。 相信读者已经想了解如何开发Spark程序，接下来将就Spark的开发环境配置进行阐述。