京东|京东：Flink SQL 优化实战( 三 )

Flink 在 join 之上又做了一些细分，具体包括：

特点
使用
Repartition-Repartition strategy
对数据集分别进行分区和shuffle ，如果数据集大的时候效率极差
两个数据集相差不大
Broadcast-Forward strategy
将小表的数据全部发送到大表数据的机器上
两个数据集有较大的差距
Repartition-Repartition strategy：Join 的两个数据集分别对它们的 key 使用相同的分区函数进行分区，并经过网络发送数据； Broadcast-Forward strategy：大的数据集不做处理，另一个比较小的数据集全部复制到集群中一部分数据的机器上。众所周知， batch 的 shuffle 非常耗时间。
如果两个数据集有较大差距，建议采用 Broadcast-Forward strategy；如果两个数据集差不多，建议采用 Repartition-Repartition strategy 。可以通过：table.optimizer.join.broadcast-threshold 来设置采用 broadcast 的 table 大小，如果设置为 “-1” ，表示禁用 broadcast 。
下图为禁用前后的效果：

【京东|京东：Flink SQL 优化实战】
4. multiple input
在 Flink SQL 任务里，降低 shuffle 可以有效的提高 SQL 任务的吞吐量，在实际的业务场景中经常遇到这样的情况：上游产出的数据已经满足了数据分布要求 (如连续多个 join 算子，其中 key 是相同的) ，此时 Flink 的 forward shuffle 是冗余的 shuffle ，我们希望将这些算子 chain 到一起。 Flink 1.12 引入了 mutiple input 的特性，可以消除大部分没必要的 forward shuffle ，把 source 的算子 chain 到一起。
table.optimizer.multiple-input-enabled：true
下图为开了 multiple input 和没有开的拓扑图 ( operator chain 功能已经打开)：?

5. 对象重用
上下游 operator 之间会经过序列化 / 反序列化 / 复制阶段来进行数据传输，这种行为非常影响 Flink SQL 程序的性能，可以通过启用对象重用来提高性能。但是这在 DataStream 里面非常危险，因为可能会发生以下情况：在下一个算子中修改对象意外影响了上面算子的对象。
但是 Flink 的 Table / SQL API 中是非常安全的，可以通过如下方式来启用：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment（）;env.getConfig().enableObjectReuse(); 或者是通过设置：pipeline-object-reuse:true
为什么启用了对象重用会有这么大的性能提升？在 Blink planner 中，同一任务的两个算子之间的数据交换最终将调用 BinaryString#copy ，查看实现代码，可以发现 BinaryString#copy 需要复制底层 MemorySegment 的字节，通过启用对象重用来避免复制，可以有效提升效率。
下图为没有开启对象重用时相应的火焰图：

6. SQL 任务的 failover 策略
batch 任务模式下 checkpoint 以及其相关的特性全部都不可用，因此针对实时任务的基于 checkpoint 的 failover 策略是不能应用在批任务上面的，但是 batch 任务允许 Task 之间通过 Blocking Shuffle 进行通信，当一个 Task 因为任务未知的原因失败之后，由于 Blocking Shuffle 中存储了这个 Task 所需要的全部数据，所以只需要重启这个 Task 以及通过 Pipeline Shuffle 与其相连的全部下游任务即可：
jobmanager.execution.failover-strategy:region (已经 finish 的 operator 可直接恢复)
table.exec.shuffle-mode:ALL_EDGES_BLOCKING (shuffle 策略) 。
7. shuffle
Flink 里的 shuffle 分为 pipeline shuffle 和 blocking shuffle 。
pipeline shuffle 性能好，但是对资源的要求高，而且容错比较差 (会将该 operator 分到前面的一个 region 里面，对于 batch 任务来说，如果这个算子出问题，将从上一个 region 恢复)； blocking shuffle 就是传统的 batch shuffle ，会将数据落盘，这种 shuffle 的容错好，但是会产生大量的磁盘、网络 io (如果为了省心的话，建议用 blocking suffle) 。 blocking shuffle 又分为 hash shuffle 和 sort shuffle ，如果你的磁盘是 ssd 并且并发不太大的话，可以选择使用 hash shuffle ，这种 shuffle 方式产生的文件多、随机读多，对磁盘 io 影响较大；如果你是 sata 并且并发比较大，可以选择用 sort-merge shuffle ，这种 shuffle 产生的数据少，顺序读，不会产生大量的磁盘 io ，不过开销会更大一些 (sort merge) 。相应的控制参数：