ClickHouse使用姿势系列之分布式JOIN

JOIN操作是OLAP场景无法绕开的,且使用广泛的操作。对ClickHouse而言,非常有必要对分布式JOIN实现作深入研究。

在介绍分布式JOIN之前,我们看看ClickHouse 单机JOIN是如何实现的。

1. ClickHouse单机JOIN实现

ClickHouse 单机JOIN操作默认采用HASH JOIN算法,可选MERGE JOIN算法。其中,MERGE JOIN算法数据会溢出到磁盘,性能相比前者较差。本文重点介绍基于HASH JOIN算法的实现JOIN操作。

ClickHouse JOIN查询语法如下:

SELECT <expr_list>
FROM <left_table>
[GLOBAL] [INNER|LEFT|RIGHT|FULL|CROSS] [OUTER|SEMI|ANTI|ANY|ASOF] JOIN <right_table>
(ON <expr_list>)|(USING <column_list>) ...

ClickHouse 的 HASH JOIN算法实现比较简单:

  • 从right_table 读取该表全量数据,在内存中构建HASH MAP;
  • 从left_table 分批读取数据,根据JOIN KEY到HASH MAP中进行查找,如果命中,则该数据作为JOIN的输出;
1.jpg

从这个实现中可以看出,如果right_table的数据量超过单机可用内存空间的限制,则JOIN操作无法完成。通常,两表JOIN时,将较小表作为right_table.

2. ClickHouse分布式JOIN实现

ClickHouse 是去中心化架构,非常容易水平扩展集群。当以集群模式提供服务时候,分布式JOIN查询就无法避免。这里的分布式JOIN通常指,JOIN查询中涉及到的left_table 与 right_table 是分布式表。

通常,分布式JOIN实现机制无非如下几种:

  • Broadcast JOIN
  • Shuffle Join
  • Colocate JOIN

ClickHouse集群并未实现完整意义上的Shuffle JOIN,实现了类Broadcast JOIN,通过事先完成数据重分布,能够实现Colocate JOIN。

ClickHouse 的分布式JOIN查询可以分为两类,带GLOBAL关键字的,和不带GLOBAL关键字的情况。

2.1 普通JOIN实现

2.1 中描述了GLOBAL JOIN的实现。接下来看看无GLOBAL关键字的JOIN如何实现的:

  • a. initiator 将SQL S中左表分布式表替换为对应的本地表,形成S'
  • b. initiator 将a.中的S'分发到集群每个节点
  • c. 集群节点执行S',并将结果汇总到initiator 节点
  • d. initiator 节点将结果返回给客户端

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

a.png

如图所示,执行的SQL为:

SELECT a_.i, a_.s, b_.t FROM a_all as a_ JOIN b_all AS b_ ON a_.i = b_.i

其中,a_all, b_all为分布式表,对应的本地表名为a_local, b_local。则改SQL在分布式执行的时序为:

  • 1)initiator 收到查询请求SELECT a_.i, a_.s, b_.t FROM a_local AS a_ JOIN b_all as b_ ON a_.i = b_.i即左表分布式表更改为本地表名。该SQL在集群范围内并行执行。
  • 2) initiator 执行分布式查询,本节点和其他节点执行
  • 3)集群节点收到2)中SQL后,分析出右表时分布式表,则触发一次分布式查询:SELECT b_.i, b_.t FROM b_local AS b_集群各节点并发执行,并合并结果,记为subquery.
  • 4)集群节点完成3)中SQL执行后,执行SELECT a_.i, a_.s, b_.t FROM a_local AS a_ JOIN subquery as b_ ON a_.i = b_.i其中subquery表示2中执行的结果
  • 5) 各节点执行完成JOIN计算后,向initiator节点发送数据

可以看出,ClickHouse 普通分布式JOIN查询是一个简单版的Shuffle JOIN的实现,或者说是一个不完整的实现。不完整的地方在于,并未按JOIN KEY去Shuffle数据,而是每个节点全量拉去右表数据。这里实际上是存在着优化空间的。

在生产环境中,查询放大对查询性能的影响是不可忽略的。

2.2 GLOBAL JOIN 实现

GLOBAL JOIN 计算过程如下:

  • a. 若右表为子查询,则initiator完成子查询计算;
  • b. initiator 将右表数据发送给集群其他节点;
  • c. 集群节点将左表本地表与右表数据进行JOIN计算;
  • d. 集群其他节点将结果发回给initiator节点;
  • e. initiator 将结果汇总,发给客户端;

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

b.jpg

如图所示,执行的SQL为:

SELECT a_.i, a_.s, b_.t FROM a_all as a_ GLOBAL JOIN b_all AS b_ ON a_.i = b_.i

其中,a_all, b_all为分布式表,对应的本地表名为a_local, b_local。则改SQL在分布式执行的时序为:

  • 1) initiator 收到查询请求SELECT b_.i, b_.t FROM b_local AS b_即左表分布式表更改为本地表名。该SQL在集群范围内并行执行。汇总结果,记录为subquery。
  • 2) initiator 和集群其他节点均执行
  • 3)initiator 将2)中subquery发送到集群中其他节点,并触发分布式查询:SELECT a_.i, a_.s, b_.t FROM a_local AS a_ JOIN subquery as b_ ON a_.i = b_.i其中subquery表示2)中执行的结果
  • 4) 各节点执行完成JOIN计算后,向initiator节点发送数据

可以看出,GLOBAL JOIN 将右表的查询在initiator节点上完成后,通过网络发送到其他节点,避免其他节点重复计算,从而避免查询放大。

3. 分布式JOIN最佳实践

在清楚了ClickHouse 分布式JOIN查询实现后,我们总结一些实际经验。

  • 一、尽量减少JOIN右表数据量

ClickHouse根据JOIN的右表数据,构建HASH MAP,并将SQL中所需的列全部读入内存中。如果右表数据量过大,节点内存无法容纳后,无法完成计算。

在实际中,我们通常将较小的表作为右表,并尽可能增加过滤条件,降低进入JOIN计算的数据量。

  • 二、利用GLOBAL JOIN 避免查询放大带来性能损失

如果右表或者子查询的数据量可控,可以使用GLOBAL JOIN来避免读放大。需要注意的是,GLOBAL JOIN 会触发数据在节点之间传播,占用部分网络流量。如果数据量较大,同样会带来性能损失。

  • 三、数据预分布实现Colocate JOIN

当JOIN涉及的表数据量都非常大时,读放大,或网络广播都带来巨大性能损失时,我们就需要采取另外一种方式来完成JOIN计算了。

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

如果如下操作:

  • 将涉及JOIN的表按JOIN KEY分片
  • 根据2.2节描述,将JOIN预计中右表换成相应的本地表
c.jpg

如图所示,执行的SQL为:

SELECT a_.i, a_.s, b_.t FROM a_all as a_ JOIN b_local AS b_ ON a_.i = b_.i

其中,a_all, b_all为分布式表,对应的本地表名为a_local, b_local。则改SQL在分布式执行的时序为:

  • 1) initiator 收到查询请求SELECT a_.i, a_.s, b_.t FROM a_local AS a_ JOIN b_local as b_ ON a_.i = b_.i
  • 2) initiator 发起一次分布式查询,本机以及其他节点执行:
  • 3) 各节点执行完成JOIN计算后,向initiator节点发送数据

由于数据以及预分区了,相同的JOIN KEY对应的数据一定在一起,不会跨节点存在,所以无需对右表做分布式查询,也能获得正确结果。

4. 总结

本文介绍了ClickHouse JOIN实现原理,总结了ClickHouse JOIN的最佳实践:

  • 减少JOIN右表数据量
  • 避免查询放大带来性能损失
  • 数据预分布实现Colocate JOIN;
本站文章资源均来源自网络,除非特别声明,否则均不代表站方观点,并仅供查阅,不作为任何参考依据!
如有侵权请及时跟我们联系,本站将及时删除!
如遇版权问题,请查看 本站版权声明
THE END
分享
二维码
海报
ClickHouse使用姿势系列之分布式JOIN
JOIN操作是OLAP场景无法绕开的,且使用广泛的操作。对ClickHouse而言,非常有必要对分布式JOIN实现作深入研究。
<<上一篇
下一篇>>