阅读 1053

一文带你过完Spark RDD的基础概念

前言

上一篇权当吹水了,从这篇开始进入正题。

二、Spark 的内存计算框架(重点😶)

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

Dataset:就是一个集合,存储很多数据.
Distributed:它内部的元素进行了分布式存储,方便于后期进行分布式计算.
Resilient:表示弹性,rdd的数据是可以保存在内存或者是磁盘中.
复制代码

在代码中的表现是这样的,每一个方法所对应的结果都是一个RDD,比如上面的scala代码,下一个RDD的结果会依赖于上一个RDD

我知道到目前大家都没整懂,没事,继续往下看就会懂的😏

2.1 RDD 的五大特性


接下来是在源码中对于RDD的解释,分点中我使用了我觉得(接受反驳👌)较为合理的语句来解释

2.1.1 A list of partitions

一个分区(Partition)列表,组成了该RDD的数据。

这里表示一个rdd有很多分区,每一个分区内部是包含了该rdd的部分数据,spark中任务是以task线程的方式运行, 一个分区就对应一个task线程。

用户可以在创建RDD时指定RDD的分区个数,如果没有指定,那么就会采用默认值。(比如:读取HDFS上数据文件产生的RDD分区数跟block的个数相等)

RDD的分区其实可以简单这样理解,比如说我现在要来一个wordCount,这个文本的大小是300M,那按照我们 HDFS 的套路,每128M是一个block块,那这个300M的文件就是3个block,然后我们的RDD会按照你这个文件的拥有的block块数来决定RDD的分区数,此时RDD的分区数就是3,但是如果我这个文件本身就小于128M呢,那RDD就会默认为2个分区数

2.1.2 A function for computing each split

每个分区的计算函数都算是一个RDD --- Spark中RDD的计算是以分区为单位的,每个RDD都会实现compute函数以达到这个目的.

2.1.3 A list of dependencies on other RDDs

一个rdd会依赖于其他多个rdd --- rdd与rdd之间的依赖关系,spark任务的容错机制就是根据这个特性而来。

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

针对key-value类型的RDD才有分区函数,分区函数其实就是把计算的结果丢到不同的分区中。

当前Spark中实现了两种类型的分区函数,一个是基于哈希的 HashPartitioner,另外一个是基于范围的 RangePartitioner
只有对于key-value的RDD,并且产生shuffle,才会有 Partitioner,非key-value的RDD的 Parititioner 的值是None。

HashPartitioner的套路在之前的 MapReduce 的那篇已经有提到过了,其实大数据的那些分区套路好多都是这个套路,RangePartitioner就是类似规定了多少到多少丢这个分区,多少到多少又丢那个分区这样的玩法。

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

计算任务的位置优先为存储每个Partition的位置 (可自定义)

这里涉及到数据的本地性,数据块位置最优。简单点说也就是说哪里有数据我们就在哪里做计算的意思。

spark任务在调度的时候会优先注意只是优先,而不是必然)考虑存有数据的节点开启计算任务,减少数据的网络传输,提升计算效率。

2.2 基于wordCount分析RDD的5大属性

需求就是HDFS上有一个大小为300M的文件,通过 Spark 实现文件单词统计,最后把结果数据保存到 HDFS 上,代码如下,注意我们使用的是 Scala 而不是 Java 代码

sc.textFile("/words.txt"// 读取数据文件
    .flatMap(_.split(" ")) // 切分每一行,获取所有单词
        .map((_,1)) // 每个单词计为1
            .reduceByKey(_+_) // 相同单词(key)出现的1累加
                .saveAsTextFile("/out"// 保存输出到/out
复制代码

因为我本地没有部署好环境,所以我以下过程不会截图,但是会把操作步骤和结果说明一下

2.2.1 RDD1 : sc.textFile("/words.txt")

把 Spark-shell跑起来,然后我们可以一步一步地把代码运行一下,此时我们可以看到


此时我们就可以执行scala的代码了,第一步先把sc.textFile("/words.txt")给打上去

因为RDD是一个抽象类,所以sc.textFile("/words.txt")的结果由它的子类 MapPartitionsRDD 进行了接收,这个代码会得出RDD的分区结果,我们也可以通过

sc.textFile("/words.txt").partitions
复制代码

查看分区,此时会得出一个数组,这个数组的长度为3(300M的文件会有3个block块,而RDD的分区数是由block块来决定的,注意RDD的分区数不一定都在不同的服务器上。但是如果block块只有1,RDD的分区数就会默认2

sc.textFile("/words.txt").partitions.length
复制代码

执行这个查看数组的长度,必然为3

此时RDD2,RDD3的结果和RDD1的结果都是由 MapPartitionsRDD 进行接收,不同的是map得出来的RDD会是key-value类型而已

2.2.2 RDD4 : sc.textFile("/words.txt").flatMap(.split(" ")).map((,1)).reduceByKey(+)

RDD4的接收类型为 ShuffleRDD,因为此时的结果需要按照key来分组,就必定会产生 shuffle,那shuffle可以先简单这么理解,比如我们现在的wordCount中words.txt只有3个key分别是"zookeeper","kafka","spark",那我shuffle时我就做出这样的规定

此时我们 saveAsTextFile 会得到3个文件,不难发现其实 RDD 有几个分区就会有几个文件了。每个RDD所对应体现的5大特性也已经写在了图右侧。

2.3 RDD 的创建方式

2.3.1 通过已经存在的scala集合去构建

val rdd1=sc.parallelize(List(1,2,3,4,5))
val rdd2=sc.parallelize(Array("zookeeper","kafka","spark"))
val rdd3=sc.makeRDD(List(1,2,3,4))
复制代码

2.3.2 加载外部的数据源去构建

val rdd1=sc.textFile("/words.txt")
复制代码

2.3.3 从已经存在的rdd进行转换生成一个新的rdd

val rdd2=rdd1.flatMap(_.split(" "))
val rdd3=rdd2.map((_,1))
复制代码

2.4 RDD 的算子分类

2.4.1 transformation(转换)

根据已经存在的rdd转换生成一个新的rdd, 它是延迟加载,它不会立即执行

就例如刚刚wordCount中用到的 map,flatMap,reduceByKey

2.4.2 action (动作)

它会真正触发任务的运行。将rdd的计算的结果数据返回给Driver端,或者是保存结果数据到外部存储介质中

就例如刚刚wordCount中用到的 collect,saveAsTextFile 等

2.5 RDD 的常见算子(后面结合代码说明)

2.5.1 transformation算子

转换 含义
map(func) 返回一个新的RDD,该RDD由每一个输入元素经过func函数转换后组成
filter(func) 返回一个新的RDD,该RDD由经过func函数计算后返回值为true的输入元素组成
flatMap(func) 类似于map,但是每一个输入元素可以被映射为0或多个输出元素(所以func应该返回一个序列,而不是单一元素)
mapPartitions(func) 类似于map,但独立地在RDD的每一个分片上运行,因此在类型为T的RDD上运行时,func的函数类型必须是Iterator[T] => Iterator[U]
mapPartitionsWithIndex(func) 类似于mapPartitions,但func带有一个整数参数表示分片的索引值,因此在类型为T的RDD上运行时,func的函数类型必须是(Int, Interator[T]) => Iterator[U]
union(otherDataset) 对源RDD和参数RDD求并集后返回一个新的RDD
intersection(otherDataset) 对源RDD和参数RDD求交集后返回一个新的RDD
distinct([numTasks])) 对源RDD进行去重后返回一个新的RDD
groupByKey([numTasks]) 在一个(K,V)的RDD上调用,返回一个(K, Iterator[V])的RDD
reduceByKey(func, [numTasks]) 在一个(K,V)的RDD上调用,返回一个(K,V)的RDD,使用指定的reduce函数,将相同key的值聚合到一起,与groupByKey类似,reduce任务的个数可以通过第二个可选的参数来设置
sortByKey([ascending], [numTasks]) 在一个(K,V)的RDD上调用,K必须实现Ordered接口,返回一个按照key进行排序的(K,V)的RDD
sortBy(func,[ascending], [numTasks]) 与sortByKey类似,但是更灵活
join(otherDataset, [numTasks]) 在类型为(K,V)和(K,W)的RDD上调用,返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD
cogroup(otherDataset, [numTasks]) 在类型为(K,V)和(K,W)的RDD上调用,返回一个(K,(Iterable,Iterable))类型的RDD
coalesce(numPartitions) 减少 RDD 的分区数到指定值。
repartition(numPartitions) 重新给 RDD 分区
repartitionAndSortWithinPartitions(partitioner) 重新给 RDD 分区,并且每个分区内以记录的 key 排序

2.5.2 action算子

动作 含义
reduce(func) reduce将RDD中元素前两个传给输入函数,产生一个新的return值,新产生的return值与RDD中下一个元素(第三个元素)组成两个元素,再被传给输入函数,直到最后只有一个值为止。
collect() 在驱动程序中,以数组的形式返回数据集的所有元素
count() 返回RDD的元素个数
first() 返回RDD的第一个元素(类似于take(1))
take(n) 返回一个由数据集的前n个元素组成的数组
takeOrdered(n, [ordering]) 返回自然顺序或者自定义顺序的前 n 个元素
saveAsTextFile(path) 将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统,对于每个元素,Spark将会调用toString方法,将它装换为文件中的文本
saveAsSequenceFile(path) 将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下,可以使HDFS或者其他Hadoop支持的文件系统。
saveAsObjectFile(path) 将数据集的元素,以 Java 序列化的方式保存到指定的目录下
countByKey() 针对(K,V)类型的RDD,返回一个(K,Int)的map,表示每一个key对应的元素个数。
foreach(func) 在数据集的每一个元素上,运行函数func
foreachPartition(func) 在数据集的每一个分区上,运行函数func

2.6 常见算子的代码说明

好吧我忘了我本地没环境了···

这里只提醒一下,执行了transformation算子操作其实并没有真正意义上的得出结果,要跑collect()才真正的开始计算

2.6.1 注意:repartition 和 coalesce

val rdd1 = sc.parallelize(1 to 10,3)
//打印rdd1的分区数
rdd1.partitions.size

//利用repartition改变rdd1分区数
//减少分区
rdd1.repartition(2).partitions.size

//增加分区
rdd1.repartition(4).partitions.size

//利用coalesce改变rdd1分区数
//减少分区
rdd1.coalesce(2).partitions.size
复制代码

repartition : 重新分区,有shuffle,可以用来处理小文件的问题

coalesce : 合并分区 / 减少分区,默认不shuffle

默认 coalesce 不能扩大分区数量。除非添加true的参数,或者使用repartition。

适用场景:

  1. 如果要shuffle,都用 repartition
  2. 不需要shuffle,仅仅是做分区的合并,coalesce
  3. repartition常用于扩大分区。

2.6.2 注意:map、mapPartitions 和 mapPartitionsWithIndex

map:用于遍历RDD,将函数f应用于每一个元素,返回新的RDD(transformation算子)。

mapPartitions:用于遍历操作RDD中的每一个分区,返回生成一个新的RDD(transformation算子)。

总结:
如果在映射的过程中需要频繁创建额外的对象,使用mapPartitions要比map高效
比如,将RDD中的所有数据通过JDBC连接写入数据库,如果使用map函数,可能要为每一个元素都创建一个connection,这样开销很大,如果使用mapPartitions,那么只需要针对每一个分区建立一个connection。

2.6.3 注意:foreach、foreachPartition

foreach:用于遍历RDD,将函数f应用于每一个元素,无返回值(action算子)。

foreachPartition: 用于遍历操作RDD中的每一个分区。无返回值(action算子)。

总结:
一般使用mapPartitions或者foreachPartition算子比map和foreach更加高效,推荐使用。

所以我们可以使用 foreachPartition 算子实现

2.7 RDD的依赖关系

RDD和它依赖的父RDD的关系有两种不同的类型:窄依赖(narrow dependency)和宽依赖(wide dependency)

2.7.1 窄依赖

窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用,比如map/flatMap/filter/union等等,且所有的窄依赖不会产生shuffle

2.7.2 宽依赖

宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition,比如reduceByKey/sortByKey/groupBy/groupByKey/join等等
。所有的宽依赖会产生shuffle

上图也可以看出join操作分为宽依赖和窄依赖,如果RDD有相同的partitioner,那么将不会引起shuffle,这种join是窄依赖,反之就是宽依赖

2.8 lineage

lineage翻译过来是血统的意思。还是之前的那张图。


RDD的 Lineage 会记录RDD的元数据信息和转换行为,lineage保存了RDD的依赖关系,当该RDD的部分分区数据丢失时,它可以根据这些信息来重新运算和恢复丢失的数据分区。

不过需要注意RDD只支持 粗粒度转换 (即只记录单个块上执行的单个操作),比如此时如果我是RDD4的0号莫得了,我就要重新拿到RDD3的所有数据,然后重新reducebyKey一次,这样才能恢复结果,这种宽依赖需要经过shuffle的操作,恢复起来的成本就会高很多

值得再提的是:我们其实不需要人为干预分区数据的恢复,程序自身就可以帮我们根据 RDD 的血统关系自行恢复

2.9 RDD 的缓存机制

可以把一个rdd的数据缓存起来,后续有其他的job需要用到该rdd的结果数据,可以直接从缓存中获取得到,避免了重复计算。缓存是加快后续对该数据的访问操作。

比如上图中我只要对RDD2的结果做一个缓存,到时候假如用的上,恢复起来就非常方便

2.9.1 如何对 RDD 设置缓存

RDD通过 persist 方法或 cache 方法可以将前面的计算结果缓存。但是需要注意并不是这两个方法被调用时立即缓存,而是触发后面的action时,该RDD将会被缓存在计算节点的内存中,并供后面重用

我们可以看看去RDD的源码汇总找出来look look,这俩方法是放在了一起的


通过查看源码发现cache最终也是调用了persist方法,默认的缓存级别就是 MEMORY_ONLY 都是仅在内存存储一份,Spark的存储级别还有好多种,存储级别在object StorageLevel中定义的

StorageLevel 在Scala中是 object (注意这里我用的是小写,object 是单例对象的意思,而非Java中的Object),这里面还有大量不同的存储级别,这里就不展开了,这些英文其实也不难懂

cache和persist区别是:cache: 默认是把数据缓存在内存中,其本质就是调用persist方法,而persist:可以把数据缓存在内存或者是磁盘,有丰富的缓存级别,这些缓存级别都被定义在StorageLevel这个object中。

2.9.2 缓存的使用时机

当第一次使用RDD2做相应的算子操作得到RDD3的时候,就会从RDD1开始计算,先读取HDFS上的文件,然后对RDD1 做对应的算子操作得到RDD2,再由RDD2计算之后得到RDD3。同样为了计算得到RDD4,前面的逻辑会被重新计算。

默认情况下多次对一个RDD执行算子操作, RDD都会对这个RDD及之前的父RDD全部重新计算一次。 这种情况在实际开发代码的时候会经常遇到,但是我们一定要避免一个RDD重复计算多次,否则会导致性能急剧降低

为了获取得到一个RDD的结果数据,经过了大量的算子操作或者是计算逻辑比较复杂,也就是某个RDD的数据来之不易的时候,就可以设置缓存

总结:可以把多次使用到的RDD,也就是公共RDD进行持久化,避免后续需要,再次重新计算,提升效率。

2.9.3 清除缓存数据

  1. 自动清除 : 一个application应用程序结束之后,对应的缓存数据也就自动清除

  2. 手动清除 : 调用 RDD 的 unpersist 方法

虽然我们可以对 RDD 的数据进行缓存,保存在内存或者是磁盘中,之后就可以直接从内存或者磁盘中获取得到,但是注意这个方式不是特别安全

cache 它是直接把数据保存在内存中,后续操作起来速度比较快,直接从内存中获取得到。但这种方式很不安全,由于服务器挂掉或者是进程终止,会导致数据的丢失。

persist 它可以把数据保存在本地磁盘中,后续可以从磁盘中获取得到该数据,但它也不是特别安全,由于系统管理员一些误操作删除了,或者是磁盘损坏,也有可能导致数据的丢失。

那么我们有没有一种更为安全的方式呢?

2.10 RDD 的 checkpoint 机制

checkpoint 提供了一种相对而言更加可靠的数据持久化方式。它是把数据保存在分布式文件系统,比如HDFS上。这里就是利用了HDFS高可用性,高容错性(多副本)来最大程度保证数据的安全性。

2.10.1 如何设置checkpoint

1.在 HDFS 上设置一个 checkpoint 目录

sc.setCheckpointDir("hdfs://node1:9000/checkpoint"
复制代码

2.对需要做 checkpoint 操作的rdd调用 checkpoint 方法

val rdd1=sc.textFile("/words.txt")
rdd1.checkpoint
val rdd2=rdd1.flatMap(_.split(" ")) 
复制代码

3.最后需要有一个 action 操作去触发任务的运行(checkpoint操作要执行需要有一个action操作,一个action操作对应后续的一个job。该job执行完成之后,它会再次单独开启另外一个job来执行 rdd1.checkpoint操作。

对checkpoint在使用的时候进行优化,在调用checkpoint操作之前,可以先来做一个cache操作,缓存对应rdd的结果数据,后续就可以直接从cache中获取到rdd的数据写入到指定checkpoint目录中

rdd2.collect
复制代码

所以我们总结一下 cache、persist、checkpoint 三者区别

cache和persist

cache默认数据缓存在内存中
persist可以把数据保存在内存或者磁盘中
后续要触发 cache 和 persist 持久化操作,需要有一个action操作
它不会开启其他新的任务,一个action操作就对应一个job 
它不会改变RDD的依赖关系,程序运行完成后对应的缓存数据就自动消失
复制代码

checkpoint

可以把数据持久化写入到 HDFS 上
后续要触发checkpoint持久化操作,需要有一个action操作,后续会开启新的job执行checkpoint操作
它会改变RDD的依赖关系,后续数据丢失了不能够在通过血统进行数据的恢复。
    (因为它判断你已经持久化到 HDFS 中所以把依赖关系删除了)
程序运行完成后对应的checkpoint数据就不会消失
复制代码

2.11 DAG 有向无环图生成

DAG(Directed Acyclic Graph) 叫做有向无环图(有方向,无闭环,代表着数据的流向),原始的RDD通过一系列的转换就形成了DAG。

这个就是我们的 wordCount 例子里生成的 DAG,这个图我们可以在上一讲中提到的web界面中的 Running Application 的 spark-shell ---> Completed Jobs 的 Description 中查看

2.11.1 DAG划分stage

stage是什么: 一个Job会被拆分为多组Task,每组任务被称为一个stage,stage表示不同的调度阶段,一个spark job会对应产生很多个stage

stage类型一共有2种:

  1. ShuffleMapStage: 最后一个shuffle之前的所有变换叫ShuffleMapStage,它对应的task是shuffleMapTask
  2. ResultStage: 最后一个shuffle之后的操作叫ResultStage,它是最后一个Stage.它对应的task是ResultTask

2.11.2 为啥我们要划分stage

根据RDD之间依赖关系的不同将DAG划分成不同的Stage(调度阶段)
对于窄依赖,partition的转换处理在一个Stage中完成计算
对于宽依赖,由于有Shuffle的存在,只能在parent RDD处理完成后,才能开始接下来的计算,

由于划分完stage之后,在同一个stage中只有窄依赖,没有宽依赖,可以实现流水线计算,
stage中的每一个分区对应一个task,在同一个stage中就有很多可以并行运行的task。

2.11.3 如何划分stage

我们划分stage的依据就是宽依赖

  1. 首先根据rdd的算子操作顺序生成DAG有向无环图,接下里从最后一个 RDD 往前推,创建一个新的stage,把该rdd加入到该stage中,它是最后一个stage。

  2. 在往前推的过程中运行遇到了窄依赖就把该 RDD 加入到本stage中,如果遇到了宽依赖,就从宽依赖的位置切开,那么最后一个stage也就被划分出来了。

  3. 重新创建一个新的stage,按照第二个步骤继续往前推,一直到最开始的 RDD,整个划分stage也就划分结束了

2.11.4 stage与stage之间的关系

划分完stage之后,每一个stage中有很多可以并行运行的task,后期把每一个stage中的task封装在一个taskSet集合中,最后把一个一个的taskSet集合提交到worker节点上的executor进程中运行。

RDD 与 RDD 之间存在依赖关系,stage与stage之前也存在依赖关系,前面stage中的task先运行,运行完成了再运行后面stage中的task,也就是说后面stage中的task输入数据是前面stage中task的输出结果数据。

finally

以上我们就是把 RDD 的一些基础知识点给简单地过了一遍,还有一些更加深入,更加细致的方面我们要放到 Spark Core,Spark Streaming中去阐述,感兴趣的朋友可以继续地关注一下

关注下面的标签,发现更多相似文章
评论