Spark计算模型

[TOC]

Spark计算模型

Spark程序模型

一个经典的示例模型

• SparkContext中的textFile函数从HDFS读取日志文件,输出变量file

  var file = sc.textFile("hdfs://***")

• RDD中的filter函数过滤带有'ERROR'的行,输出errors(一个RDD)

  var errors = file.filter(line=>line.contains("ERROR"))

• RDD中的count函数返回"ERROR"的行数:errors.count()

小结

• 用户程序对RDD通过多个函数进行操作,将RDD进行转换

• BlockManager管理RDD的物理分区,每个Block节点上对应一个数据块,可以存储在内存中或者磁盘

• RDD中的partition是一个逻辑数据块,对应相应的物理块Block

• 一个RDD在代码中相当于是数据的一个元数据结构,存储着数据分区及其逻辑结构映射关系,存储着RDD之前的依赖转换关系

弹性分布式数据集(resilient distributed dataset,RDD)

RDD简介

• 它是一个分布式数据架构,是逻辑集中的实体,在集群中的多台机器上进行了数据分区

• RDD是Spark的核心数据结构,通过RDD的依赖关系形成了Spark的调度顺序

• 通过对RDD的操作形成整个Spark程序

RDD的两种创建方式

• 从Hadoop文件系统(或其他与Hadoop兼容的持久化存储系统,如Hive,Cassandra,Hbase...)输入创建

• 从父RDD转换得到新的RDD

RDD的两种操作算子

RDD有两种计算操作算子:Transformation(转换)和Action(行动)

Transformation(转换)

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

Action(行动)

Action算子会触发Spark提交作业,并将数据输出到Spark系统

RDD的重要内部属性

• 分区列表

• 计算每个分片的函数

• 对父RDD的依赖列表

• 对Key-Value对数据类型RDD的分区器,控制分区策略和分区数

• 每个数据分区的地址列表(如HDFS上的数据块的地址)

RDD与DSM的异同

DSM

• DSM(distributed shared memory, 分布式共享内存).

• 在DSM系统中,应用可以向全局地址空间的任意位置进行读写操作

• DSM是一种通用的内存数据抽象

RDD与DSM的区别

• RDD不仅可以通过批量转换创建RDD,还可以对任意内存位置读写

• RDD限制应用执行批量写操作,这样有利于实现有效的容错

Spark的数据存储

• 核心是RDD

• RDD可被抽象为一个大的数组,但是这个数组是分布在集群上的

• 逻辑上RDD的每个分区叫一个Partition

• RDD对象实质上是一个元数据结构,存储着Block,Node等的映射关系

• 一个RDD就是一组分区,在物理数据存储上,RDD的每个分区对应的就是一个Block,Block可以存储在内存,也可以缓存在磁盘上

• 每个Block中存储着RDD所有数据项的一个子集,暴露给用户的可以是一个Block的迭代器也可以是一个数据项

• HDFS中的一个分区对应Spark的一个分区

• 支持Hash分区和Range分区

算子的分类及功能

算子的分类

• Value数据类型的Transformation算子

  这种变换并不触发提交作业,只对处理的数据项是Value型的数据

• Key-Value性Transformation算子

  这种变换不触发提交作业,针对处理的数据项是Key-Value型的数据对

• Action算子,这类算子触发SparkContext提交Job作业

Value型Transformation算子

根据RDD变换算子的输入分区和输出分区关系分为以下几种类型.

输入分区和输出分区一对一型

map

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

flatMap

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

mapPartitions

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

glom

glom函数将每个分区形成一个数组,内部实现是返回的GlommeRDD.

输入分区和输出分区多对一型

union

使用union函数时需要保证两个RDD元素的数据类型相同,返回的RDD数据类型和被合并的RDD元素数据类型相同,并不进行去重操作,保存所有元素.

如果想去重可以适用distinct().++符号相当于union函数操作

cartesian

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

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

groupBy

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

函数实现如下:

• sc.clean()函数将用户函数预处理:

  var cleanF = sc.clean(f)

• 对数据map进行函数操作,最后再对groupByKey进行分组操作

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

  其中,p中确定了分区个数和分区函数,也就决定了并行化的程度.

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

filter

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

下面代码是函数的本质实现:

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

distinct

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

subtract

subtract相当于进行集合的差操作.RDD1去除RDD1和RDD2交集中的所有元素.

sample

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

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

• withReplacement:是否有放回的抽样.=true为有放回抽样

• fraction:采样百分比,0-1

• seed:随机种子

takeSample

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

Cache型

cache

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

persist

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

下面为函数定义,StorageLevel是枚举类型,代表存储模式:

persist(newLevel: StorageLevel)

Key-Value型Transformation算子

Transformation处理的数据为Key-Value形式的算子,大致分为一下3中类型.

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

mapValues

针对(Key, Value)型数据中的Value进行Map操作,而不对Key进行处理

聚集

对单个RDD聚集

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端先做.

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

例如相当于把(Int, Int)的RDD变成了(Int, Seq[Int])类型元素的RDD

reduceByKey

多个值合并成一个值(将相同Key的数据的value相加).

函数实现如下.

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

partitionBy

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

函数定义如下.

partitionBy(partitioner: Partitioner)

如果原有RDD的分区器和现有分区器一致,则不重新分区,否则相当于根据分区器生成一个新的ShuffledRDD.

对两个RDD进行聚集

cogroup

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, (Itetable[V], Iterable[W]).其中Key和Value,Value是两个RDD想相同Key的两个数据集合的迭代器所构成的元组.

连接

join

join对两个需要连接的RDD进行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) }

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))
}

Actions算子

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

下面根据Action算子的输出空间将Action算子进行分类.

无输出

foreach

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

HDFS

saveAsTextFile

函数将数据输出,存储到HDFS的指定目录.

下面为函数的内部实现.

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

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

saveAsObjectFile

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

下面是代码实现.

map(x=>(NullWirtable.get(), new BytesWritable(Utils.serialize(x))))

Scala集合和数据类型

collect

collect相当于toArray,将分布式的RDD返回为一个单机的scala Array数组.在这个数组上运用scala的函数式操作.

collectAsMap

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

reduceByKeyLocally

实现的是先reduce再collectAsMap的功能,相对RDD的整体进行reduce操作,然后再收集所有结果返回为一个HashMap.

lookup

下面代码是lookup的声明.

lookup(key: K): Seq[V]

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

count

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

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

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].返回的是一个数组

reduce

reduce函数相当于对RDD中的元素进行reduceLeft函数的操作.函数定义如下.

Some(iter.reduceLeft(cleanF))

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

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

fold

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

aggregate

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

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

函数定义如下.

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