2021年大数据Spark(十二):Spark Core的RDD详解

2021-10-09 17:19:34 浏览数 (1)


RDD详解

为什么需要RDD?

没有RDD之前,

1.MR:只提供了map和reduce的API,而且编写麻烦,运行效率低!---早就淘汰了!

2.使用Scala/Java的本地集合:但是只能完成本地单机版的,如果要实现分布式的,很困难!

所以需要有一个分布式的数据抽象,也就是用该抽象,可以表示分布式的集合,那么基于这个分布式集合进行操作,就可以很方便的完成分布式的WordCount!(该分布式集合底层应该将实现的细节封装好,提供简单易用的API!)---在此背景之下,RDD就诞生了!

AMP实验室发表的一篇关于RDD的论文:《Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing》就是为了解决这些问题的

RDD提供了一个抽象的数据模型,不必担心底层数据的分布式特性,只需将具体的应用逻辑表达为一系列转换操作(函数),不同RDD之间的转换操作之间还可以形成依赖关系,进而实现管道化,从而避免了中间结果的存储,大大降低了数据复制、磁盘IO和序列化开销,并且还提供了更多的API(map/reduec/filter/groupBy...)。

什么是RDD?

在Spark开山之作Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing这篇paper中(以下简称 RDD Paper),Matei等人提出了RDD这种数据结构,文中开头对RDD的定义是:

A Resilient Distributed Dataset (RDD), the basic abstraction in Spark. Represents an immutable, partitioned collection of elements that can be operated on in parallel.

RDD(Resilient Distributed Dataset)弹性分布式数据集,是Spark中最基本的数据抽象,代表一个不可变、可分区、里面的元素可并行计算的集合

所有的运算以及操作都建立在 RDD 数据结构的基础之上。

RDD设计的核心点为:

拆分核心要点三个方面

可以认为RDD是分布式的列表List或数组Array,抽象的数据结构,RDD是一个抽象类Abstract Class和泛型Generic Type:

RDD弹性分布式数据集核心点示意图如下:

本地List集合 | 单机硬盘存储

RDD分布式集合 | HDFS分布式存储

分布式的List

RDD的5大特性

RDD 数据结构内部有五个特性(摘录RDD 源码):

前三个特征每个RDD都具备的,后两个特征可选的。

第一个:A list of partitions

  1. 一组分片(Partition)/一个分区(Partition)列表,即数据集的基本组成单位;
  2. 对于RDD来说,每个分片都会被一个计算任务处理,分片数决定并行度;
  3. 用户可以在创建RDD时指定RDD的分片个数,如果没有指定,那么就会采用默认值;

第二个:A function for computing each split

  1. 一个函数会被作用在每一个分区;
  2. Spark中RDD的计算是以分片为单位的,compute函数会被作用到每个分区上;

第三个:A list of dependencies on other RDDs

  1. 一个RDD会依赖于其他多个RDD;
  2. RDD的每次转换都会生成一个新的RDD,所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时,Spark可以通过这个依赖关系重新计算丢失的分区数据,而不是对RDD的所有分区进行重新计算(Spark的容错机制);

 第四个:Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)

  1. 可选项,对于KeyValue类型的RDD会有一个Partitioner,即RDD的分区函数;
  2. 当前Spark中实现了两种类型的分区函数,一个是基于哈希的HashPartitioner,另外一个是基于范围的RangePartitioner。
  3. 只有对于于key-value的RDD,才会有Partitioner,非key-value的RDD的Parititioner的值是None。
  4. Partitioner函数不但决定了RDD本身的分片数量,也决定了parent RDD Shuffle输出时的分片数量。

第五个:Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

  1. 可选项,一个列表,存储存取每个Partition的优先位置(preferred location);
  2. 对于一个HDFS文件来说,这个列表保存的就是每个Partition所在的块的位置。
  3. 按照"移动数据不如移动计算"的理念,Spark在进行任务调度的时候,会尽可能选择那些存有数据的worker节点来进行任务计算。(数据本地性)

RDD 是一个数据集的表示,不仅表示了数据集,还表示了这个数据集从哪来、如何计算,主要属性包括五个方面(必须牢记,通过编码加深理解,面试常问):

RDD将Spark的底层的细节都隐藏起来(自动容错、位置感知、任务调度执行,失败重试等),让开发者可以像操作本地集合一样以函数式编程的方式操作RDD这个分布式数据集,进行各种并行计算,RDD中很多处理数据函数与列表List中相同与类似。

WordCount中RDD

在内部,每个RDD都有五个主要特性:

1.-分区列表:每个RDD都有会分区的概念,类似与HDFS的分块, 分区的目的:提高并行度!

2.-用于计算每个分区的函数:用函数来操作各个分区中的数据

3.-对其他RDD的依赖列表:后面的RDD需要依赖前面的RDD

4.-可选地,键值RDDs的分区器(例如,reduceByKey中的默认的Hash分区器)

5.-可选地,计算每个分区的首选位置列表/最佳位置(例如HDFS文件)--移动计算比移动数据更划算!

以词频统计WordCount程序为例,查看整个Job中各个RDD类型及依赖关系

运行程序结束后,查看WEB UI监控页面,此Job(RDD调用foreach触发)执行DAG图:

上图中相关说明如下:

 第一点、黑色圆圈表示一个RDD

  1. 上图中有5个黑色圆圈,说明整个Job中有个5个RDD
  2. 【1号】RDD类型:HadoopRDD,从HDFS或LocalFS读取文件数据;
  3. 【2号、3号和4号】RDD类型:MapPartitionsRDD,从一个RDD转换而来,没有经过shuffle操作;
  4. 【5号】RDD类型:ShuffledRDD,从一个RDD转换而来,经过Shuffle重分区操作,Spark Shuffle类似MapReduce流程中Map Phase和Reduce Phase中的Shuffle;

 第二点、浅蓝色矩形框表示调用RDD函数

  1. 上图中【5号】RDD所在在蓝色矩形框上的函数【reduceByKey】,表明【5号】RDD是【4号】RDD调用reduceByKey函数得到;

 第三点、查看ShuffleRDD源码,实现RDD的5个特性

RDD 设计的一个重要优势是能够记录 RDD 间的依赖关系,即所谓血统(lineage)。

通过丰富的转移操作(Transformation),可以构建一个复杂的有向无环图,并通过这个图来一步步进行计算。

0 人点赞