Spark的RDD原理以及2.0特性的介绍

《Spark的RDD原理以及2.0特性的介绍》要点：
本文介绍了Spark的RDD原理以及2.0特性的介绍，希望对您有用。如果有疑问，可以联系我们。

Spark 是什么?

Spark 是 Apache 顶级项目里面最火的大数据处理的计算引擎,它目前是负责大数据计算的工作.包括离线计算或交互式查询、数据挖掘算法、流式计算以及图计算等.全世界有许多公司和组织使用或给社区贡献代码,社区的活跃度见 www.github.com/apache/spark.

2013 年开始 Spark开发团队成立 Databricks,来对 Spark 进行运作和管理,并提供 Cloud 服务.Spark 社区基本保持一个季度一个版本,不出意外的话 Spark 2.0 将在五月底发布.

与 Mapreduce 相比,Spark 具备 DAG 执行引擎以及基于内存的多轮迭代计算等优势,在SQL 层面上,比 Hive/Pig 相比,引入关系数据库的许多特性,以及内存管理技术.另外在 Spark 上所有的计算模型最终都统一基于 RDD 之上运行执行,包括流式和离线计算.Spark 基于磁盘的性能是 MR 的 10 倍,基于内存的性能是 MR 的 100 倍 ? (见文后参考阅读?,下同)?.

Spark 提供 SQL、机器学习库 MLlib、流计算 Streaming 和图计算 Graphx,同时也支持 Scala、Java、Python 和 R 语言开发的基于 API 的应用程序.

RDD 的原理

RDD,英文全称叫 Resilient Distributed Datasets.

an RDD is a read-only,partitioned collection of records?.?字面意思是只读的分布式数据集.

但其实个人觉得可以把 RDD 理解为关系数据库 里的一个个操作,比如 map,filter,Join 等.在 Spark 里面实现了许多这样的 RDD 类,即可以看成是操作类.当我们调用一个 map 接口,底层实现是会生成一个 MapPartitionsRDD 对象,当 RDD 真正执行时,会调用 MapPartitionsRDD 对象里面的 compute 方法来执行这个操作的计算逻辑.但是不同的是,RDD 是 lazy 模式,只有像 count,saveasText 这种 action 动作被调用后再会去触发 runJob 动作.

RDD 分为二类：transformation 和 action.

• transformation 是从一个 RDD 转换为一个新的 RDD 或者从数据源生成一个新的 RDD；
• action 是触发 job 的执行.所有的?transformation 都是 lazy 执行,只有在 action 被提交的时候才触发前面整个 RDD 的执行图.如下

val file = sc.textFile(args(0))

val words = file.flatMap(line => line.split(” “))

val wordCounts = words.map(x => (x,1)).reduceByKey(_ + _,2) wordCounts.saveAsTextFile(args(1))

这段代码生成的 RDD 的执行树是如下图所示：

最终在 saveAsTextFile 方法时才会将整个 RDD 的执行图提交给 DAG 执行引擎,根据相关信息切分成一个一个 Stage,每个 Stage 去执行多个 task,最终完成整个 Job 的执行.

还有一个区别就是,RDD 计算后的中间结果是可以被持久化,当下一次需要使用时,可以直接使用之前持久化好的结果,而不是重新计算,并且这些结果被存储在各个结点的 executor 上.下一次使用时,调度器可以直接把 task 分发到存储持久化数据的结点上,减少数据的网络传输开稍.这种场景在数据挖掘迭代计算是经常出现.如下代码

val links = spark.textFile(…).map(…).persist() var ranks = // RDD of (URL,rank) pairs

for (i <- 1 to ITERATIONS) {

// Build an RDD of (targetURL,float) pairs // with the contributions sent by each page val contribs = links.join(ranks).flatMap {

(url,(links,rank)) =>

links.map(dest => (dest,rank/links.size)) }

// Sum contributions by URL and get new ranks

?

ranks = contribs.reduceByKey((x,y) => x+y)

.mapValues(sum => a/N + (1-a)*sum) }

以上代码生成的 RDD 执行树如下图所示：

计算 contribs-0 时需要使用 links 的计算逻辑,当 links 每个分片计算完后,会将这个结果保存到本地内存或磁盘上,下一次 contribs-1 计算要使用 links 的数据时,直接从上一次保存的内存和磁盘上读取就可以了.这个持久化系统叫做 blockManager,类似于在内部再构建了一个 KV 系统,K 表示每个分区 ID 号,V 表示这个分区计算后的结果.

另外在 streaming 计算时,每个 batch 会去消息队列上拉取这个时间段的数据,每个 Recevier 接收过来数据形成 block 块并存放到 blockManager 上,为了可靠性,这个 block 块可以远程备份,后续的 batch 计算就直接在之前已读取的 block 块上进行计算,这样不断循环迭代来完成流处理.

一个 RDD 一般会有以下四个函数组成.

1. 操作算子的物理执行逻辑

定义为：

def compute(split: Partition,context: TaskContext): Iterator[T]

如在 MapPartitionsRDD 里的实现是如下：

override def compute(split: Partition,context: TaskContext): Iterator[U] = f(context,split.index,firstParent[T].iterator(split,context))

（编辑：ASP站长网）