【ClickHouse系列】分布式下的IN/JOIN及GLOBAL关键字

青旅半醒 2022-12-01 03:56 125阅读 0赞

## 分布式下的IN/JOIN ##

如果是在单机情况下，涉及到IN/JOIN时是没有什么问题的，但是在分布式情况下就不一样了，ClickHouse是支持多分片多副本的，创建表也提供了友好的`ON CLUSTER [name]`的方式，所以就是建议使用者将数据进行分片处理增加读的效率，但也随之产生了单节点数据不完整的问题。如果SQL中涉及子查询就不得不有特殊的处理方式。

## ClickHouse中的分布式子查询 ##

在ClickHouse中为了方便做分布式查询，特意提供了`Distributed`表引擎，这个表引擎实际上是不存储数据的，单查这个表时，实际上是将SQL分发到该表引擎所关联的本地及远程节点执行，并把结果再汇总回来，类似一种分布式视图的效果。所以针对`Distributed`表的分布式查询情况做个汇总。

### 1.主子查询都查询本地表 ###

例如：

SELECT uniq(UserID) FROM local_table WHERE CounterID = 101500 AND UserID IN (SELECT UserID FROM local_table WHERE CounterID = 34)

这种情况不会涉及网络数据传输，所有查询操作仅仅涉及执行节点，也仅仅会返回执行节点上的匹配数据。

### 2.主查询使用分布式表，子查询使用本地表 ###

例如：

SELECT uniq(UserID) FROM distributed_table WHERE CounterID = 101500 AND UserID IN (SELECT UserID FROM local_table WHERE CounterID = 34)

这种情况ck就会对SQL进行重写，将分布式表转换成本地表，并将SQL分发到存在`local_table`表的所有节点进行执行，再将结果进行汇总回传。其实可以理解成在每个存在`local_table`表的节点上都分别执行一遍第一种查询情况，最后进行合并回传。这种方式会因为数据不全导致结果错误，如果数据冗余，也会造成结果重复。

这种情况下，如果有n个节点，就会有n次查询操作。

### 3.主查询本地表，子查询使用使用分布式表 ###

例如：

SELECT uniq(UserID) FROM local_table WHERE CounterID = 101500 AND UserID IN (SELECT UserID FROM distributed_table WHERE CounterID = 34)

这种情况ck同样会对SQL进行重写，但此时不是将分布式表转换为本地表，而是直接分发这个SQL语句到存在`local_table`表的所有节点，在子查询是分布式表的情况下，每个接收到分发请求的节点先进行子查询，即到各个存在`local_table`表的节点执行`SELECT UserID FROM local_table WHERE CounterID = 34`再汇总回来，再和主查询语句继续执行查询，即每个被分发的节点都需要走一次这个流程，将最终的结果回传给初始执行节点，由初始执行节点将结果集合并，完成查询。

好像稍微有点绕，可以把这个SQL的主查询当作A，子查询当作B，假设有3个分片节点，那么需要先考虑第一个节点的执行情况，第一个节点为了能拿到子查询所有的数据，会先上自己上执行B，再去第二个节点执行B，再去第三个节点上执行B，然后再自己上执行A。同样，换到第二个节点，它同样会去一和三以及自己上执行B，再执行A。第三个几点以此类推，最终回传到初始执行节点进行汇总。

这种情况下，如果有n个节点，就会有n\*n次查询操作。

### 4.主子查询都查询分布式表 ###

例如：

SELECT uniq(UserID) FROM distributed_table WHERE CounterID = 101500 AND UserID IN (SELECT UserID FROM distributed_table WHERE CounterID = 34)

这种情况其实和第三种情况类似，但是主查询中也使用了分布式表，所以又多了一次查询汇总，所以如果有n个几点，那么会产生2\*n\*n次查询操作。

### 5.主子查询都查询分布式表，且使用GLOBAL关键字 ###

例如：

SELECT uniq(UserID) FROM local_table WHERE CounterID = 101500 AND UserID GLOBAL IN (SELECT UserID FROM distributed_table WHERE CounterID = 34)

为什么要用`GLOBAL`关键字？

是因为上述的方法虽然能得到正确的查询结果，但是产生了查询放大，而且放大倍数非常大，为了解决这个问题，引入了`GLOBAL`关键字。

使用`GLOBAL`修饰后，会将子查询在初始执行节点进行查询汇总，存储为临时表，并在SQL分发时携带该临时表数据到各个节点进行查询，最终汇总结果到初始查询节点。

这种情况下，如果有n个节点，就会仅有2\*n次查询操作。大限度的降低了查询放大问题。

### 补充： ###

*  在使用GLOBAL关键字时，虽然最大限度的降低了查询放大，但是如果数据量过大，产生的临时表就会很大，也会受到网络稳定性和网络带宽的限制。ck在做JOIN时都是采用发送右表，所以ck在做分布式IN/JOIN时的效率不太好，所以在编写SQL时一定要多考虑这部分影响。
 *  ck不支持数据的重分布，并不能将join key相同的数据落到同一节点，所以还不能支持将分布式join转换为本地join并汇总的方式。