揭秘字节跳动解决ClickHouse复杂查询问题的技术方案( 三 ) _ClickHouse

6. 查询片段执行器（InterpreterPlanSegment）
下面介绍下计划片段是如何执行的，原本 ClickHouse的查询和节点执行主要是 SQL 形式，切分Stag后需要支持执行一个单独的PlanSemgent 。因此 InterpreterPlanSegment 的主要功能就是接受一个序列化后的 PlanSemgent，能够在 Worker 节点上运行整个 PlanSemgent 的逻辑。主要的步骤为：
①根据 input 信息读取数据，如果 input 是具体的 table，则从本地读取数据；如果 input 是一个 exchange input，则从对应的 ExchangeManager 读取数据；
②执行 PlanSemgent 的逻辑；
③输出处理后的结果数据，如果是 Coordinator 节点，就将数据发给 Client；如果是非Coordinator 节点，就按照数据的exchange方式写给本实例对应的 ExchangeManager 。
Interpreter部分我们尽量复用当前ClickHouse的执行逻辑，例如processor 执行方式，process list管理等等。相比于InterpreterSelect逻辑要更简单一些，可以认为1 个Stage只有1个阶段。当然我们也做了很多功能和性能的增强，例如我们支持1个 stage处理多个join等，这样可以减少stage数目和不必要的数据传输，在一张大表（通常情况下是事实表） join 多个维度表的场景有比较好的帮助。
InterpreterPlan Segment执行完会向coordinator上报对应的状态信息。执行异常的时候会将异常信息报告给查询片段调度器，取消Query其他worker的执行。
7. 数据交换（ExchangeManager）

文章插图

ExchangeManager是PlanSegment数据交换的媒介，更是平衡数据上下游处理能力的重要组件。整体上采用 push 的方式，当上游数据 ready 时主动推送给下游，并支持反压。其架构如下图所示：

文章插图

具体的流程如下：
①下游PlanSegment执行时，当input为exchange input时，根据一定的 token 规则（通常由 query_id+segment_id+index_id 等组成）和数据 source 信息，向上游 ExchangeManager 注册对应的数据请求；
②上游ExchangeManager收到请求后，建立上下游数据通道，并将上游的数据推送到下游，如果通道一直建立不了会 block 上游的执行。
在这个过程中，上下游都会通过队列来优化发送和读取，当队列饱和的时候通过反压的机制控制上游的执行速度。由于采用了 push 和队列，这里我们要考虑一个特殊的场景，在某些 case 下下游的 Stage 并不需要读取全部的上游数据，一个典型的场景是 limit 。例如 limit 100，下游 stage 是需要读取 100 条数据即可，而上游可能会输出更大规模的数据，因此在这种情况下，当下游 stage 读到足够的数据后，需要能主动取消上游数据的执行并清空队列。这是一个特定场景的优化，能够大大加速查询时间。
ExchangeManager 需要考虑和优化的点还有：
①细粒度的内存控制，能够按照实例、query、segment 多层次进行内存控制，避免 OOM，更长期的考虑是支持 spill 到磁盘上，降低对内存的使用。为了提升传输效率，小数据需要进行 merge，大数据要 split 。同时，网络处理在某些场景要保证有序性，比如 sort 时，partial sort 和 merge sort 的网络传输必须有序，否则数据可能是有问题的。
②连接复用和网络优化，包括针对上下游在同一个节的场景下选择走内存的交换不走网络，可以减少网络的开销和减少数据序列化、反序列化的代价。另外，由于 ClickHouse 在计算方面做了非常充足的优化，有些场景下甚至内存带宽成为瓶颈，我们在ExchangeManager的一些场景上也应用zero copy等技术来减少内存的拷贝。
③异常处理和监控，相比于单机执行，分布式情况下异常情况更复杂且不好感知。通过重试能避免一些节点的暂时高负载或者异常，以及出问题时能够快速感知、排查和做针对性解决和优化。这里的工程实践更多一些。
03
优化与诊断
1. Join 多种实现
根据数据的规模和分布，我们支持了多种Join实现，目前已经支持的有：
①Shuffle Join，最通用的 Join；
②Broadcast Join，针对大表Join小表的场景，通过把右表广播到左表的所有 worker 节点来减少左表的传输；
③Colocate Join，针对左右表根据Join key保持相通分布的场景，减少左右表数据传输。