spark学习进度13（spark总体介绍）

如何生成 RDD ?

生成 RDD 的常见方式有三种

从本地集合创建
从外部数据集创建
从其它 RDD 衍生

通过外部数据集创建 RDD, 是通过 Hadoop 或者其它外部数据源的 SDK 来进行数据读取, 同时如果外部数据源是有分片的话, RDD 会将分区与其分片进行对照

通过其它 RDD 衍生的话, 其实本质上就是通过不同的算子生成不同的 RDD 的子类对象, 从而控制 compute 函数的行为来实现算子功能

生成哪些 RDD ?

不同的算子生成不同的 RDD, 生成 RDD 的类型取决于算子, 例如 map 和 flatMap 都会生成 RDD 的子类 MapPartitions 的对象

如何计算 RDD 中的数据 ?

虽然前面我们提到过 RDD 是偏向计算的, 但是其实 RDD 还只是表示数据, 纵观 RDD 的五大属性中有三个是必须的, 分别如下

Partitions List 分区列表
Compute function 计算函数
Dependencies 依赖

虽然计算函数是和计算有关的, 但是只有调用了这个函数才会进行计算, RDD 显然不会自己调用自己的 Compute 函数, 一定是由外部调用的, 所以 RDD 更多的意义是用于表示数据集以及其来源, 和针对于数据的计算

所以如何计算 RDD 中的数据呢? 一定是通过其它的组件来计算的, 而计算的规则, 由 RDD 中的 Compute 函数来指定, 不同类型的 RDD 子类有不同的 Compute 函数什么是 RDD 之间的依赖关系?

什么是关系(依赖关系) ?

从算子视角上来看, splitRDD 通过 map 算子得到了 tupleRDD, 所以 splitRDD 和 tupleRDD 之间的关系是 map

但是仅仅这样说, 会不够全面, 从细节上来看, RDD 只是数据和关于数据的计算, 而具体执行这种计算得出结果的是一个神秘的其它组件, 所以, 这两个 RDD 的关系可以表示为 splitRDD 的数据通过 map 操作, 被传入 tupleRDD, 这是它们之间更细化的关系

但是 RDD 这个概念本身并不是数据容器, 数据真正应该存放的地方是 RDD 的分区, 所以如果把视角放在数据这一层面上的话, 直接讲这两个 RDD 之间有关系是不科学的, 应该从这两个 RDD 的分区之间的关系来讨论它们之间的关系
那这些分区之间是什么关系?

如果仅仅说 splitRDD 和 tupleRDD 之间的话, 那它们的分区之间就是一对一的关系

但是 tupleRDD 到 reduceRDD 呢? tupleRDD 通过算子 reduceByKey 生成 reduceRDD, 而这个算子是一个 Shuffle 操作, Shuffle 操作的两个 RDD 的分区之间并不是一对一, reduceByKey 的一个分区对应 tupleRDD 的多个分区

reduceByKey 算子会生成 ShuffledRDD

reduceByKey 是由算子 combineByKey 来实现的, combineByKey 内部会创建 ShuffledRDD 返回, 具体的代码请大家通过 IDEA 来进行查看, 此处不再截图, 而整个 reduceByKey 操作大致如下过程

去掉两个 reducer 端的分区, 只留下一个的话, 如下

所以, 对于 reduceByKey 这个 Shuffle 操作来说, reducer 端的一个分区, 会从多个 mapper 端的分区拿取数据, 是一个多对一的关系

至此为止, 出现了两种分区见的关系了, 一种是一对一, 一种是多对一

整体上的流程图

物理图的作用是什么?

问题一: 物理图的意义是什么?

物理图解决的其实就是 RDD 流程生成以后, 如何计算和运行的问题, 也就是如何把 RDD 放在集群中执行的问题

问题二: 如果要确定如何运行的问题, 则需要先确定集群中有什么组件

首先集群中物理元件就是一台一台的机器
其次这些机器上跑的守护进程有两种: Master, Worker
- 每个守护进程其实就代表了一台机器, 代表这台机器的角色, 代表这台机器和外界通信
- 例如我们常说一台机器是 Master, 其含义是这台机器中运行了一个 Master 守护进程, 如果一台机器运行了 Master 的同时又运行了 Worker, 则说这台机器是 Master 也可以, 说它是 Worker 也行
真正能运行 RDD 的组件是: Executor, 也就是说其实 RDD 最终是运行在 Executor 中的, 也就是说, 无论是 Master 还是 Worker 其实都是用于管理 Executor 和调度程序的

结论是 RDD 一定在 Executor 中计算, 而 Master 和 Worker 负责调度和管理 Executor

问题三: 物理图的生成需要考虑什么问题?

要计算 RDD, 不仅要计算, 还要很快的计算 → 优化性能
要考虑容错, 容错的常见手段是缓存 → RDD 要可以缓存

结论是在生成物理图的时候, 不仅要考虑效率问题, 还要考虑一种更合适的方式, 让 RDD 运行的更好

谁来计算 RDD ?

问题一: RDD 是什么, 用来做什么 ?

回顾一下 RDD 的五个属性

A list of partitions
A function for computing each split
A list of dependencies on other RDDs
Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

简单的说就是: 分区列表, 计算函数, 依赖关系, 分区函数, 最佳位置

分区列表, 分区函数, 最佳位置, 这三个属性其实说的就是数据集在哪, 在哪更合适, 如何分区
计算函数和依赖关系, 这两个属性其实说的是数据集从哪来

所以结论是 RDD 是一个数据集的表示, 不仅表示了数据集, 还表示了这个数据集从哪来, 如何计算

但是问题是, 谁来计算 ? 如果为一台汽车设计了一个设计图, 那么设计图自己生产汽车吗 ?

问题二: 谁来计算 ?

前面我们明确了两件事, RDD 在哪被计算? 在 Executor 中. RDD 是什么? 是一个数据集以及其如何计算的图纸.

直接使用 Executor 也是不合适的, 因为一个计算的执行总是需要一个容器, 例如 JVM 是一个进程, 只有进程中才能有线程, 所以这个计算 RDD 的线程应该运行在一个进程中, 这个进程就是 Exeutor, Executor 有如下两个职责

和 Driver 保持交互从而认领属于自己的任务
接受任务后, 运行任务

所以, 应该由一个线程来执行 RDD 的计算任务, 而 Executor 作为执行这个任务的容器, 也就是一个进程, 用于创建和执行线程, 这个执行具体计算任务的线程叫做 Task

问题三: Task 该如何设计 ?

第一个想法是每个 RDD 都由一个 Task 来计算第二个想法是一整个逻辑执行图中所有的 RDD 都由一组 Task 来执行第三个想法是分阶段执行

第一个想法: 为每个 RDD 的分区设置一组 Task: 大概就是每个 RDD 都有三个 Task, 每个 Task 对应一个 RDD 的分区, 执行一个分区的数据的计算

但是这么做有一个非常难以解决的问题, 就是数据存储的问题, 例如 Task 1, 4, 7, 10, 13, 16 在同一个流程上, 但是这些 Task 之间需要交换数据, 因为这些 Task 可能被调度到不同的机器上上, 所以 Task1 执行完了数据以后需要暂存, 后交给 Task4 来获取

这只是一个简单的逻辑图, 如果是一个复杂的逻辑图, 会有什么表现? 要存储多少数据? 无论是放在磁盘还是放在内存中, 是不是都是一种极大的负担?
第二个想法: 让数据流动: 很自然的, 第一个想法的问题是数据需要存储和交换, 那不存储不就好了吗? 对, 可以让数据流动起来

第一个要解决的问题就是, 要为数据创建管道(Pipeline), 有了管道, 就可以流动

简单来说, 就是为所有的 RDD 有关联的分区使用同一个 Task, 但是就没问题了吗? 请关注红框部分

这两个 RDD 之间是 Shuffle 关系, 也就是说, 右边的 RDD 的一个分区可能依赖左边 RDD 的所有分区, 这样的话, 数据在这个地方流不动了, 怎么办?
第三个想法: 划分阶段: 既然在 Shuffle 处数据流不动了, 那就可以在这个地方中断一下, 后面 Stage 部分详解

如何划分阶段 ?

为了减少执行任务, 减少数据暂存和交换的机会, 所以需要创建管道, 让数据沿着管道流动, 其实也就是原先每个 RDD 都有一组 Task, 现在改为所有的 RDD 共用一组 Task, 但是也有问题, 问题如下

就是说, 在 Shuffle 处, 必须断开管道, 进行数据交换, 交换过后, 继续流动, 所以整个流程可以变为如下样子

把 Task 断开成两个部分, Task4 可以从 Task 1, 2, 3 中获取数据, 后 Task4 又作为管道, 继续让数据在其中流动

但是还有一个问题, 说断开就直接断开吗? 不用打个招呼的呀? 这个断开即没有道理, 也没有规则, 所以可以为这个断开增加一个概念叫做阶段, 按照阶段断开, 阶段的英文叫做 Stage, 如下

所以划分阶段的本身就是设置断开点的规则, 那么该如何划分阶段呢?

第一步, 从最后一个 RDD, 也就是逻辑图中最右边的 RDD 开始, 向前滑动 Stage 的范围, 为 Stage0
第二步, 遇到 ShuffleDependency 断开 Stage, 从下一个 RDD 开始创建新的 Stage, 为 Stage1
第三步, 新的 Stage 按照同样的规则继续滑动, 直到包裹所有的 RDD

总结来看, 就是针对于宽窄依赖来判断, 一个 Stage 中只有窄依赖, 因为只有窄依赖才能形成数据的 Pipeline.

如果要进行 Shuffle 的话, 数据是流不过去的, 必须要拷贝和拉取. 所以遇到 RDD 宽依赖的两个 RDD 时, 要切断这两个 RDD 的 Stage.

这样一个 RDD 依赖的链条, 我们称之为 RDD 的血统, 其中有宽依赖也有窄依赖

数据怎么流动 ?

val sc = ...

val textRDD = sc.parallelize(Seq("Hadoop Spark", "Hadoop Flume", "Spark Sqoop"))
val splitRDD = textRDD.flatMap(_.split(" "))
val tupleRDD = splitRDD.map((_, 1))
val reduceRDD = tupleRDD.reduceByKey(_ + _)
val strRDD = reduceRDD.map(item => s"${item._1}, ${item._2}")

strRDD.collect.foreach(item => println(item))

上述代码是这个章节我们一直使用的代码流程, 如下是其完整的逻辑执行图

如果放在集群中运行, 通过 WebUI 可以查看到如下 DAG 结构

Step 1: 从 ResultStage 开始执行: 最接近 Result 部分的 Stage id 为 0, 这个 Stage 被称之为 ResultStage

由代码可以知道, 最终调用 Action 促使整个流程执行的是最后一个 RDD, strRDD.collect, 所以当执行 RDD 的计算时候, 先计算的也是这个 RDD
Step 2: RDD 之间是有关联的: 前面已经知道, 最后一个 RDD 先得到执行机会, 先从这个 RDD 开始执行, 但是这个 RDD 中有数据吗 ? 如果没有数据, 它的计算是什么? 它的计算是从父 RDD 中获取数据, 并执行传入的算子的函数

简单来说, 从产生 Result 的地方开始计算, 但是其 RDD 中是没数据的, 所以会找到父 RDD 来要数据, 父 RDD 也没有数据, 继续向上要, 所以, 计算从 Result 处调用, 但是从整个逻辑图中的最左边 RDD 开始, 类似一个递归的过程