本文分享自华为云社区《ClickHouse一种高性能分布式join查询模型（Colocate Join）》，作者：tiantangniao 。

ClickHouse 是一款开源的面向联机分析处理的列式数据库，具有极致的压缩率和极速查询性能。

ClickHouse 支持 SQL 查询，基于大宽表的聚合分析查询性能非常优异，在特定场景下 ClickHouse 也具备较优的 join 性能。本文将介绍业界 MPP 分布式数据库 join 查询模型，以及 ClickHouse 的分布式查询原理解析和 Colocate join 性能表现。

1. ClickHouse 分布式加入

ClicHouse 分布式 join 通常涉及到左右表为分布式表，分布式执行过程中需要将数据在节点间进行交换，我们将数据在节点间交换的动作在分布式执行计划中称为数据的流动 streaming 算子，ClickHouse 支持的 streaming 算子有如下三种：

• 广播加入

• 舒弗加盟

• 共置联接

以上第一种其实是数据广播算子，第二种为数据重分布算子，第三种为数据在本地不需要分布式交换。其实对于 ClickHouse 来说，说是实现了 Shuffle JOIN 比较勉强，其只实现了类 Broadcast JOIN 类型，ClickHouse 的当前的分布式 join 查询框架更多的还是实现了两阶段查询按任务（这里不详细讲解，后续几个章节分别进行展开讲解，大家可以体会），业界 MPP 数据库分布式 join 查询框架模型的数据在节点间交换 Streaming 算子通常为以下几种：

第一种 Gather 算子类似于在 ClickHouse 中的 SQL 发起 initiator 节点，第一阶段在各个节点完成本地 join 后，会将各节点结果发送给 initiator 节点进行第二阶段的汇总工作，initiator 节点再讲结果发送给客户端;第二种为数据广播，单个节点将自己拥有的数据发送给目标节点，对应到 ClickHouse 为 Broadcast JOIN;第三种为数据重分布，数据重分布会将数据按照一定的重分布规则发送到对应节点的目标，对应到 ClickHouse 为 Shuffer JOIN;最后一种数据会在本地进行 join，对应到 ClickHouse 为 Colocate join，其不需要数据重分布或广播，节点间和网络上无数据交换和传播，此实现方式的 join 性能也最佳。以下分别将几种 join 方式在 ClickHouse 中实现方式进行介绍。

1.1 舒弗加盟

1）有如下分布式 Join SQL 语句：

2）执行过程如下：

（1） 客户端将 SQL1 发送给集群中一个节点 host-0（initiator/coordinator）;

（2） host-0 节点将任务改写为 SQL2 查询任务;

（3） Coordinator 节点将 SQL2 查询任务下发到集群各个节点执行;

（4） 各节点将 SQL2 解析为 SQL3 子查询;

（5） 子查询被下发到所有节点执行；

（6） 子查询执行完成后将结果集返回到协调节点，如：host-j;

（7） 协调节点将各个子结果集汇总为一个结果集；

（8） 协调节点将结果集发送到集群各个节点，同时将 SQL4 任务下发到各个节点执行；

（9） 各节点在本地将左表的分片和右表子查询结果集进行 join 计算，然后将结果返回到客户端。

3）总结：

• ClickHouse 普通分布式 JOIN 查询并未按 JOIN KEY 去 Shuffle 数据，而是每个节点全量拉取右表数据跟左表分片进行 join 计算;

• 如果右表为分布式表，则集群中每个节点会去执行分布式查询，查询会存在一个非常严重的读放大现象。假设集群有 N 个节点，右表查询会在集群中执行 N*N 次;

• ClickHouse 的这种 join 方式和业界 MPP 的区别：虽然是叫做 Shuffle join/redistribute join，但是从根本来说不是真正的 redistribute join，存在查询放大问题，也是性能较差的一种查询方式。

1.2 广播加入

1）有如下分布式 Join SQL 语句：

2）执行过程如下：

（1） 客户端将 SQL1 发送给集群中一个节点 host-0（initiator/coordinator）;

（2） host-0 节点将任务改写为 SQL2 子查询任务;

（3） Coordinator 节点将 SQL2 子查询任务下发到集群各个节点执行;

（4） 各子节点任务执行完成之后将结果发回到协调节点；

（5） 协调节点将上一步接收到的结果汇总为结果集；

（6） 协调节点将结果集发送到集群各个节点，同时将 SQL3 任务下发到各个节点;

（7） 各节点在本地将左表的分片和右表子查询结果集进行 join 计算，然后将结果及发回到协调节点;

（8） 协调节点将最终结果返回给客户端。

3）总结：

• 右表的查询在 initiator 节点完成后，通过网络发送到其他节点，避免其他节点重复计算，从而避免查询放大问题;

• GLOBAL JOIN 可以看做一个不完整的 Broadcast JOIN 实现。如果 JOIN 的右表数据量较大，就会占用大量网络带宽，导致查询性能降低;

• ClickHouse 的 global join 方式和业界 MPP 的区别：

1. ClickHouse 会将右表过滤结果汇总到一个节点，然后又发送到所有节点，对单节点内存/磁盘空间占用较大，全量数据发送到所有节点，对网络带宽消耗也较大;

2. 而业界 MPP 数据库每个节点并行的将自己一部分数据广播发到所有节点，之后就可以直接进行下一阶段的本地 join 动作，多个节点都能并行执行，同时数据也不需要从一个节点发送到所有节点，对网络和单节点磁盘及内存消耗较少。

1.3 共置联接

1）有如下分布式 Join SQL 语句：

2）执行过程如下：

（1） 客户端将 SQL1 发送给集群中一个节点 host-0（initiator/coordinator）;

（2） host-2 节点将任务改写为 SQL<>子查询任务;

（3） Coordinator 节点将 SQL2 子查询任务下发到集群各个节点执行;

（4） 各子节点任务执行完成之后将结果发回到协调节点；

（5） 协调节点将上一步接收到的结果汇总为结果集返回给客户端。

3）总结：

• 由于数据已经进行了预分区/分布，相同的 JOIN KEY 对应的数据一定存储在同一个计算节点，join 计算过程中不会进行跨节点的数据交换工作，所以无需对右表做分布式查询，也能获得正确结果，并且性能较优。

2. ClickHouse 共置加入

2.1 共置连接原理：

根据“相同 JOIN KEY 必定相同分片”原理，我们将涉及 JOIN 计算的表，按 JOIN KEY 在集群维度作分片。将分布式 JOIN 转为节点的本地 JOIN，极大减少了查询放大问题。按如下操作：

1）将涉及 JOIN 的表字段按 JOIN KEY 用同样分片算法进行分片;

2）将 JOIN SQL 中右表换成相应的本地表名称进行 join。

2.2 共置连接性能：

数据和用例准备

1）环境：准备 2 shard，2 副本共 4 个节点的 ClickHouse 计算节点集群;

2）用例：分别创建 join 字段按 id % 2（2 为 shard 个数，可根据实际集群环境进行调整）取余数据分布方式（相同 id 数据分布到同一个节点），以及 RandRowbin （数据随机 rand 分布）数据分布方式分布式表和本

地表，分布式表指定分布方式，本地表为 Replicated 表，具体用例如下：

• colocate_join_a_local 数据按照 2 分片（id % 2 或哈希取模）进行数据分布;

• 相同分布列的字段 key 的数据会分布到同一个节点;

• 数据通过分布式表 colocate_join_a_dis 把数据写入分布到各数据节点。

• colocate_join_a_local_rand 数据（rand（））随机分布;

• 相同分布列的字段 key 数据会随机分布到各节点;

• 数据通过分布式表 colocate_join_a_dis_rand 写入进行分布。

• 结果对比

2.3 共置加入场景约束

1）数据写入

Colocate join 场景需要用户在系统建设前提前进行数据规划，数据写入时 join 的左右表 join 条件字段需要使用相同哈希算法入库分布，保证 join key 相同数据写入到同一个计算节点上。

• 如果对数据写入时效性要求不太高的场景，可通过分布式表进行生成数据，生成数据简单快捷，性能较慢；

• 如果对数据写入时效性要求较高的场景，可通过应用/中间件写入数据到 local 表，中间件需要实现入库数据分布算法，入库性能较好。

2）扩缩容

• 扩缩容完成后，需要将全部数据重写/重分布一遍，缺点：耗时长，占用存储可能暂时会翻倍，一种节省空间的方式是：逐个表进行重分布，每个表数据重分布完成后可删除重分布前的数据，避免占用过多存储。将来的改进/增强：重分布过程中支持可写在线，重分布尽量少或不影响写入查询的在线操作，减少重分布过程中对客户业务的影响。

3. 总结

业界所宣称的 ClickHouse 只能做大宽表查询，而通过以上分析，事实上在特定场景下 ClickHouse 也可以进行高效的 join（Broadcast join 和 Colocate join）查询，如果将表结构设计及数据分布的足够好，查询性能也并不会太差：Broadcast join 对于大小表关联，需要将小表数据放在右边;Colocate join 需要将 join key 字段使用相同的分布算法，将分布键相同数据分布在同一个计算节点。

对于 ClickHouse 而言，当前优化器能力较弱，如 join 场景 reorder 以及统计信息缺失，基于成本代价估算 CBO 的优化能力较弱，用户 SQL 所写即所得，可能会要求人人都是 DBA，人人都要对 ClickHouse 或数据库有深入的理解及经验才能设计出较优的数据库结构以及写出较高性能的 SQL 语句。对于 ClickHouse 手动挡数据库，将来我们也会在统计信息、CBO 优化器、分布式 join 模型框架、大大表等多表关联查询以及复杂查询上进行优化增强，以降低用户使用门槛，提升用户使用体验。

点击关注，第一时间了解华为云新鲜技术~