-
从部署图了解
Spark部署了什么, 有什么组件运行在集群中 -
通过对
WordCount案例的解剖, 来理解执行逻辑计划的生成 -
通过对逻辑执行计划的细化, 理解如何生成物理计划
标签：常用 组件 extern bool 分数 字符串表 优化 做什么 hdfs
如何判断宽窄依赖：
===================================
从部署图了解 Spark 部署了什么, 有什么组件运行在集群中
通过对 WordCount 案例的解剖, 来理解执行逻辑计划的生成
通过对逻辑执行计划的细化, 理解如何生成物理计划
|
如无特殊说明, 以下部分均针对于 |
在 Spark 部分的底层执行逻辑开始之前, 还是要先认识一下 Spark 的部署情况, 根据部署情况, 从而理解如何调度.
针对于上图, 首先可以看到整体上在集群中运行的角色有如下几个:
Master Daemon
负责管理 Master 节点, 协调资源的获取, 以及连接 Worker 节点来运行 Executor, 是 Spark 集群中的协调节点
Worker Daemon
Workers 也称之为叫 Slaves, 是 Spark 集群中的计算节点, 用于和 Master 交互并管理 Executor.
当一个 Spark Job 提交后, 会创建 SparkContext, 后 Worker 会启动对应的 Executor.
Executor Backend
上面有提到 Worker 用于控制 Executor 的启停, 其实 Worker 是通过 Executor Backend 来进行控制的, Executor Backend 是一个进程(是一个 JVM 实例), 持有一个 Executor 对象
另外在启动程序的时候, 有三种程序需要运行在集群上:
Driver
Driver 是一个 JVM 实例, 是一个进程, 是 Spark Application 运行时候的领导者, 其中运行了 SparkContext.
Driver 控制 Job 和 Task, 并且提供 WebUI.
Executor
Executor 对象中通过线程池来运行 Task, 一个 Executor 中只会运行一个 Spark Application 的 Task, 不同的 Spark Application 的 Task 会由不同的 Executor 来运行
对于上面代码中的 reduceRDD 如果使用 toDebugString 打印调试信息的话, 会显式如下内容
(6) MapPartitionsRDD[4] at map at WordCount.scala:20 []
| ShuffledRDD[3] at reduceByKey at WordCount.scala:19 []
+-(6) MapPartitionsRDD[2] at map at WordCount.scala:18 []
| MapPartitionsRDD[1] at flatMap at WordCount.scala:17 []
| ParallelCollectionRDD[0] at parallelize at WordCount.scala:16 []根据这段内容, 大致能得到这样的一张逻辑执行图
其实 RDD 并没有什么严格的逻辑执行图和物理执行图的概念, 这里也只是借用这个概念, 从而让整个 RDD 的原理可以解释, 好理解.
对于 RDD 的逻辑执行图, 起始于第一个入口 RDD 的创建, 结束于 Action 算子执行之前, 主要的过程就是生成一组互相有依赖关系的 RDD, 其并不会真的执行, 只是表示 RDD 之间的关系, 数据的流转过程.
当触发 Action 执行的时候, 这一组互相依赖的 RDD 要被处理, 所以要转化为可运行的物理执行图, 调度到集群中执行.
因为大部分 RDD 是不真正存放数据的, 只是数据从中流转, 所以, 不能直接在集群中运行 RDD, 要有一种 Pipeline 的思想, 需要将这组 RDD 转为 Stage 和 Task, 从而运行 Task, 优化整体执行速度.
以上的逻辑执行图会生成如下的物理执行图, 这一切发生在 Action 操作被执行时.
从上图可以总结如下几个点
在第一个 Stage 中, 每一个这样的执行流程是一个 Task, 也就是在同一个 Stage 中的所有 RDD 的对应分区, 在同一个 Task 中执行
Stage 的划分是由 Shuffle 操作来确定的, 有 Shuffle 的地方, Stage 断开
如何生成 RDD
如何控制 RDD 之间的关系
本章要回答如下三个问题
如何生成 RDD
生成什么 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}")
println(strRDD.toDebugString)
strRDD.collect.foreach(item => println(item))在 Action 调用之前, 会生成一系列的 RDD, 这些 RDD 之间的关系, 其实就是整个逻辑计划
例如上述代码, 如果生成逻辑计划的, 会生成如下一些 RDD, 这些 RDD 是相互关联的, 这些 RDD 之间, 其实本质上生成的就是一个 计算链
接下来, 采用迭代渐进式的方式, 一步一步的查看一下整体上的生成过程
textFile 算子的背后研究 RDD 的功能或者表现的时候, 其实本质上研究的就是 RDD 中的五大属性, 因为 RDD 透过五大属性来提供功能和表现, 所以如果要研究 textFile 这个算子, 应该从五大属性着手, 那么第一步就要看看生成的 RDD 是什么类型的 RDD
textFile 生成的是 HadoopRDD
|
除了上面这一个步骤以外, 后续步骤将不再直接基于代码进行讲解, 因为从代码的角度着手容易迷失逻辑, 这个章节的初心有两个, 一个是希望大家了解 Spark 的内部逻辑和原理, 另外一个是希望大家能够通过本章学习具有代码分析的能力 |
HadoopRDD 的 Partitions 对应了 HDFS 的 Blocks
其实本质上每个 HadoopRDD 的 Partition 都是对应了一个 Hadoop 的 Block, 通过 InputFormat 来确定 Hadoop中的 Block 的位置和边界, 从而可以供一些算子使用
HadoopRDD 的 compute 函数就是在读取 HDFS 中的 Block
本质上, compute 还是依然使用 InputFormat 来读取 HDFS 中对应分区的 Block
textFile 这个算子生成的其实是一个 MapPartitionsRDD
textFile 这个算子的作用是读取 HDFS 上的文件, 但是 HadoopRDD 中存放是一个元组, 其 Key 是行号, 其 Value 是 Hadoop 中定义的 Text 对象, 这一点和 MapReduce 程序中的行为是一致的
但是并不适合 Spark 的场景, 所以最终会通过一个 map 算子, 将 (LineNum, Text) 转为 String 形式的一行一行的数据, 所以最终 textFile 这个算子生成的 RDD 并不是 HadoopRDD, 而是一个 MapPartitionsRDD
map 算子的背后
map 算子生成了 MapPartitionsRDD
由源码可知, 当 val rdd2 = rdd1.map() 的时候, 其实生成的新 RDD 是 rdd2, rdd2 的类型是 MapPartitionsRDD, 每个 RDD 中的五大属性都会有一些不同, 由 map 算子生成的 RDD 中的计算函数, 本质上就是遍历对应分区的数据, 将每一个数据转成另外的形式
MapPartitionsRDD 的计算函数是 collection.map( function )
真正运行的集群中的处理单元是 Task, 每个 Task 对应一个 RDD 的分区, 所以 collection 对应一个 RDD 分区的所有数据, 而这个计算的含义就是将一个 RDD 的分区上所有数据当作一个集合, 通过这个 Scala 集合的 map 算子, 来执行一个转换操作, 其转换操作的函数就是传入 map 算子的 function
传入 map 算子的函数会被清理
这个清理主要是处理闭包中的依赖, 使得这个闭包可以被序列化发往不同的集群节点运行
flatMap 算子的背后
flatMap 和 map 算子其实本质上是一样的, 其步骤和生成的 RDD 都是一样, 只是对于传入函数的处理不同, map 是 collect.map( function ) 而 flatMap 是 collect.flatMap( function )
从侧面印证了, 其实 Spark 中的 flatMap 和 Scala 基础中的 flatMap 其实是一样的
textRDD → splitRDD → tupleRDD由 textRDD 到 splitRDD 再到 tupleRDD 的过程, 其实就是调用 map 和 flatMap 算子生成新的 RDD 的过程, 所以如下图所示, 就是这个阶段所生成的逻辑计划
RDD ?RDD ?RDD 中的数据 ?讨论什么是 RDD 之间的依赖关系
继而讨论 RDD 分区之间的关系
最后确定 RDD 之间的依赖关系分类
完善案例的逻辑关系图
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 算子会生成 ShuffledRDDreduceByKey 是由算子 combineByKey 来实现的, combineByKey 内部会创建 ShuffledRDD 返回, 具体的代码请大家通过 IDEA 来进行查看, 此处不再截图, 而整个 reduceByKey 操作大致如下过程
去掉两个 reducer 端的分区, 只留下一个的话, 如下
所以, 对于 reduceByKey 这个 Shuffle 操作来说, reducer 端的一个分区, 会从多个 mapper 端的分区拿取数据, 是一个多对一的关系
至此为止, 出现了两种分区见的关系了, 一种是一对一, 一种是多对一
上个小节通过例子演示了 RDD 的分区间的关系有两种形式
一对一, 一般是直接转换
多对一, 一般是 Shuffle
本小节会说明如下问题:
如果分区间得关系是一对一或者多对一, 那么这种情况下的 RDD 之间的关系的正式命名是什么呢?
RDD 之间的依赖关系, 具体有几种情况呢?
假如 rddB = rddA.transform(…?), 如果 rddB 中一个分区依赖 rddA 也就是其父 RDD 的少量分区, 这种 RDD 之间的依赖关系称之为窄依赖
换句话说, 子 RDD 的每个分区依赖父 RDD 的少量个数的分区, 这种依赖关系称之为窄依赖
举个栗子
val sc = ...
val rddA = sc.parallelize(Seq(1, 2, 3))
val rddB = sc.parallelize(Seq("a", "b"))
/**
* 运行结果: (1,a), (1,b), (2,a), (2,b), (3,a), (3,b)
*/
rddA.cartesian(rddB).collect().foreach(println(_))上述代码的 cartesian 是求得两个集合的笛卡尔积
上述代码的运行结果是 rddA 中每个元素和 rddB 中的所有元素结合, 最终的结果数量是两个 RDD 数量之和
rddC 有两个父 RDD, 分别为 rddA 和 rddB
对于 cartesian 来说, 依赖关系如下
上述图形中清晰展示如下现象
rddC 中的分区数量是两个父 RDD 的分区数量之乘积
rddA 中每个分区对应 rddC 中的两个分区 (因为 rddB 中有两个分区), rddB 中的每个分区对应 rddC 中的三个分区 (因为 rddA 有三个分区)
它们之间是窄依赖, 事实上在 cartesian 中也是 NarrowDependency 这个所有窄依赖的父类的唯一一次直接使用, 为什么呢?
因为所有的分区之间是拷贝关系, 并不是 Shuffle 关系
rddC 中的每个分区并不是依赖多个父 RDD 中的多个分区
rddC 中每个分区的数量来自一个父 RDD 分区中的所有数据, 是一个 FullDependence, 所以数据可以直接从父 RDD 流动到子 RDD
不存在一个父 RDD 中一部分数据分发过去, 另一部分分发给其它的 RDD
|
并没有所谓的宽依赖, 宽依赖应该称作为 |
在 ShuffleDependency 的类声明上如下写到
Represents a dependency on the output of a shuffle stage.上面非常清楚的说道, 宽依赖就是 Shuffle 中的依赖关系, 换句话说, 只有 Shuffle 产生的地方才是宽依赖
那么宽窄依赖的判断依据就非常简单明确了, 是否有 Shuffle ?
举个 reduceByKey 的例子, rddB = rddA.reduceByKey( (curr, agg) ⇒ curr + agg ) 会产生如下的依赖关系
rddB 的每个分区都几乎依赖 rddA 的所有分区
对于 rddA 中的一个分区来说, 其将一部分分发给 rddB 的 p1, 另外一部分分发给 rddB 的 p2, 这不是数据流动, 而是分发
其实分辨宽窄依赖的本身就是在分辨父子 RDD 之间是否有 Shuffle, 大致有以下的方法
如果是 Shuffle, 两个 RDD 的分区之间不是单纯的数据流动, 而是分发和复制
一般 Shuffle 的子 RDD 的每个分区会依赖父 RDD 的多个分区
但是这样判断其实不准确, 如果想分辨某个算子是否是窄依赖, 或者是否是宽依赖, 则还是要取决于具体的算子, 例如想看 cartesian 生成的是宽依赖还是窄依赖, 可以通过如下步骤
查看 map 算子生成的 RDD
进去 RDD 查看 getDependence 方法
RDD 的逻辑图本质上是对于计算过程的表达, 例如数据从哪来, 经历了哪些步骤的计算
每一个步骤都对应一个 RDD, 因为数据处理的情况不同, RDD 之间的依赖关系又分为窄依赖和宽依赖 *
常见的窄依赖其实也是有分类的, 而且宽窄以来不太容易分辨, 所以通过本章, 帮助同学明确窄依赖的类型
其实 RDD 中默认的是 OneToOneDependency, 后被不同的 RDD 子类指定为其它的依赖类型, 常见的一对一依赖是 map 算子所产生的依赖, 例如 rddB = rddA.map(…?)
每个分区之间一一对应, 所以叫做一对一窄依赖
Range 窄依赖其实也是一对一窄依赖, 但是保留了中间的分隔信息, 可以通过某个分区获取其父分区, 目前只有一个算子生成这种窄依赖, 就是 union 算子, 例如 rddC = rddA.union(rddB)
rddC 其实就是 rddA 拼接 rddB 生成的, 所以 rddC 的 p5 和 p6 就是 rddB 的 p1 和 p2
所以需要有方式获取到 rddC 的 p5 其父分区是谁, 于是就需要记录一下边界, 其它部分和一对一窄依赖一样
多对一窄依赖其图形和 Shuffle 依赖非常相似, 所以在遇到的时候, 要注意其 RDD 之间是否有 Shuffle 过程, 比较容易让人困惑, 常见的多对一依赖就是重分区算子 coalesce, 例如 rddB = rddA.coalesce(2, shuffle = false), 但同时也要注意, 如果 shuffle = true 那就是完全不同的情况了
因为没有 Shuffle, 所以这是一个窄依赖
物理图的意义
如何划分 Task
如何划分 Stage
物理图解决的其实就是 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 的五个属性
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
第一个想法是每个 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 结构
ResultStage 开始执行最接近 Result 部分的 Stage id 为 0, 这个 Stage 被称之为 ResultStage
由代码可以知道, 最终调用 Action 促使整个流程执行的是最后一个 RDD, strRDD.collect, 所以当执行 RDD 的计算时候, 先计算的也是这个 RDD
RDD 之间是有关联的前面已经知道, 最后一个 RDD 先得到执行机会, 先从这个 RDD 开始执行, 但是这个 RDD 中有数据吗 ? 如果没有数据, 它的计算是什么? 它的计算是从父 RDD 中获取数据, 并执行传入的算子的函数
简单来说, 从产生 Result 的地方开始计算, 但是其 RDD 中是没数据的, 所以会找到父 RDD 来要数据, 父 RDD 也没有数据, 继续向上要, 所以, 计算从 Result 处调用, 但是从整个逻辑图中的最左边 RDD 开始, 类似一个递归的过程
生成逻辑图和物理图的系统组件
Job 和 Stage, Task 之间的关系
如何调度 Job
是什么 怎么生成 具体怎么生成
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}")上述代码在 Spark Application 的 main 方法中执行, 而 Spark Application 在 Driver 中执行, 所以上述代码在 Driver 中被执行, 那么这段代码执行的结果是什么呢?
一段 Scala 代码的执行结果就是最后一行的执行结果, 所以上述的代码, 从逻辑上执行结果就是最后一个 RDD, 最后一个 RDD 也可以认为就是逻辑执行图, 为什么呢?
例如 rdd2 = rdd1.map(…?) 中, 其实本质上 rdd2 是一个类型为 MapPartitionsRDD 的对象, 而创建这个对象的时候, 会通过构造函数传入当前 RDD 对象, 也就是父 RDD, 也就是调用 map 算子的 rdd1, rdd1 是 rdd2 的父 RDD
一个 RDD 依赖另外一个 RDD, 这个 RDD 又依赖另外的 RDD, 一个 RDD 可以通过 getDependency 获得其父 RDD, 这种环环相扣的关系, 最终从最后一个 RDD 就可以推演出前面所有的 RDD
逻辑图其实本质上描述的就是数据的计算过程, 数据从哪来, 经过什么样的计算, 得到什么样的结果, 再执行什么计算, 得到什么结果
可是数据的计算是描述好了, 这种计算该如何执行呢?
Job 是什么 ?Job 和 Stage 的关系Job 是一个最大的调度单位, 也就是说 DAGScheduler 会首先创建一个 Job 的相关信息, 后去调度 Job, 但是没办法直接调度 Job, 比如说现在要做一盘手撕包菜, 不可能直接去炒一整颗包菜, 要切好撕碎, 再去炒
Job 需要切分 ?因为 Job 的含义是对整个 RDD 血统求值, 但是 RDD 之间可能会有一些宽依赖
如果遇到宽依赖的话, 两个 RDD 之间需要进行数据拉取和复制
如果要进行拉取和复制的话, 那么一个 RDD 就必须等待它所依赖的 RDD 所有分区先计算完成, 然后再进行拉取
由上得知, 一个 Job 是无法计算完整个 RDD 血统的
创建一个 Stage, 从后向前回溯 RDD, 遇到 Shuffle 依赖就结束 Stage, 后创建新的 Stage 继续回溯. 这个过程上面已经详细的讲解过, 但是问题是切分以后如何执行呢, 从后向前还是从前向后, 是串行执行多个 Stage, 还是并行执行多个 Stage
在图中, Stage 0 的计算需要依赖 Stage 1 的数据, 因为 reduceRDD 中一个分区可能需要多个 tupleRDD 分区的数据, 所以 tupleRDD 必须先计算完, 所以, 应该在逻辑图中自左向右执行 Stage
还是同样的原因, Stage 0 如果想计算, Stage 1 必须先计算完, 因为 Stage 0 中每个分区都依赖 Stage 1 中的所有分区, 所以 Stage 1 不仅需要先执行, 而且 Stage 1 执行完之前 Stage 0 无法执行, 它们只能串行执行
一个 Stage 就是物理执行计划中的一个步骤, 一个 Spark Job 就是划分到不同 Stage 的计算过程
Stage 之间的边界由 Shuffle 操作来确定
Stage 内的 RDD 之间都是窄依赖, 可以放在一个管道中执行
而 Shuffle 后的 Stage 需要等待前面 Stage 的执行
Stage 有两种
ShuffMapStage, 其中存放窄依赖的 RDD
ResultStage, 每个 Job 只有一个, 负责计算结果, 一个 ResultStage 执行完成标志着整个 Job 执行完毕
Stage 和 Task 的关系前面我们说到 Job 无法直接执行, 需要先划分为多个 Stage, 去执行 Stage, 那么 Stage 可以直接执行吗?
第一点: Stage 中的 RDD 之间是窄依赖
因为 Stage 中的所有 RDD 之间都是窄依赖, 窄依赖 RDD 理论上是可以放在同一个 Pipeline(管道, 流水线) 中执行的, 似乎可以直接调度 Stage 了? 其实不行, 看第二点
第二点: 别忘了 RDD 还有分区
一个 RDD 只是一个概念, 而真正存放和处理数据时, 都是以分区作为单位的
Stage 对应的是多个整体上的 RDD, 而真正的运行是需要针对 RDD 的分区来进行的
第三点: 一个 Task 对应一个 RDD 的分区
一个比 Stage 粒度更细的单元叫做 Task, Stage 是由 Task 组成的, 之所以有 Task 这个概念, 是因为 Stage 针对整个 RDD, 而计算的时候, 要针对 RDD 的分区
假设一个 Stage 中有 10 个 RDD, 这些 RDD 中的分区各不相同, 但是分区最多的 RDD 有 30 个分区, 而且很显然, 它们之间是窄依赖关系
那么, 这个 Stage 中应该有多少 Task 呢? 应该有 30 个 Task, 因为一个 Task 计算一个 RDD 的分区. 这个 Stage 至多有 30 个分区需要计算
总结
一个 Stage 就是一组并行的 Task 集合
Task 是 Spark 中最小的独立执行单元, 其作用是处理一个 RDD 分区
一个 Task 只可能存在于一个 Stage 中, 并且只能计算一个 RDD 的分区
本章节重点是介绍 Shuffle 的流程, 因为根据 ShuffleWriter 的实现不同, 其过程也不同, 所以前半部分根据默认的存储引擎 SortShuffleWriter 来讲解
后半部分简要介绍一下其它的 ShuffleWriter
Shuffle 过程的组件结构从整体视角上来看, Shuffle 发生在两个 Stage 之间, 一个 Stage 把数据计算好, 整理好, 等待另外一个 Stage 来拉取
放大视角, 会发现, 其实 Shuffle 发生在 Task 之间, 一个 Task 把数据整理好, 等待 Reducer 端的 Task 来拉取
如果更细化一下, Task 之间如何进行数据拷贝的呢? 其实就是一方 Task 把文件生成好, 然后另一方 Task 来拉取
现在是一个 Reducer 的情况, 如果有多个 Reducer 呢? 如果有多个 Reducer 的话, 就可以在每个 Mapper 为所有的 Reducer 生成各一个文件, 这种叫做 Hash base shuffle, 这种 Shuffle 的方式问题大家也知道, 就是生成中间文件过多, 而且生成文件的话需要缓冲区, 占用内存过大
那么可以把这些文件合并起来, 生成一个文件返回, 这种 Shuffle 方式叫做 Sort base shuffle, 每个 Reducer 去文件的不同位置拿取数据
如果再细化一下, 把参与这件事的组件也放置进去, 就会是如下这样
ShuffleWriter ?SortShuffleWriter 的执行过程
整个 SortShuffleWriter 如上述所说, 大致有如下几步
首先 SortShuffleWriter 在 write 方法中回去写文件, 这个方法中创建了 ExternalSorter
write 中将数据 insertAll 到 ExternalSorter 中
在 ExternalSorter 中排序
如果要聚合, 放入 AppendOnlyMap 中, 如果不聚合, 放入 PartitionedPairBuffer 中
在数据结构中进行排序, 排序过程中如果内存数据大于阈值则溢写到磁盘
使用 ExternalSorter 的 writePartitionedFile 写入输入文件
将所有的溢写文件通过类似 MergeSort 的算法合并
将数据写入最终的目标文件中
理解闭包以及 Spark 分布式运行代码的根本原理
理解累加变量的使用场景
理解广播的使用场景
sc.textFile("dataset/access_log_sample.txt")
.flatMap(item => item.split(""))
.collect()上述这段代码中, flatMap 中传入的是另外一个函数, 传入的这个函数就是一个闭包, 这个闭包会被序列化运行在不同的 Executor 中
class MyClass {
val field = "Hello"
def doStuff(rdd: RDD[String]): RDD[String] = {
rdd.map(x => field + x)
}
}这段代码中的闭包就有了一个依赖, 依赖于外部的一个类, 因为传递给算子的函数最终要在 Executor 中运行, 所以需要 序列化MyClass 发给每一个 Executor, 从而在 Executor 访问 MyClass 对象的属性
闭包就是一个封闭的作用域, 也是一个对象
Spark 算子所接受的函数, 本质上是一个闭包, 因为其需要封闭作用域, 并且序列化自身和依赖, 分发到不同的节点中运行
var count = 0
val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)
sc.parallelize(Seq(1, 2, 3, 4, 5))
.foreach(count += _)
println(count)上面这段代码是一个非常错误的使用, 请不要仿照, 这段代码只是为了证明一些事情
先明确两件事, var count = 0 是在 Driver 中定义的, foreach(count += _) 这个算子以及传递进去的闭包运行在 Executor 中
这段代码整体想做的事情是累加一个变量, 但是这段代码的写法却做不到这件事, 原因也很简单, 因为具体的算子是闭包, 被分发给不同的节点运行, 所以这个闭包中累加的并不是 Driver 中的这个变量
Accumulators(累加器) 是一个只支持 added(添加) 的分布式变量, 可以在分布式环境下保持一致性, 并且能够做到高效的并发.
原生 Spark 支持数值型的累加器, 可以用于实现计数或者求和, 开发者也可以使用自定义累加器以实现更高级的需求
val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)
val counter = sc.longAccumulator("counter")
sc.parallelize(Seq(1, 2, 3, 4, 5))
.foreach(counter.add(_))
// 运行结果: 15
println(counter.value)注意点:
Accumulator 是支持并发并行的, 在任何地方都可以通过 add 来修改数值, 无论是 Driver 还是 Executor
只能在 Driver 中才能调用 value 来获取数值
在 WebUI 中关于 Job 部分也可以看到 Accumulator 的信息, 以及其运行的情况
累计器件还有两个小特性, 第一, 累加器能保证在 Spark 任务出现问题被重启的时候不会出现重复计算. 第二, 累加器只有在 Action 执行的时候才会被触发.
val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)
val counter = sc.longAccumulator("counter")
sc.parallelize(Seq(1, 2, 3, 4, 5))
.map(counter.add(_)) // 这个地方不是 Action, 而是一个 Transformation
// 运行结果是 0
println(counter.value)开发者可以通过自定义累加器来实现更多类型的累加器, 累加器的作用远远不只是累加, 比如可以实现一个累加器, 用于向里面添加一些运行信息
class InfoAccumulator extends AccumulatorV2[String, Set[String]] {
private val infos: mutable.Set[String] = mutable.Set()
override def isZero: Boolean = {
infos.isEmpty
}
override def copy(): AccumulatorV2[String, Set[String]] = {
val newAccumulator = new InfoAccumulator()
infos.synchronized {
newAccumulator.infos ++= infos
}
newAccumulator
}
override def reset(): Unit = {
infos.clear()
}
override def add(v: String): Unit = {
infos += v
}
override def merge(other: AccumulatorV2[String, Set[String]]): Unit = {
infos ++= other.value
}
override def value: Set[String] = {
infos.toSet
}
}
@Test
def accumulator2(): Unit = {
val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)
val infoAccumulator = new InfoAccumulator()
sc.register(infoAccumulator, "infos")
sc.parallelize(Seq("1", "2", "3"))
.foreach(item => infoAccumulator.add(item))
// 运行结果: Set(3, 1, 2)
println(infoAccumulator.value)
sc.stop()
}注意点:
可以通过继承 AccumulatorV2 来创建新的累加器
有几个方法需要重写
reset 方法用于把累加器重置为 0
add 方法用于把其它值添加到累加器中
merge 方法用于指定如何合并其他的累加器
value 需要返回一个不可变的集合, 因为不能因为外部的修改而影响自身的值
理解为什么需要广播变量, 以及其应用场景
能够通过代码使用广播变量
广播变量允许开发者将一个 Read-Only 的变量缓存到集群中每个节点中, 而不是传递给每一个 Task 一个副本.
集群中每个节点, 指的是一个机器
每一个 Task, 一个 Task 是一个 Stage 中的最小处理单元, 一个 Executor 中可以有多个 Stage, 每个 Stage 有多个 Task
所以在需要跨多个 Stage 的多个 Task 中使用相同数据的情况下, 广播特别的有用
广播变量用于将变量缓存在集群中的机器中, 避免机器内的 Executors 多次使用网络拉取数据
广播变量的使用步骤: (1) 创建 (2) 在 Task 中获取值 (3) 销毁
标签：常用 组件 extern bool 分数 字符串表 优化 做什么 hdfs
原文地址：https://www.cnblogs.com/mediocreWorld/p/11610059.html
