JOIN操作是OLAP场景无法绕开的,且使用广泛的操作。对ClickHouse而言,非常有必要对分布式JOIN实现作深入研究。
在介绍分布式JOIN之前,我们看看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的数据量超过单机可用内存空间的限制,则JOIN操作无法完成。通常,两表JOIN时,将较小表作为right_table.
ClickHouse 是去中心化架构,非常容易水平扩展集群。当以集群模式提供服务时候,分布式JOIN查询就无法避免。这里的分布式JOIN通常指,JOIN查询中涉及到的left_table 与 right_table 是分布式表。
通常,分布式JOIN实现机制无非如下几种:
ClickHouse集群并未实现完整意义上的Shuffle JOIN,实现了类Broadcast JOIN,通过事先完成数据重分布,能够实现Colocate JOIN。
ClickHouse 的分布式JOIN查询可以分为两类,带GLOBAL关键字的,和不带GLOBAL关键字的情况。
2.1 中描述了GLOBAL JOIN的实现。接下来看看无GLOBAL关键字的JOIN如何实现的:
如果右表为分布式表,则集群中每个节点会去执行分布式查询。这里就会存在一个非常严重的读放大现象。假设集群有N个节点,右表查询会在集群中执行N*N次。
如图所示,执行的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在分布式执行的时序为:
可以看出,ClickHouse 普通分布式JOIN查询是一个简单版的Shuffle JOIN的实现,或者说是一个不完整的实现。不完整的地方在于,并未按JOIN KEY去Shuffle数据,而是每个节点全量拉去右表数据。这里实际上是存在着优化空间的。
在生产环境中,查询放大对查询性能的影响是不可忽略的。
GLOBAL JOIN 计算过程如下:
GLOBAL JOIN 可以看做一个不完整的Broadcast JOIN实现。如果JOIN的右表数据量较大,就会占用大量网络带宽,导致查询性能降低。
如图所示,执行的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在分布式执行的时序为:
可以看出,GLOBAL JOIN 将右表的查询在initiator节点上完成后,通过网络发送到其他节点,避免其他节点重复计算,从而避免查询放大。
在清楚了ClickHouse 分布式JOIN查询实现后,我们总结一些实际经验。
ClickHouse根据JOIN的右表数据,构建HASH MAP,并将SQL中所需的列全部读入内存中。如果右表数据量过大,节点内存无法容纳后,无法完成计算。
在实际中,我们通常将较小的表作为右表,并尽可能增加过滤条件,降低进入JOIN计算的数据量。
如果右表或者子查询的数据量可控,可以使用GLOBAL JOIN来避免读放大。需要注意的是,GLOBAL JOIN 会触发数据在节点之间传播,占用部分网络流量。如果数据量较大,同样会带来性能损失。
当JOIN涉及的表数据量都非常大时,读放大,或网络广播都带来巨大性能损失时,我们就需要采取另外一种方式来完成JOIN计算了。
根据“相同JOIN KEY必定相同分片”原理,我们将涉及JOIN计算的表,按JOIN KEY在集群维度作分片。将分布式JOIN转为为节点的本地JOIN,极大减少了查询放大问题。
如果如下操作:
如图所示,执行的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在分布式执行的时序为:
由于数据以及预分区了,相同的JOIN KEY对应的数据一定在一起,不会跨节点存在,所以无需对右表做分布式查询,也能获得正确结果。
本文介绍了ClickHouse JOIN实现原理,总结了ClickHouse JOIN的最佳实践:
游戏市场的热度已经不言而喻,随着民众生活水平的提升,大家对于精神娱乐生活的...
一、数据中台是真的热 在2018年之前可能只有一少部分人在谈中台,从2018年下半年...
开源 RPC 框架有哪些呢?一类是跟某种特定语言平台绑定的,另一类是与语言无关即...
为了使伸缩组自动加入的实例自动部署应用,您需要创建私有镜像,确保该镜像上有...
最近,在为 Coco 优化分层架构之时,我陷入了各种决策困难之中。所以我通过不断...
来源 | 阿里飞天CIO学堂微信公众号 金融数字化转型过程中,市场的细微变化,客户...
与普通的IDC机房或服务器厂商相比,阿里云提供的云服务器ECS具有高可用性、安全...
本文转载自公众号读芯术(ID:AI_Discovery) 如果你即将要面临大型科技公司的技术...
计算的下一步发展是什么,将如何影响组织的战略?专家预测了边缘计算在2021年的发...
一、背景 ? 我们大部分人的编程习惯都是线性编程,所谓线性编程就是一个请求涉及...