《大数据离线分析技术》课件-项目5：RDD核心概念与操作

RDD概述什么是RDDRDD的属性什么是RDD/01

RDD（ResilientDistributedDataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象。代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。什么是RDDRDD中的弹性是指：存储的弹性：内存与磁盘的自动切换。容错的弹性：数据丢失可以自动恢复。计算的弹性：计算出错重试机制。分片的弹性：可根据需要重新分片。什么是RDD

RDD中的不可变是指创建一个RDD如果更改，并不是真正意义上的更改，只是又创建了一个新的RDD。RDD中可分区是指能进行分区。RDD中并行计算是指，因为RDD的分区特性，所以支持并行处理的特性。即不同节点上的数据可以分别被处理，然后生成一个新的RDD。什么是RDD

RDD的结构如图所示：什么是RDDRDD的属性/02RDD的属性如下：一组分区（Partition），即数据集的基本组成单位;一个计算每个分区的函数;RDD之间的依赖关系;一个Partitioner，即RDD的分片函数;一个列表，存储存取每个Partition的优先位置（preferredlocation）。RDD的属性1.什么是RDD

2.RDD的属性

RDD的特点RDD特点概述RDD的特点详解RDD特点概述/01

RDD表示只读的分区的数据集，对RDD进行改动，只能通过RDD的转换操作，由一个RDD得到一个新的RDD，新的RDD包含了从其他RDD衍生所必需的信息。RDDs之间存在依赖，RDD的执行是按照血缘关系延时计算的。如果血缘关系较长，可以通过持久化RDD来切断血缘关系。RDD特点概述RDD的特点详解/02

存储的弹性：内存与磁盘的自动切换；容错的弹性：数据丢失可以自动恢复；计算的弹性：计算出错重试机制；分片的弹性：可根据需要重新分片。RDD的特点详解-弹性RDD逻辑上是分区的，每个分区的数据是抽象存在的，计算时通过compute函数得到每个分区的数据。如果RDD是通过已有的文件系统构建，则compute函数是读取指定文件系统中的数据，如果RDD是通过其他RDD转换而来，则compute函数是执行转换逻辑将其他RDD的数据进行转换。RDD的特点详解-分区由一个RDD转换到另一个RDD，可以通过算子实现。RDD的操作算子包括两类，一类叫做transformations，它是用来将RDD进行转化，构建RDD的血缘关系；另一类叫做actions，它是用来触发RDD的计算，得到RDD的相关计算结果或者将RDD保存的文件系统中。RDD的特点详解-只读RDDs通过操作算子进行转换，转换得到的新RDD包含了从其他RDDs衍生所必需的信息，RDDs之间维护着这种血缘关系，也称之为依赖。如下图所示，依赖包括两种，一种是窄依赖，RDDs之间分区是一一对应的，另一种是宽依赖，下游RDD的每个分区与上游RDD的每个分区都有关，是多对多的关系。RDD的特点详解-依赖如果在应用程序中多次使用同一个RDD，可以将该RDD缓存起来，该RDD只有在第一次计算的时候会根据血缘关系得到分区的数据，在后续其他地方用到该RDD的时候，会直接从缓存处取而不用再根据血缘关系计算，这样就加速后期的重用。

RDD的特点详解-缓存如下图所示，RDD-1经过一系列的转换后得到RDD-n并保存到hdfs，RDD-1在这一过程中会有个中间结果，如果将其缓存到内存，那么在随后的RDD-1转换到RDD-m这一过程中，就不会计算其之前的RDD-0。RDD的特点详解-缓存1.RDD特点概述

2.RDD的特点详解

RDD的创建创建RDD的三种方式创建方式详解创建RDD的三种方式/01

在RDD中，通常就代表和包含了Spark应用程序的输入源数据。在创建了初始的RDD之后，才可以通过SparkCore提供的transformation算子，对该RDD进行transformation(转换)操作，来获取其他的RDD。创建RDD的三种方式SparkCore为我们提供了三种创建RDD的方式，包括：1.使用程序中的集合创建RDD2.使用本地文件创建RDD3.使用HDFS文件创建RDD创建RDD的三种方式创建方式详解/02

1、使用程序中的集合创建RDD，主要用于进行测试，可以在实际部署到集群运行之前，使用集合构造测试数据，来测试后面的spark应用的流程。

创建方式详解2、使用本地文件创建RDD，主要用于临时性地处理一些存储了大量数据的文件。集群上运行时需要所有集群上都有该文件。

创建方式详解3、使用HDFS文件创建RDD，是最常用的生产环境处理方式，主要可以针对HDFS上存储的大数据，进行离线批处理操作。

创建方式详解1.创建RDD的三种方式

2.创建方式详解

RDD的分区RDD分区介绍RDD分区原则RDD设置分区的方法RDD分区介绍/01

RDD是弹性分布式数据集，通常RDD很大，会被分成很多个分区，分别保存在不同的节点上，作用有二：增加并行度和减少通信开销（连接操作），例如下图：

RDD分区介绍RDD分区原则/02

RDD分区的一个原则是使得分区的个数尽量等于集群中的CPU核心（core）数目。对于不同的Spark部署模式而言（本地模式、Standalone模式、YARN模式、Mesos模式），都可以通过设置spark.default.parallelism这个参数的值，来配置默认的分区数目。RDD分区原则*本地模式：默认为本地机器的CPU数目，若设置了local[N],则默认为N，local[*]则自动判断。*ApacheMesos：默认的分区数为8。*Standalone或YARN：在“集群中所有CPU核心数目总和”和“2”二者中取较大值作为默认值。RDD分区原则RDD设置分区方法/03

创建RDD时手动指定分区个数。使用reparititon方法重新设置分区个数。自定义分区方法：Spark提供了自带的HashPartitioner（哈希分区）与RangePartitioner（区域分区），能够满足大多数应用场景的需求。与此同时，Spark也支持自定义分区方式，即通过提供一个自定义的Partitioner对象来控制RDD的分区方式，从而利用领域知识进一步减少通信开销。RDD设置分区方法1.RDD分区介绍

2.RDD分区原则

3.RDD设置分区方法

窄依赖窄依赖介绍窄依赖对优化的优势窄依赖介绍/01

窄依赖是指1个父RDD分区对应1个子RDD的分区。换句话说，一个父RDD的分区对应于一个子RDD的分区，或者多个父RDD的分区对应于一个子RDD的分区。窄依赖介绍

窄依赖分为两种情况：1个子RDD的分区对应于1个父RDD的分区，比如map，filter，union等算子。1个子RDD的分区对应于N个父RDD的分区，比如co-partionedjoin。窄依赖介绍

窄依赖可以支持在同一个集群Executor上，以pipeline管道形式顺序执行多条命令，例如在执行了map后，紧接着执行filter。分区内的计算收敛，不需要依赖所有分区的数据，可以并行地在不同节点进行计算。所以它的失败回复也更有效，因为它只需要重新计算丢失的parentpartition即可窄依赖介绍窄依赖对优化的优势/021.依赖往往对应着Shuffle操作，需要在运行过程中将同一个父RDD的分区传入到不同的子RDD分区中，中间可能涉及多个节点之间的数据传输；而窄依赖的每个父RDD的分区只会传入到一个子RDD分区中，通常可以在一个节点内完成转换。2.当RDD分区丢失时（某个节点故障），Spark会对数据进行重算窄依赖对优化的优势3.对于窄依赖，由于父RDD的一个分区只对应一个子RDD分区，这样只需要重算和子RDD分区对应的父RDD分区即可，所以这个重算对数据的利用率是100%的。4.对于依赖，重算的父RDD分区对应多个子RDD分区的，这样实际上父RDD中只有一部分的数据是被用于恢复这个丢失的子RDD分区的，另一部分对应子RDD的其他未丢失分区，这就造成了多余的计算；更一般的，宽依赖中子RDD分区通常来自多个父RDD分区，极端情况下，所有的父RDD分区都要进行重新计算。窄依赖对优化的优势1.窄依赖介绍

2.窄依赖对优化的优势

宽依赖宽依赖介绍宽依赖和窄依赖的区别宽依赖介绍/01

宽依赖是指父RDD的每个分区都可能被多个子RDD分区所使用，子RDD分区通常对应所有的父RDD分区，即一个父rdd的数据到多个子rdd（一对多）。

宽依赖介绍

宽依赖分为两种情况：1个父RDD对应非全部多个子RDD分区，比如groupByKey，reduceByKey，sortByKey。1个父RDD对应所有子RDD分区，比如未经协同划分的join。宽依赖介绍

宽依赖需要所有的父分区都是可用的，必须等RDD的parentpartition数据全部ready之后才能开始计算，可能还需要调用类似MapReduce之类的操作进行跨节点传递。从失败恢复的角度看，shuffledependency牵涉RDD各级的多个parentpartition。宽依赖介绍宽依赖和窄依赖的区别/021、窄依赖是将分区聚合到一起，收拢数据，这样就可以考虑到一些算子做此功能比如：map,filter,union,join(父RDD是hash-partitioned),mapPartitions,mapValues；2、而宽依赖则不同，宽依赖将分区数据进行打散分开，走shuffle机制与mapreduce相同。他主要将一些数据进行洗牌和重新分组发牌。比如算子做此功能：groupByKey,join(父RDD不是hash-partitioned),partitionBy，sort宽依赖和窄依赖的区别1.宽依赖介绍

2.宽依赖和窄依赖的区别

RDD的依赖关系RDD依赖关系的本质依赖关系下的数据流视图RDD依赖关系的本质/01

由于RDD是粗粒度的操作数据集，每个Transformation操作都会生成一个新的RDD，所以RDD之间就会形成类似流水线的前后依赖关系；RDD依赖关系的本质

在spark中，RDD之间存在两种类型的依赖关系：窄依赖和宽依赖；如图所示显示了RDD之间的依赖关系。RDD依赖关系的本质依赖关系下的数据流视图/02

如下图是RDD依赖关系下的数据流视图：依赖关系下的数据流视图

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

因此spark划分stage的整体思路是：从后往前推，遇到宽依赖就断开，划分为一个stage；遇到窄依赖就将这个RDD加入该stage中。依赖关系下的数据流视图

在spark中，Task的类型分为2种：ShuffleMapTask和ResultTask；简单来说，DAG的最后一个阶段会为每个结果的partition生成一个ResultTask，即每个Stage里面的Task的数量是由该Stage中最后一个RDD的Partition的数量所决定的,而其余所有阶段都会生成ShuffleMapTask；之所以称之为ShuffleMapTask,是因为它需要将自己的计算结果通过shuffle到下一个stage中；依赖关系下的数据流视图1.RDD依赖关系的本质

2.依赖关系下的数据流视图

RDD的缓存RDD的缓存机制使用缓存RDD的缓存机制/01

默认情况下，RDD只使用一次，用完即扔，再次使用时需要重新计算得到，而持久化操作避免了这里的重复计算，实际测试也显示持久化对性能提升明显。RDD的缓存机制

假设首先进行了RDD0→RDD1→RDD2的计算作业，那么计算结束时，RDD1就已经缓存在系统中了。在进行RDD0→RDD1→RDD3的计算作业时，由于RDD1已经缓存在系统中，因此RDD0→RDD1的转换不会重复进行。RDD的缓存机制

计算作业只须进行RDD1→RDD3的计算就可以了，因此计算速度可以得到很大提升。RDD的缓存机制使用缓存/02

使用如下代码使用缓存：scala>importorg.apache.spark.storage._scala>valrdd1=sc.makeRDD(1to5)scala>rdd1.cache//cache只有一种默认的缓存级别，即MEMORY_ONLYscala>rdd1.persist(StorageLevel.MEMORY_ONLY)使用缓存

Spark会自动监控每个节点上的缓存数据，然后使用least-recently-used(LRU)机制来处理旧的缓存数据。如果想手动清理这些缓存的RDD数据而不是去等待它们被自动清理掉，使用RDD.unpersist()方法。使用缓存1.RDD缓存机制

2.使用缓存RDD的CheckPointCheckPoint介绍RDD的检查点（checkpoint）机制CheckPoint介绍/01

检查点是为了通过lineage（血统）做容错的辅助，lineage过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的RDD开始重做Lineage，就会减少开销。

CheckPoint介绍

在设置检查点之后，该RDD之前的有依赖关系的父RDD都会被销毁，下次调用的时候直接从检查点开始计算。