关于后端:Hybrid-Shuffle-测试分析和使用建议

作者|郭伟杰

摘要：Apache Flink 社区在 1.16 版本引入了 Hybrid Shuffle Mode[1]，它是传统的 Batch Shuffle 和 Pipelined Shuffle 的联合，让 Flink 批处理具备了更弱小的能力。Hybrid Shuffle 的核心思想是突破调度束缚，依据可用资源的状况来决定是否须要调度上游工作，同时在条件容许时反对全内存不落盘的数据传输。

为了全面了解 Hybrid Shuffle 的后劲，咱们基于 Flink 1.17 版本在多个场景下对 Hybrid Shuffle 进行了测试。本文将基于测试后果详细分析 Hybrid Shuffle 的劣势场景，并基于咱们的教训给出一些应用倡议。

查看更多技术内容

Hybrid Shuffle 的劣势剖析

相比于传统的批式 Shuffle, Hybrid Shuffle 次要具备以下劣势：

• 调度：

  Hybrid Shuffle 突破了 Pipelined Shuffle 所有 Task 必须同时调度，Blocking Shuffle 必须分 Stage 调度的束缚：

  • 在资源短缺时，上下游 Task 能够同时运行
  • 在资源有余时，上下游 Task 能够分批先后执行

• IO 开销：

  Hybrid Shuffle 突破了批作业所有数据必须全副落盘并从磁盘生产数据的束缚，在上下游同时运行的状况下，它反对间接从内存生产数据，从而在晋升作业性能的同时大幅缩小磁盘 IO 带来的额定开销。

  Hybrid Shuffle 的上述两个劣势让它具备了传统批处理所没有的能力，咱们对其进行了一系列的试验和剖析，次要分为以下几个方面。

填补资源空隙

资源空隙 指在作业运行的某些工夫点，存在一些闲暇的 Slot，导致集群资源不能被充分利用。Flink Blocking Shuffle 因为上下游 Stage 之间的调度束缚，在上游 Task 没有齐全完结时，上游 Task 无奈被调度，从而产生了资源空隙。这种景象在局部 Task 存在数据歪斜的场景下尤为显著。

下图展现了一种 Blocking Shuffle 存在资源空隙的例子以及与之对应的 Hybrid Shuffle 的状况。能够看出 Blocking Shuffle 在这种状况下有 2 个 Slot 是无奈被利用的，而 Hybrid Shuffle 的全副 3 个 Slot 都是在应用中的。

值得一提的是：数据歪斜景象是宽泛存在的，以 TPC-DS q4 为例：其中一个 HashJoin 算子均匀读取的数据量为 204MB，而其中有一个歪斜的 Task 读取的数据量达到了 7.03 GB。测试发现，Hybrid Shuffle 相比 Blocking Shuffle 在该 Query 上的总执行工夫缩小了 18.74%。

缩小磁盘负载

Flink Blocking Shuffle 的两头数据会全量落盘，Shuffle Write 和 Shuffle Read 阶段别离进行磁盘的写和读操作。这会带来两个次要问题：

• 磁盘的 IO 负载变高，影响整个集群的吞吐。随着集群上的作业量增多，磁盘读写成为作业执行的瓶颈。
• 大规模 Batch 作业的 Shuffle 数据会占据相当一部分磁盘存储空间且大小难以预估，在以 Kubernetes 为代表的云原生环境下问题更为突出：如果配置过小，则会遇到存储空间有余的问题；如果配置过大，因为资源多是以 Pod 为粒度进行隔离，又造成了存储资源的节约。

Hybrid Shuffle 引入了全落盘和选择性落盘两种落盘策略：

• 选择性落盘策略下只有在内存空间有余时才会溢写一部分数据到磁盘。这种策略能够同时缩小磁盘读写指令。
• 全落盘策略下所有两头数据全量落盘，然而上游反对从内存间接生产未被开释的数据。这种策略能够在无效缩小磁盘读指令的同时兼顾更好的容错能力。

为了比照不同 Shuffle 模式和落盘策略对磁盘 IO 负载的影响，咱们进行了如下试验：

• 测试不同 Shuffle 和落盘策略下从磁盘读取和写入的数据量占总数据量的比例：

• 测试 Hybrid Shuffle 选择性落盘模式不同网络层内存大小下从磁盘读取和写入的数据量占总数据量的比例：

从试验后果能够看出：

• 相比 Blocking Shuffle，Hybrid Shuffle 极大地升高了从磁盘读写的数据量。
• 相比全落盘策略，选择性落盘能够大幅缩小磁盘写的数据量
• 随着网络层内存的减少，Hybrid Shuffle 从内存中读取数据的比例逐步减少。

咱们还能够察看到两个乏味的景象：

• 对于选择性落盘来说，其磁盘读的数据量少于磁盘写的数据量。这阐明在选择性落盘模式下，仍有很多落盘操作是非必要的。这是因为局部数据在落盘的过程中被间接从内存生产了，针对这种状况，将来还能够做进一步优化。
• 全落盘和选择性落盘从磁盘读取的数据量是不统一的。选择性落盘缩小了磁盘写操作，IO 负载的缩小使得磁盘读变快了。上游的生产进度更容易追上上游的生产进度，从而又促成了从内存读的比例。

Hybrid Shuffle 应用倡议

基于上述剖析和试验后果，咱们总结了以下三条 Hybrid Shuffle 的应用倡议：

适当缩小算子的并行度

算子的并行度是影响 Flink 作业执行性能的一个重要维度。对于应用 Blocking Shuffle 的批作业来说，个别会把算子的并行度调得比拟大，来取得更好的分布式执行能力。

而在 Hybrid Shuffle 模式下，因为其具备提前调度上游工作的能力。在总资源不变的状况下，适当缩小算子的并行度能够让更多的 Stage 同时运行，缩小落盘的 IO 数据量，从而取得更好的性能。

为了证实这个论断，咱们对 Hybrid Shuffle 和 Blocking Shuffle 在总资源 (Slot 数) 固定的状况下别离调整不同的算子并行度 (500, 750, 1000, 1500, 2000) 进行 TPC-DS 测试。按总执行工夫掂量，测试后果如下：

从试验后果能够看出：

Hybrid Shuffle 获得最优的并行度绝对较小，然而 Blocking 获得最优成果的并行度却和总 Slot 数保持一致。这是因为 Hybrid Shuffle 能够以缩小并行度为代价来换取上下游更好的并行。而 Blocking Shuffle 如果并行度设置得比拟小，会存在闲暇资源无奈被利用的状况。

同样须要留神的是，对于 Hybrid Shuffle 来说，尽管在较低并行度下整体执行工夫是最优的。但咱们也发现有些 Query 并行度比拟大的时候才会有更好的成果。这是因为这些 Query 中存在多数计算比拟重的算子，在并行度比拟小的时候，这些算子会成为整个作业的瓶颈。

以 TPC-DS q93.sql 为例, 其拓扑如下：

绿框中的 MultipleInput -> Calc 节点是整个作业的瓶颈，通过采样剖析咱们发现：其解决的数据量远大于其余算子，且单条数据处理得比较慢。即便上游工作全副曾经被调度起来，依然要期待该瓶颈节点解决实现。一旦该节点变成 Finished 状态，整个作业马上就会完结。

对于由 n 个 Stage 串联而成的拓扑，将第 i 个 Stage 在并行度较高 (上下游无奈同时运行) 和并行度较低 (上下游能够同时运行) 时的执行工夫别离记作 \(T_i^h \)和 \(T_i^l \)。则两种并行度下的总执行工夫别离为：

$$
\displaystyle\sum_{i=1}^n T_i^h \qquad
$$

$$
Max(T_1^l, \ T_2^l \ … \ T_n^l) \qquad
$$

注：为了更简略的阐明问题，这里只思考了多个 Stage 同时运行或先后运行，没有思考一个 Stage 局部完结另一个 Stage 局部开始的状况。

缩减并行度的实质是让 Stage 本身的执行工夫变慢, 也就是让 \(T_i^l \) 大于 \(T_i^h \)，然而让其能够同时运行。如果上游的 Stage 执行很慢而上游 Stage 执行很快，那么缩减并行度后上游 Stage 变慢减少的工夫会比拟多，而上游 Stage 其实不须要提前那么多工夫开始执行，就会造成损失大于收益。

回到上述的 Query 中：MultipleInput -> Calc 是整个作业的瓶颈, 该 Stage 在高 / 低并行度下的执行工夫别离记作 \(T_M^h \) 和 \(T_M^l \)。则 (1) 式的后果次要取决于 \(T_M^h \)，而 (2) 式的后果等于 \(T_M^l \)，而缩减并行度带来的性能损失 (\( T_M^l – T_M^h \)) 绝对较大，总体体现为作业执行工夫变长了。当咱们把该作业的默认并行度从 500 减少到 1500 时，作业性能失去显著晋升，总执行工夫缩小了 47%。因而，在 Hybrid Shuffle 模式下算子的并行度也并非设置的越小越好。

适当减少网络层内存

网络层内存的大小对 Flink Shuffle 阶段的性能会产生较大的影响。如果这部分内存不足，网络层 Buffer 竞争会变得强烈，从而导致作业的反压。

防止因网络层内存不够而报错

Hybrid Shuffle 须要更多的内存能力保障 Shuffle 阶段的失常运行。次要起因是：相比 Blocking Shuffle，Hybrid Shuffle 目前的实现中网络内存需要和 Task 的并行度不解耦。社区在 Hybrid Shuffle 方向上接下来工作的重点之一就是对此进行改良。

两种 Shuffle 模式 Shuffle Write 和 Shuffle Read 阶段对网络层内存的最小需要如下表所示：

注：上表所列是简化版本，理论规定更为简单

从表中能够看出：

• Blocking Shuffle 的网络层内存需要和并行度没有关联关系，减少作业并行度不必太过关怀网络层内存的大小。
• Hybrid Shuffle 的网络层内存需要基本上是和并行度线性相关的。随着并行度的减少，可能导致总网络层内存无奈满足作业运行的最低要求，从而产生 Insufficient Netwrok Memory 的报错。减少作业并行度时，须要对网络层内存也做相应的调整。

晋升从内存读取的比例

对于 Blocking Shuffle 来说，数据只能从磁盘进行生产，积攒到肯定水平之后间接落盘就能够开释所占据的内存，因而网络层内存只有能保障不产生强烈的 Buffer 竞争即可。即使配置得十分短缺，对性能也不会产生很大的影响。而在 Hybrid Shuffle 的模式下，减少网络层内存能够晋升从内存读取的比例。这是因为 Hybrid Shuffle 对内存中数据的驱赶策略是思考内存池的使用率的，内存越短缺，数据在内存中的存活工夫就越久，也就越有可能被上游间接生产，进而缩小磁盘 IO 开销。

为了探索网络层内存大小对不同 Shuffle 实现的影响，别离在 TPC-DS 10T 数据集上进行了测试。以 TaskManager 总内存 24G，网络层内存 2.5G 为基准, 同时增大 TaskManager 总内存和网络层内存(每减少 1G 网络层内存，TaskManager 总内存也随之增大 1G)。性能绝对基准配置的晋升比例如下图所示：

从试验后果能够看出，随着网络层内存的晋升，两者的性能都有晋升。Blocking Shuffle 晋升的比例不是很显著，而 Hybrid Shuffle 则对网络层内存大小比拟敏感。

尽量避免同时应用 Hybrid Shuffle 和动静并行度

Flink 反对在运行时对批作业动静设置并行度，其原理是: 按 Stage 对作业进行调度，依据上游曾经完结 Stage 的统计信息 (次要是产出的数据量) 推断上游 Stage 的并行度而后进行调度。

动静并行度模式对调度有人造束缚：上游 Stage 必须等上游 Stage Finished 之后才能够调度。Hybrid Shuffle 能够反对这种模式，然而这也就意味着 Hybrid Shuffle 在调度上的劣势无奈施展进去。

为了验证两种 Shuffle 模式在动静并行度和非动静并行度下的体现，别离对 Blocking Shuffle 和 Hybrid Shuffle 在 TPC-DS 数据集上进行测试，配置默认并行度(parallelism.default) 为 1500，试验后果如图所示。

从图中能够看出，Hybrid Shuffle 在动静并行度模式下相比 Blocking Shuffle 总执行工夫差异不大，性能基本相同。同时，其非动静并行度模式相比动静并行度有肯定的性能劣势，这次要是因为非动静并行度模式下，Hybrid Shuffle 能够在局部上游工作完结之后提前调度上游工作。而 Blocking Shuffle 的动静并行度却比非动静并行度模式性能要好，这是因为动静并行度升高了数据量比拟小的 Task 在调度，部署等方面的额定开销。

总结

本文次要剖析了 Hybrid Shuffle 产生性能劣势的起因，基于此进行了试验测试和剖析，并给出了相应的应用倡议：

1. 适当缩小算子的并行度，个别调整到能让 2～3 个 Stage 并行即可。
2. 适当减少网络层内存。
3. 尽量避免同时应用 Hybrid Shuffle 和动静并行度。

心愿本文能够帮忙读者理解到在什么样的场景下应该抉择 Hybrid Shuffle 以及如何对其进行调优。

[1] FLIP-235: Hybrid Shuffle Mode

查看更多技术内容

更多内容

流动举荐

阿里云基于 Apache Flink 构建的企业级产品 - 实时计算 Flink 版现开启流动：
0 元试用 实时计算 Flink 版（5000CU* 小时，3 个月内）
理解流动详情：https://click.aliyun.com/m/1000372333/