浅谈Flink批模式Adaptive Hash Join

Flink批Hash Join递归超限问题

随着Flink流批一体能力的迅速发展以及Flink SQL易用性的提升，越来越多的厂商开始将Flink作为离线批处理引擎使用。在我们使用Flink进行大规模join操作时，也许会发生如下的异常，导致任务失败：

Hash join exceeded maximum number of recursions, without reducing partitions enough to be memory resident. 

字面意思即为Hash Join的递归次数超出限制。Flink批模式下的join算法有两种，即Hybrid Hash Join和Sort-Merge Join。顾名思义，Hybrid Hash Join就是Simple Hash Join和Grace Hash Join两种算法的结合（关于它们，看官可参考这篇文章）。引用一张Flink官方博客中的手绘图来说明。

Flink的Hybrid Hash Join在build阶段会积极地利用TaskManager的托管内存，并将内存无法容纳的哈希分区spill到磁盘中。在probe阶段，当内存中的哈希分区处理完成后，会释放掉对应的MemorySegment，并将先前溢写到磁盘的分区读入，以提升probe效率。特别注意，如果溢写分区对空闲的托管内存而言仍然过大（特别是存在数据倾斜的情况时），就会将其递归拆分成更小的分区，原理如下图所示。

当然，递归拆分也不能是无限制的。在Blink Runtime中，如果递归拆分3次仍然不能满足内存需求，就会抛出前文所述的异常了。

笔者在今年7月ApacheCon Asia 2022流处理专场的分享内容里谈到了这个问题，并且将其归咎于Flink SQL的CBO优化器的代价模型不太科学，导致其十分偏向选择Hash Join。由于修改的难度很大，所以暂时的workaround就是在任务失败后，自动设置table.exec.disabled-operators参数来禁用掉ShuffleHashJoin算子，从而强制使用Sort-Merge Join。

当然这仍然不算优雅的解决方法，接下来简要看看Flink 1.16版本中提出的更好一点的方案：Adaptive Hash Join。

Adaptive Hash Join的实现

所谓adaptive（自适应），就是指Hash Join递归超限时，不必让任务失败，而是将这些大分区自动转为Sort-Merge Join来处理。

Blink Runtime中的哈希表有两种，即BinaryHashTable（key的类型为BinaryRowData）和LongHybridHashTable（key的类型为Long）。以前者为例，查看其prepareNextPartition()方法，该方法负责递归地取得下一个要处理的哈希分区。

 private boolean prepareNextPartition() throws IOException { // finalize and cleanup the partitions of the current table // ...... // there are pending partitions final BinaryHashPartition p = this.partitionsPending.get(0); // ...... final int nextRecursionLevel = p.getRecursionLevel() + 1; if (nextRecursionLevel == 2) { LOG.info("Recursive hash join: partition number is " + p.getPartitionNumber()); } else if (nextRecursionLevel > MAX_RECURSION_DEPTH) { LOG.info( "Partition number [{}] recursive level more than {}, process the partition using SortMergeJoin later.", p.getPartitionNumber(), MAX_RECURSION_DEPTH); // if the partition has spilled to disk more than three times, process it by sort merge // join later this.partitionsPendingForSMJ.add(p); // also need to remove it from pending list this.partitionsPending.remove(0); // recursively get the next partition return prepareNextPartition(); } // build the next table; memory must be allocated after this call buildTableFromSpilledPartition(p, nextRecursionLevel); // set the probe side setPartitionProbeReader(p); // unregister the pending partition this.partitionsPending.remove(0); this.currentRecursionDepth = p.getRecursionLevel() + 1; // recursively get the next return nextMatching(); } 

注意当递归深度超过MAX_RECURSION_DEPTH（常量定义即为3）时，会将分区直接放入一个名为partitionsPendingForSMJ的容器中，等待做Sort-Merge Join。另外，在该方法调用的buildTableFromSpilledPartition()方法（对溢写分区执行build操作）开头，去掉了对递归超限的判断，也就是说Hash join exceeded maximum number of recursions异常已经成为历史。

那么等待做Sort-Merge Join的分区是如何被处理的？查看Blink Runtime中的HashJoinOperator算子，在构造该算子时，需要比原来多传入一个SortMergeJoinFunction的实例：

private final SortMergeJoinFunction sortMergeJoinFunction; 

SortMergeJoinFunction实际上是将旧版的SortMergeJoinOperator处理逻辑抽离出来的类，算法本身没有任何变化。然后从哈希表中读取前述的partitionsPendingForSMJ容器，对每个分区的build侧和probe侧分别执行Sort-Merge Join操作即可。

 / * If here also exists partitions which spilled to disk more than three time when hash join end, * means that the key in these partitions is very skewed, so fallback to sort merge join * algorithm to process it. */ private void fallbackSMJProcessPartition() throws Exception { if (!table.getPartitionsPendingForSMJ().isEmpty()) { // release memory to MemoryManager first that is used to sort merge join operator table.releaseMemoryCacheForSMJ(); // initialize sort merge join operator LOG.info("Fallback to sort merge join to process spilled partitions."); initialSortMergeJoinFunction(); fallbackSMJ = true; for (BinaryHashPartition p : table.getPartitionsPendingForSMJ()) { // process build side RowIterator<BinaryRowData> buildSideIter = table.getSpilledPartitionBuildSideIter(p); while (buildSideIter.advanceNext()) { processSortMergeJoinElement1(buildSideIter.getRow()); } // process probe side ProbeIterator probeIter = table.getSpilledPartitionProbeSideIter(p); BinaryRowData probeNext; while ((probeNext = probeIter.next()) != null) { processSortMergeJoinElement2(probeNext); } } // close the HashTable closeHashTable(); // finish build and probe sortMergeJoinFunction.endInput(1); sortMergeJoinFunction.endInput(2); LOG.info("Finish sort merge join for spilled partitions."); } } private void initialSortMergeJoinFunction() throws Exception { sortMergeJoinFunction.open( true, this.getContainingTask(), this.getOperatorConfig(), (StreamRecordCollector) this.collector, this.computeMemorySize(), this.getRuntimeContext(), this.getMetricGroup()); } private void processSortMergeJoinElement1(RowData rowData) throws Exception { if (leftIsBuild) { sortMergeJoinFunction.processElement1(rowData); } else { sortMergeJoinFunction.processElement2(rowData); } } private void processSortMergeJoinElement2(RowData rowData) throws Exception { if (leftIsBuild) { sortMergeJoinFunction.processElement2(rowData); } else { sortMergeJoinFunction.processElement1(rowData); } } 

与BinaryHashTable不同，LongHybridHashTable的join逻辑全部是代码生成的，在对应的生成器LongHashJoinGenerator中，可以看到与上文类似的代码，看官可以自行找来读读。

The End

民那晚安晚安。