博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
RDD浅谈
阅读量:6854 次
发布时间:2019-06-26

本文共 4396 字,大约阅读时间需要 14 分钟。

RDD概念:Resilient Distributed Datasets

RDD(Resilient Distributed Dataset)叫做弹性分布式数据集,是Spark中最基本的数据抽象,它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点:自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中,后续的查询能够重用工作集,这极大地提升了查询速度。

同时,RDD还提供了一组丰富的操作来操作数据。RDD是只读的记录分区的集合,只能通过在其他RDD执行确定的转换操作(transformation操作)而创建。RDD可看作一个spark的对象,它本身存在于内存中,如对文件计算是一个RDD,等。一个RDD可以包含多个分区,每个分区就是一个dataset片段。但是RDD抽象出来的东西里面实际的数据,是分散在各个节点上面的,RDD可分区,分区的个数是我们可以指定的。默认情况下,一个hdfs块就是一个分区。

RDD的属性

  1. 一组分片(Partition),即数据集的基本组成单位。对于RDD来说,每个分片都会被一个计算任务处理,并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数,如果没有指定,那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。
  2. 一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的,每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合,不需要保存每次计算的结果。
  3. RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD,所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时,Spark可以通过这个依赖关系重新计算丢失的分区数据,而不是对RDD的所有分区进行重新计算。
  4. 一个Partitioner,即RDD的分片函数。当前Spark中实现了两种类型的分片函数,一个是基于哈希的HashPartitioner,另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD,才会有Partitioner,非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量,也决定了parent RDD Shuffle输出时的分片数量。
  5. 一个列表,存储存取每个Partition的优先位置(preferred location)。对于一个HDFS文件来说,这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念,Spark在进行任务调度的时候,会尽可能地将计算任务分配到其所要处理数据块的存储位置。

wordcount粗解RDD

懒惰调用

RDD将操作分为两类:transformation与action。无论执行了多少次transformation操作,RDD都不会真正执行运算,只有当action操作被执行时,运算才会触发。这就是Spark的惰性调用机制。

正式这种执行机制为是spark的高效计算的基础,正是因为懒惰执行,spark才能更有效的运行于于内存,使得高效的共享内存机制避免了大量中间结果,从而避免了磁盘写入写出带来的性能消耗,同时内部的存储对象可以是JAVA对象也避免了不必要的序列化和反序列化。

而在这基础之上,spark又实现了高效的容错机制,以及根据依赖关系进行工作优化。

宽依赖和窄依赖

由于RDD是粗粒度的操作数据集,每个Transformation操作都会生成一个新的RDD,所以RDD之间就会形成类似流水线的前后依赖关系;RDD和它依赖的父RDD(s)的关系有两种不同的类型,即窄依赖(narrow dependency)和宽依赖(wide dependency)。如图所示显示了RDD之间的依赖关系。

从图中可知:

窄依赖:是指每个父RDD的一个Partition最多被子RDD的一个Partition所使用,例如map、filter、union等操作都会产生窄依赖;(独生子女)

宽依赖:是指一个父RDD的Partition会被多个子RDD的Partition所使用,例如groupByKey、reduceByKey、sortByKey等操作都会产生宽依赖;(超生)

需要特别说明的是对join操作有两种情况:

(1)图中左半部分join:如果两个RDD在进行join操作时,一个RDD的partition仅仅和另一个RDD中已知个数的Partition进行join,那么这种类型的join操作就是窄依赖,例如图1中左半部分的join操作(join with inputs co-partitioned);

(2)图中右半部分join:其它情况的join操作就是宽依赖,例如图1中右半部分的join操作(join with inputs not co-partitioned),由于是需要父RDD的所有partition进行join的转换,这就涉及到了shuffle,因此这种类型的join操作也是宽依赖。

同时从这两种依赖关系我们也可以明确的看出了,窄依赖在运行上要优于宽依赖,但是在很多执行中宽依赖又是不可避免的。为了能够更好更高效的运行数据,spark基于这两种依赖关系对工作划分进行可设计(具体内容请见文末)

RDD的高效容错机制

一般来说,分布式数据集的容错性有两种方式:数据检查点和记录数据的更新。 面向大规模数据分析,数据检查点操作成本很高,需要通过数据中心的网络连接在机器之间复制庞大的数据集,而网络带宽往往比内存带宽低得多,同时还需要消耗更多的存储资源。

因此,Spark选择记录更新的方式,简单点来说就是RDD运行之后保存RDD之间的依赖关系,当出现错误时,通过重新计算的方式来恢复数据。但是,如果更新粒度太细太多,那么记录更新成本也不低。因此,RDD只支持粗粒度转换,即只记录单个块上执行的单个操作,然后将创建RDD的一系列变换序列(每个RDD都包含了他是如何由其他RDD变换过来的以及如何重建某一块数据的信息。因此RDD的容错机制又称“血统(Lineage)”容错)记录下来,以便恢复丢失的分区。

Lineage本质上很类似于数据库中的重做日志(Redo Log),只不过这个重做日志粒度很大,是对全局数据做同样的重做进而恢复数据。

除此之外RDD中有时候还需要设置检查点来提高性能:

  1. DAG中的Lineage过长,如果重算,则开销太大(如在PageRank中)。
  2. 在宽依赖上做Checkpoint获得的收益更大

Spark 的任务划分

  1. 构建Spark Application的运行环境(启动SparkContext),SparkContext向资源管理器(可以是Standalone、Mesos或YARN)注册并申请运行Executor资源;
  2. 资源管理器分配Executor资源并启动StandaloneExecutorBackend,Executor运行情况将随着心跳发送到资源管理器上;
  3. SparkContext构建成DAG图,将DAG图分解成Stage,并把Taskset发送给Task Scheduler。Executor向SparkContext申请Task,Task (我们之前提到的宽依赖与窄依赖的执行优化问题就在这里处理) Scheduler将Task发放给Executor运行同时SparkContext将应用程序代码发放给Executor。
  4. Task在Executor上运行,运行完毕释放所有资源。

RDD运行原理

那么 RDD在Spark架构中是如何运行的呢?总高层次来看,主要分为三步:

  1. 创建 RDD 对象
  2. DAGScheduler模块介入运算,计算RDD之间的依赖关系。RDD之间的依赖关系就形成了DAG
  3. 每一个JOB被分为多个Stage,划分Stage的一个主要依据是当前计算因子的输入是否是确定的,如果是则将其分在同一个Stage,避免多个Stage之间的消息传递开销

对于宽依赖与窄依赖进行的认任务分发优化

*在spark中,会根据RDD之间的依赖关系将DAG图(有向无环图)划分为不同的阶段,对于窄依赖,由于partition依赖关系的确定性,partition的转换处理就可以在同一个线程里完成,窄依赖就被spark划分到同一个stage中,而对于宽依赖,只能等父RDD shuffle处理完成后,下一个stage才能开始接下来的计算。

因此spark划分stage的整体思路是:从后往前推,遇到宽依赖就断开,划分为一个stage;遇到窄依赖就将这个RDD加入该stage中。因此在图2中RDD C,RDD D,RDD E,RDDF被构建在一个stage中,RDD A被构建在一个单独的Stage中,而RDD B和RDD G又被构建在同一个stage中。

这种分发方式可以使不同分区实现不同的流水线操作,有利于高效的运行与容错机制,想象一下,当运行时当C1-->D1运行结束时候,就可以直接运行D1-->F1的操作了,这样进行的并行计算有效的提升了计算性能。同时当出现错误时候,良好的stage划分也减少了重新计算所带来的成本。

在spark中,Task的类型分为2种:ShuffleMapTask和ResultTask;

简单来说,DAG的最后一个阶段会为每个结果的partition生成一个ResultTask,即每个Stage里面的Task的数量是由该Stage中最后一个RDD的Partition的数量所决定的!而其余所有阶段都会生成ShuffleMapTask;之所以称之为ShuffleMapTask是因为它需要将自己的计算结果通过shuffle到下一个stage中;也就是说上图中的stage1和stage2相当于mapreduce中的Mapper,而ResultTask所代表的stage3就相当于mapreduce中的reducer。

Hadoop中MapReduce操作中的Mapper和Reducer在spark中的基本等量算子是map和reduceByKey;不过区别在于:Hadoop中的MapReduce天生就是排序的;而reduceByKey只是根据Key进行reduce,但spark除了这两个算子还有其他的算子;因此从这个意义上来说,Spark比Hadoop的计算算子更为丰富。

参考

本文原为个人笔记,参考了众多资料,包括但不止于以下内容。

转载于:https://juejin.im/post/5bcc38dfe51d450e6548e160

你可能感兴趣的文章
阿里巴巴“新18罗汉”养成记
查看>>
centos7 yum安装zabbix图解
查看>>
jQuery 解决 click 和 dblclick 冲突
查看>>
基于Numpy的线性代数运算
查看>>
研究发现线粒体自噬调控肝癌发生的新机制
查看>>
Android 学习之四大组件(二)——service
查看>>
屏幕指定区域识别
查看>>
我的.vimrc,代码完成基于YcmCompleteMe版
查看>>
JS魔法堂:那些困扰你的DOM集合类型
查看>>
贴一些 Python 的笔记
查看>>
给root用户添加远程连接权限
查看>>
CentOS7下Apache2.4.6使用MySQL5.7验证
查看>>
linux 查看并发
查看>>
Linux下FTP服务器的安装和简单配置
查看>>
jQuery基本用法二
查看>>
Asp.net网站部署时遇到的一些问题
查看>>
WinForm webbrowser控件的使用
查看>>
<Power Shell>09 利用powershell 查找旧文件
查看>>
windows phone (16) UI变换 下
查看>>
管理中用人的两种思维
查看>>