Flink SQL 性能优化：multiple input 详解

简介： 在 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_sk
    and time_dim.t_hour = 8
    and time_dim.t_minute >= 30
    and household_demographics.hd_dep_count = 5
    and 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 shuffle，Flink 在早期就已经引入了 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 的优化思路，引入新的优化机制并知足如下条件：

1. 该机制能够组合多输入的算子；
2. 该机制支持多路输入（为被组合的算子提供输入）

咱们就能够将用 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 暂未在流做业中推出推出该优化的缘由。后续版本咱们将逐步完成这些工做，也但愿更多社区的力量加入咱们一块儿尽早的将更多的优化落地。

做者：贺小令、翁才智
原文连接本文为阿里云原创内容，未经容许不得转载