做相册网站logo,佛山家居网站全网营销,网络职业有哪些,网站首页布局的设计简介#xff1a; 在 Flink 1.12 中#xff0c;针对目前 operator chaining 无法覆盖的场景#xff0c;推出了 multiple input operator 与 source chaining 优化。该优化将消除 Flink 作业中大多数冗余 shuffle#xff0c;进一步提高作业的执行效率。本文将以一个 SQL 作业… 简介 在 Flink 1.12 中针对目前 operator chaining 无法覆盖的场景推出了 multiple input operator 与 source chaining 优化。该优化将消除 Flink 作业中大多数冗余 shuffle进一步提高作业的执行效率。本文将以一个 SQL 作业为例介绍上述优化并展示 Flink 1.12 在 TPC-DS 测试集上取得的成果。 执行效率的优化一直是 Flink 追寻的目标。在大多数作业特别是批作业中数据通过网络在 task 之间传递称为数据 shuffle的代价较大。正常情况下一条数据经过网络需要经过序列化、磁盘读写、socket 读写与反序列化等艰难险阻才能从上游 task 传输到下游而相同数据在内存中的传输仅需要耗费几个 CPU 周期传输一个八字节指针即可。
Flink 在早期版本中已经通过 operator chaining 机制将并发相同的相邻单输入算子整合进同一个 task 中消除了单输入算子之间不必要的网络传输。然而join 等多输入算子之间同样存在额外的数据 shuffle 问题shuffle 数据量最大的 source 节点与多输入算子之间的数据传输也无法利用 operator chaining 机制进行优化。
在 Flink 1.12 中我们针对目前 operator chaining 无法覆盖的场景推出了 multiple input operator 与 source chaining 优化。该优化将消除 Flink 作业中大多数冗余 shuffle进一步提高作业的执行效率。本文将以一个 SQL 作业为例介绍上述优化并展示 Flink 1.12 在 TPC-DS 测试集上取得的成果。
优化案例解析订单量统计
我们将以 TPC-DS q96 为例子详细介绍如何消除冗余 shuffle该 SQL 意在通过多路 join 筛选并统计符合特定条件的订单量。
select count(*)
from store_sales,household_demographics ,time_dim, store
where ss_sold_time_sk time_dim.t_time_sk and ss_hdemo_sk household_demographics.hd_demo_sk and ss_store_sk s_store_skand time_dim.t_hour 8and time_dim.t_minute 30and household_demographics.hd_dep_count 5and store.s_store_name ese 图 1 - 初始执行计划
冗余 Shuffle 是如何产生的
由于部分算子对输入数据的分布有要求如 hash join 算子要求同一并发内数据 join key 的 hash 值相同数据在算子之间传递时可能需要经过重新排布与整理。与 map-reduce 的 shuffle 过程类似Flink shuffle 将上游 task 产生的中间结果进行整理并按需发送给需要这些中间结果的下游 task。但在一部分情况下上游产出的数据已经满足了数据分布要求如连续多个 join key 相同的 hash join 算子此时对数据的整理便不再必要由此产生的 shuffle 也就成为了冗余 shuffle在执行计划中以 forward shuffle 表示。
图 1 中的 hash join 算子是一种称为 broadcast hash join 的特殊算子。以 store_sales join time_dim 为例由于 time_dim 表数据量很小此时通过 broadcast shuffle 将该表的全量数据发送给 hash join 的每个并发就能让任何并发接受 store_sales 表的任意数据而不影响 join 结果的正确性同时提高 hash join 的执行效率。此时 store_sales 表向 join 算子的网络传输也成为了冗余 shuffle。同理几个 join 之间的 shuffle 也是不必要的。 图 2 - 冗余的shuffle红框标记
除 hash join 与 broadcast hash join 外产生冗余 shuffle 的场景还有很多例如 group key 与 join key 相同的 hash aggregate hash join、group key 具有包含关系的多个 hash aggregate 等等这里不再展开描述。
Operator Chaining 能解决吗
对 Flink 优化过程有一定了解的读者可能会知道为了消除不必要的 forward shuffleFlink 在早期就已经引入了 operator chaining 机制。该机制将并发相同的相邻单输入算子整合进同一个 task 中并在同一个线程中一起运算。Operator chaining 机制在图 1 中其实已经在发挥作用如果没有它做 broadcast shuffle 的三个 Source 节点名称中被“-”分隔的算子将会被拆分至多个不同的 task产生冗余的数据 shuffle。图 3 为 Operator chaining 关闭是的执行计划。 图 3 - Operator chaining关闭后的执行计划
减少数据在 TM 之间通过网络和文件传输并将算子链接合并入 task 是非常有效的优化它能减少线程之间的切换减少消息的序列化与反序列化减少数据在缓冲区的交换并减少延迟的同时提高整体吞吐量。然而operator chaining 对算子的整合有非常严格的限制其中一条就是“下游算子的入度为 1”也就是说下游算子只能有一路输入。这就将多路输入的算子如 join排除在外。
多输入算子的解决方案Multiple Input Operator
如果我们能仿照 operator chaining 的优化思路引入新的优化机制并满足以下条件
该机制可以组合多输入的算子该机制支持多路输入为被组合的算子提供输入
我们就可以将用 forward shuffle 连接的的多输入算子放到一个 task 里执行从而消除不必要的 shuffle。Flink 社区很早就关注到了 operator chaining 的不足在 Flink 1.11 中引入了 streaming api 层的 MultipleInputTransformation 以及对应的 MultipleInputStreamTask。这些 api 满足了上述条件 2而 Flink 1.12 在此基础上在 SQL 层中实现了满足条件 1 的新算子——multiple input operator可以参考 FLIP 文档[1]。
Multiple input operator 是 table 层一个可插拔的优化。它位于 table 层优化的最后一步遍历生成的执行计划并将不被 exchange 阻隔的相邻算子整合进一个 multiple input operator 中。图 4 展示了该优化对原本 SQL 优化步骤的修改。 图 4 - 加入 multiple input operator 后的优化步骤
读者可能会有疑问为什么不在现有的 operator chaining 上进行修改而要另起炉灶呢实际上multiple input operator 除了要完成 operator chaining 的工作之外还需要对各个输入的优先级进行排序。这是因为一部分多输入算子如 hash join 与 nested loop join对输入有严格的顺序限制若输入优先级排序不当很可能造成死锁。由于算子输入优先级的信息仅在 table 层的算子中有描述更加自然的方式是在 table 层引入该优化机制。
值得注意的是multiple input operator 不同于管理多个 operator 的 operator chaining其本身就是一整个大 operator而其内部运算在外界看来就是一个黑盒。Multiple input operator 的内部结构在 operator name 中完全体现读者在运行包含该 operator 的作业时可以从 operator name 看到哪些算子以怎样的拓扑结构被组合进了 multiple input operator 中。
图 5 展示了经过 multiple input 优化后的算子的拓扑图以及 multiple input operator 的透视图。图中三个 hash join 算子之间的冗余的 shuffle 被移除后它们可以在一个 task 里执行只不过 operator chaining 没法处理这种多输入的情况将它们放到 multiple input operator 里执行由 multiple input operator 管理各个算子的输入顺序和算子之间的调用关系。 图 5 - 经过 multiple input 优化后的算子拓扑图
Multiple input operator 的构建和运行过程较为复杂对此细节有兴趣的读者可以参考设计文档[2]。
Source 也不能遗漏Source Chaining
经过 multiple input operator 的优化我们将图 1 中的执行计划优化为图 6图 3 经过 operator chaining 优化后就变为图 6 的执行图。 图 6 - 经过 multiple input operator 优化后的执行计划
图 6 中从 store_sales 表产生的 forward shuffle如红框所示表示我们仍有优化空间。正如序言中所说在大部分作业中从 source 直接产生的数据由于没有经过 join 等算子的筛选和加工shuffle 的数据量是最大的。以 10T 数据下的 TPC-DS q96 为例如果不进行进一步优化包含 store_sales 源表的 task 将向网络中传输 1.03T 的数据而经过一次 join 的筛选后数据量急速下降至 16.5G。如果我们能将源表的 forward shuffle 省去作业整体执行效率又能前进一大步。
可惜的是multiple input operator 也不能覆盖 source shuffle 的场景这是因为 source 不同于其它任何算子它没有任何输入。Flink 1.12 为此给 operator chaining 新增了 source chaining 功能将不被 shuffle 阻隔的 source 合并到 operator chaining 中省去了 source 与下游算子之间的 forward shuffle。
目前仅有 FLIP-27 source 以及 multiple input operator 可以利用 source chaining 功能不过这已经足够解决本文中的优化场景。
结合 multiple input operator 与 source chaining 之后图 7 展示了本文优化案例的最终执行方案。 图 7 - 优化后的执行方案
TPC-DS 测试结果
Multiple input operator 与 source chaining 对大部分作业特别是批作业有显著的优化效果。我们利用 TPC-DS 测试集对 Flink 1.12 的整体性能进行了测试与 Flink 1.10 公布的 12267s 总用时相比Flink 1.12 的总用时仅为 8708s缩短了近 30% 的运行时间 图 8 - TPC-DS 测试集总用时对比 图 9 - TPC-DS 部分测试点用时对比
未来计划
通过 TPC-DS 的测试效果看到source chaining multiple input 能够给我们带来很大的性能提升。目前整体框架已完成常用批算子已支持消除冗余 exchange 的推导逻辑后续我们将支持更多的批算子和更精细的推导算法。
流作业的数据 shuffle 虽然不需要像批作业一样将数据写入磁盘但将网络传输变为内存传输带来的性能提升也是非常可观的因此流作业支持 source chaining multiple input 也是一个非常令人期待的优化。同时在流作业上支持该优化还需要很多工作例如流算子上消除冗余 exchange 的推导逻辑暂未支持一些算子需要重构以消除输入数据是 binary 的要求等等这也是为什么 Flink 1.12 暂未在流作业中推出推出该优化的原因。后续版本我们将逐步完成这些工作也希望更多社区的力量加入我们一起尽早的将更多的优化落地。
作者贺小令、翁才智
原文链接
本文为阿里云原创内容未经允许不得转载
