spark系列12：RDD,stage,job,shuffle物理执行过程

淩亂°似流年 2022-10-22 14:54 134阅读 0赞

### 1. 物理执行图生成 ###

1.  物理图的意义
2.  如何划分 Task
3.  如何划分 Stage

物理图的作用是什么?

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

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

![Snipaste 2019 05 23 14 00 33][]

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

*  首先集群中物理元件就是一台一台的机器
 *  其次这些机器上跑的守护进程有两种: `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` 保持交互从而认领属于自己的任务
    
    ![20190521111630][]
 *  接受任务后, 运行任务
    
    ![20190521111456][]

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

问题三: Task 该如何设计 ?

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

第一个想法: 为每个 RDD 的分区设置一组 Task

![20190521113535][]

大概就是每个 `RDD` 都有三个 `Task`, 每个 `Task` 对应一个 `RDD` 的分区, 执行一个分区的数据的计算

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

这只是一个简单的逻辑图, 如果是一个复杂的逻辑图, 会有什么表现? 要存储多少数据? 无论是放在磁盘还是放在内存中, 是不是都是一种极大的负担?

第二个想法: 让数据流动

很自然的, 第一个想法的问题是数据需要存储和交换, 那不存储不就好了吗? 对, 可以让数据流动起来

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

![20190521114511][]

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

![20190521114717][]

这两个 `RDD` 之间是 `Shuffle` 关系, 也就是说, 右边的 `RDD` 的一个分区可能依赖左边 `RDD` 的所有分区, 这样的话, 数据在这个地方流不动了, 怎么办?

第三个想法: 划分阶段

既然在 `Shuffle` 处数据流不动了, 那就可以在这个地方中断一下, 后面 `Stage` 部分详解

如何划分阶段 ?

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

![20190521114717][]

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

![20190521115759][]

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

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

![20190521120501][]

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

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

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

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

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>这样一个 RDD 依赖的链条, 我们称之为 RDD 的血统, 其中有宽依赖也有窄依赖</p> </td> 
  </tr> 
 </tbody> 
</table>

数据怎么流动 ?

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))

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

![20190521161456][]

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

![20190521161337][]

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` 开始, 类似一个递归的过程

![20190521162302][]

### 2. 调度过程 ###

导读

1.  生成逻辑图和物理图的系统组件
2.  `Job` 和 `Stage`, `Task` 之间的关系
3.  如何调度 `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`

![20190521165818][]

一个 `RDD` 依赖另外一个 `RDD`, 这个 `RDD` 又依赖另外的 `RDD`, 一个 `RDD` 可以通过 `getDependency` 获得其父 `RDD`, 这种环环相扣的关系, 最终从最后一个 `RDD` 就可以推演出前面所有的 `RDD`

逻辑图是什么, 干啥用

逻辑图其实本质上描述的就是数据的计算过程, 数据从哪来, 经过什么样的计算, 得到什么样的结果, 再执行什么计算, 得到什么结果

可是数据的计算是描述好了, 这种计算该如何执行呢?

物理图

数据的计算表示好了, 该正式执行了, 但是如何执行? 如何执行更快更好更酷? 就需要为其执行做一个规划, 所以需要生成物理执行图

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

上述代码其实就是最后的一个 `RDD` 调用了 `Action` 方法, 调用 `Action` 方法的时候, 会请求一个叫做 `DAGScheduler` 的组件, `DAGScheduler` 会创建用于执行 `RDD` 的 `Stage` 和 `Task`

`DAGScheduler` 是一个由 `SparkContext` 创建, 运行在 `Driver` 上的组件, 其作用就是将由 `RDD` 构建出来的逻辑计划, 构建成为由真正在集群中运行的 `Task` 组成的物理执行计划, `DAGScheduler` 主要做如下三件事

1.  帮助每个 `Job` 计算 `DAG` 并发给 `TaskSheduler` 调度
2.  确定每个 `Task` 的最佳位置
3.  跟踪 `RDD` 的缓存状态, 避免重新计算

从字面意思上来看, `DAGScheduler` 是调度 `DAG` 去运行的, `DAG` 被称作为有向无环图, 其实可以将 `DAG` 理解为就是 `RDD` 的逻辑图, 其呈现两个特点: `RDD` 的计算是有方向的, `RDD` 的计算是无环的, 所以 `DAGScheduler` 也可以称之为 `RDD Scheduler`, 但是真正运行在集群中的并不是 `RDD`, 而是 `Task` 和 `Stage`, `DAGScheduler` 负责这种转换

`Job` 是什么 ?

`Job` 什么时候生成 ?

当一个 `RDD` 调用了 `Action` 算子的时候, 在 `Action` 算子内部, 会使用 `sc.runJob()` 调用 `SparkContext` 中的 `runJob` 方法, 这个方法又会调用 `DAGScheduler` 中的 `runJob`, 后在 `DAGScheduler` 中使用消息驱动的形式创建 `Job`

简而言之, `Job` 在 `RDD` 调用 `Action` 算子的时候生成, 而且调用一次 `Action` 算子, 就会生成一个 `Job`, 如果一个 `SparkApplication` 中调用了多次 `Action` 算子, 会生成多个 `Job` 串行执行, 每个 `Job` 独立运作, 被独立调度, 所以 `RDD` 的计算也会被执行多次

`Job` 是什么 ?

如果要将 `Spark` 的程序调度到集群中运行, `Job` 是粒度最大的单位, 调度以 `Job` 为最大单位, 将 `Job` 拆分为 `Stage` 和 `Task` 去调度分发和运行, 一个 `Job` 就是一个 `Spark` 程序从 `读取 → 计算 → 运行` 的过程

一个 `Spark Application` 可以包含多个 `Job`, 这些 `Job` 之间是串行的, 也就是第二个 `Job` 需要等待第一个 `Job` 的执行结束后才会开始执行

`Job` 和 `Stage` 的关系

`Job` 是一个最大的调度单位, 也就是说 `DAGScheduler` 会首先创建一个 `Job` 的相关信息, 后去调度 `Job`, 但是没办法直接调度 `Job`, 比如说现在要做一盘手撕包菜, 不可能直接去炒一整颗包菜, 要切好撕碎, 再去炒

为什么 `Job` 需要切分 ?

![20190521161456][]

*  因为 `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` 的关系

![20190521120501][]

前面我们说到 `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 的分区

TaskSet

梳理一下这几个概念, `Job > Stage > Task`, `Job 中包含 Stage 中包含 Task`

而 `Stage` 中经常会有一组 `Task` 需要同时执行, 所以针对于每一个 `Task` 来进行调度太过繁琐, 而且没有意义, 所以每个 `Stage` 中的 `Task` 们会被收集起来, 放入一个 `TaskSet` 集合中

*  一个 `Stage` 有一个 `TaskSet`
 *  `TaskSet` 中 `Task` 的个数由 `Stage` 中的最大分区数决定

整体执行流程

![20190522015026][]

### 3. Shuffle 过程 ###

导读

本章节重点是介绍 `Shuffle` 的流程, 因为根据 `ShuffleWriter` 的实现不同, 其过程也不同, 所以前半部分根据默认的存储引擎 `SortShuffleWriter` 来讲解

后半部分简要介绍一下其它的 `ShuffleWriter`

`Shuffle` 过程的组件结构

从整体视角上来看, `Shuffle` 发生在两个 `Stage` 之间, 一个 `Stage` 把数据计算好, 整理好, 等待另外一个 `Stage` 来拉取

![20190522132537][]

放大视角, 会发现, 其实 `Shuffle` 发生在 `Task` 之间, 一个 `Task` 把数据整理好, 等待 `Reducer` 端的 `Task` 来拉取

![20190522132852][]

如果更细化一下, `Task` 之间如何进行数据拷贝的呢? 其实就是一方 `Task` 把文件生成好, 然后另一方 `Task` 来拉取

![20190522133401][]

现在是一个 `Reducer` 的情况, 如果有多个 `Reducer` 呢? 如果有多个 `Reducer` 的话, 就可以在每个 `Mapper` 为所有的 `Reducer` 生成各一个文件, 这种叫做 `Hash base shuffle`, 这种 `Shuffle` 的方式问题大家也知道, 就是生成中间文件过多, 而且生成文件的话需要缓冲区, 占用内存过大

![20190522140738][]

那么可以把这些文件合并起来, 生成一个文件返回, 这种 `Shuffle` 方式叫做 `Sort base shuffle`, 每个 `Reducer` 去文件的不同位置拿取数据

![20190522141807][]

如果再细化一下, 把参与这件事的组件也放置进去, 就会是如下这样

![20190522170646][]

有哪些 `ShuffleWriter` ?

大致上有三个 `ShufflWriter`, `Spark` 会按照一定的规则去使用这三种不同的 `Writer`

*  `BypassMergeSortShuffleWriter`
    
    这种 `Shuffle Writer` 也依然有 `Hash base shuffle` 的问题, 它会在每一个 `Mapper` 端对所有的 `Reducer` 生成一个文件, 然后再合并这个文件生成一个统一的输出文件, 这个过程中依然是有很多文件产生的, 所以只适合在小量数据的场景下使用
    
    `Spark` 有考虑去掉这种 `Writer`, 但是因为结构中有一些依赖, 所以一直没去掉
    
    当 `Reducer` 个数小于 `spark.shuffle.sort.bypassMergeThreshold`, 并且没有 `Mapper` 端聚合的时候启用这种方式
 *  `SortShuffleWriter`
    
    这种 `ShuffleWriter` 写文件的方式非常像 `MapReduce` 了, 后面详说
    
    当其它两种 `Shuffle` 不符合开启条件时, 这种 `Shuffle` 方式是默认的
 *  `UnsafeShuffleWriter`
    
    这种 `ShuffWriter` 会将数据序列化, 然后放入缓冲区进行排序, 排序结束后 `Spill` 到磁盘, 最终合并 `Spill` 文件为一个大文件, 同时在进行内存存储的时候使用了 `Java` 得 `Unsafe API`, 也就是使用堆外内存, 是钨丝计划的一部分
    
    也不是很常用, 只有在满足如下三个条件时候才会启用
    
     *  序列化器序列化后的数据, 必须支持排序
     *  没有 `Mapper` 端的聚合
     *  `Reducer` 的个数不能超过支持的上限 (2 ^ 24)

`SortShuffleWriter` 的执行过程

![20190522160031][]

整个 `SortShuffleWriter` 如上述所说, 大致有如下几步

1.  首先 `SortShuffleWriter` 在 `write` 方法中回去写文件, 这个方法中创建了 `ExternalSorter`
2.  `write` 中将数据 `insertAll` 到 `ExternalSorter` 中
3.  在 `ExternalSorter` 中排序
    
    1.  如果要聚合, 放入 `AppendOnlyMap` 中, 如果不聚合, 放入 `PartitionedPairBuffer` 中
    2.  在数据结构中进行排序, 排序过程中如果内存数据大于阈值则溢写到磁盘
4.  使用 `ExternalSorter` 的 `writePartitionedFile` 写入输入文件
    
    1.  将所有的溢写文件通过类似 `MergeSort` 的算法合并
    2.  将数据写入最终的目标文件中

[Snipaste 2019 05 23 14 00 33]: /images/20221022/2fb1ea55ea5a44429d4f0f8d0c000a7e.png
[20190521111630]: /images/20221022/cfb8dead5f844532bf88c02bbf29be76.png
[20190521111456]: /images/20221022/f5260c5b52fd4921a813c7c94e1b9c45.png
[20190521113535]: /images/20221022/4ed21430b34f4d518b37274082dd4bbc.png
[20190521114511]: /images/20221022/5552be0dd7f64e818837f6b7a295101e.png
[20190521114717]: /images/20221022/2430b928e63f4c46824d24cc7a58a3e6.png
[20190521115759]: /images/20221022/4561e2504cd242cea53483d4a616801c.png
[20190521120501]: /images/20221022/be46cc820cef474aad8de8c203abdd01.png
[20190521161456]: /images/20221022/0640239b466341a286ad6d586467abc1.png
[20190521161337]: /images/20221022/948dd79d73b2475792d12554c6234cbf.png
[20190521162302]: /images/20221022/0047ea38d4ce4a42b719f307ce7abdf5.png
[20190521165818]: /images/20221022/9b3339c3acaa4f33b60957cecb848f2e.png
[20190522015026]: /images/20221022/66b57c3d30f545b292a651c2ce411e97.png
[20190522132537]: /images/20221022/6f097946fab747ac8cf4d892f9647152.png
[20190522132852]: /images/20221022/f908f0ce9a3641d8a99788655be10714.png
[20190522133401]: /images/20221022/b9e963837b69489898f431eb497db5ba.png
[20190522140738]: /images/20221022/f58bd870c30242d389a7f9044f1ede6e.png
[20190522141807]: /images/20221022/ada484d73cf046dabf6cebc054b1a8ce.png
[20190522170646]: /images/20221022/21ef90596fb24473b3afc273141f7426.png
[20190522160031]: /images/20221022/b81d880d3ede4c4594f3617afb26ebe5.png