ClickHouse使用姿勢(shì)系列之分布式JOIN

JOIN操作是OLAP場(chǎng)景無(wú)法繞開(kāi)的，且使用廣泛的操作。對(duì)ClickHouse而言，非常有必要對(duì)分布式JOIN實(shí)現(xiàn)作深入研究。

在介紹分布式JOIN之前，我們看看ClickHouse 單機(jī)JOIN是如何實(shí)現(xiàn)的。

1. ClickHouse單機(jī)JOIN實(shí)現(xiàn)

ClickHouse 單機(jī)JOIN操作默認(rèn)采用HASH JOIN算法，可選MERGE JOIN算法。其中，MERGE JOIN算法數(shù)據(jù)會(huì)溢出到磁盤(pán)，性能相比前者較差。本文重點(diǎn)介紹基于HASH JOIN算法的實(shí)現(xiàn)JOIN操作。

ClickHouse JOIN查詢(xún)語(yǔ)法如下：

SELECT?
FROM?
[GLOBAL]?[INNER|LEFT|RIGHT|FULL|CROSS]?[OUTER|SEMI|ANTI|ANY|ASOF]?JOIN?
(ON?)|(USING?)?...

ClickHouse 的 HASH JOIN算法實(shí)現(xiàn)比較簡(jiǎn)單：

• 從right_table 讀取該表全量數(shù)據(jù)，在內(nèi)存中構(gòu)建HASH MAP；
• 從left_table 分批讀取數(shù)據(jù)，根據(jù)JOIN KEY到HASH MAP中進(jìn)行查找，如果命中，則該數(shù)據(jù)作為JOIN的輸出；

從這個(gè)實(shí)現(xiàn)中可以看出，如果right_table的數(shù)據(jù)量超過(guò)單機(jī)可用內(nèi)存空間的限制，則JOIN操作無(wú)法完成。通常，兩表JOIN時(shí)，將較小表作為right_table.

2. ClickHouse分布式JOIN實(shí)現(xiàn)

ClickHouse 是去中心化架構(gòu)，非常容易水平擴(kuò)展集群。當(dāng)以集群模式提供服務(wù)時(shí)候，分布式JOIN查詢(xún)就無(wú)法避免。這里的分布式JOIN通常指，JOIN查詢(xún)中涉及到的left_table 與 right_table 是分布式表。

通常，分布式JOIN實(shí)現(xiàn)機(jī)制無(wú)非如下幾種：

1. Broadcast JOIN
2. Shuffle Join
3. Colocate JOIN

ClickHouse集群并未實(shí)現(xiàn)完整意義上的Shuffle JOIN，實(shí)現(xiàn)了類(lèi)Broadcast JOIN，通過(guò)事先完成數(shù)據(jù)重分布，能夠?qū)崿F(xiàn)Colocate JOIN。

ClickHouse 的分布式JOIN查詢(xún)可以分為兩類(lèi)，帶GLOBAL關(guān)鍵字的，和不帶GLOBAL關(guān)鍵字的情況。

2.1 普通JOIN實(shí)現(xiàn)

2.1 中描述了GLOBAL JOIN的實(shí)現(xiàn)。接下來(lái)看看無(wú)GLOBAL關(guān)鍵字的JOIN如何實(shí)現(xiàn)的：

• initiator 將SQL S中左表分布式表替換為對(duì)應(yīng)的本地表，形成S'
• initiator 將a.中的S'分發(fā)到集群每個(gè)節(jié)點(diǎn)
• 集群節(jié)點(diǎn)執(zhí)行S'，并將結(jié)果匯總到initiator 節(jié)點(diǎn)
• initiator 節(jié)點(diǎn)將結(jié)果返回給客戶(hù)端

如果右表為分布式表，則集群中每個(gè)節(jié)點(diǎn)會(huì)去執(zhí)行分布式查詢(xún)。這里就會(huì)存在一個(gè)非常嚴(yán)重的讀放大現(xiàn)象。假設(shè)集群有N個(gè)節(jié)點(diǎn)，右表查詢(xún)會(huì)在集群中執(zhí)行N*N次。

如圖所示，假設(shè)執(zhí)行的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為分布式表，對(duì)應(yīng)的本地表名為a_local, b_local。則改SQL在分布式執(zhí)行的時(shí)序?yàn)椋?/p>

1. initiator 收到查詢(xún)請(qǐng)求
2. initiator 執(zhí)行分布式查詢(xún)，本節(jié)點(diǎn)和其他節(jié)點(diǎn)執(zhí)行SELECT a_.i, a_.s, b_.t FROM a_local AS a_ JOIN b_all as b_ ON a_.i = b_.i即左表分布式表更改為本地表名。該SQL在集群范圍內(nèi)并行執(zhí)行。
3. 集群節(jié)點(diǎn)收到2）中SQL后，分析出右表時(shí)分布式表，則觸發(fā)一次分布式查詢(xún)：SELECT b_.i, b_.t FROM b_local AS b_ 集群各節(jié)點(diǎn)并發(fā)執(zhí)行，并合并結(jié)果，記為subquery.
4. 集群節(jié)點(diǎn)完成3）中SQL執(zhí)行后，執(zhí)行 SELECT a_.i, a_.s, b_.t FROM a_local AS a_ JOIN subquery as b_ ON a_.i = b_.i其中subquery表示2中執(zhí)行的結(jié)果
5. 各節(jié)點(diǎn)執(zhí)行完成JOIN計(jì)算后，向initiator節(jié)點(diǎn)發(fā)送數(shù)據(jù)

可以看出，ClickHouse 普通分布式JOIN查詢(xún)是一個(gè)簡(jiǎn)單版的Shuffle JOIN的實(shí)現(xiàn)，或者說(shuō)是一個(gè)不完整的實(shí)現(xiàn)。不完整的地方在于，并未按JOIN KEY去Shuffle數(shù)據(jù)，而是每個(gè)節(jié)點(diǎn)全量拉去右表數(shù)據(jù)。這里實(shí)際上是存在著優(yōu)化空間的。

在生產(chǎn)環(huán)境中，查詢(xún)放大對(duì)查詢(xún)性能的影響是不可忽略的。

2.2 GLOBAL JOIN 實(shí)現(xiàn)

GLOBAL JOIN 計(jì)算過(guò)程如下：

1. 若右表為子查詢(xún)，則initiator完成子查詢(xún)計(jì)算；
2. initiator 將右表數(shù)據(jù)發(fā)送給集群其他節(jié)點(diǎn)；
3. 集群節(jié)點(diǎn)將左表本地表與右表數(shù)據(jù)進(jìn)行JOIN計(jì)算；
4. 集群其他節(jié)點(diǎn)將結(jié)果發(fā)回給initiator節(jié)點(diǎn)；
5. initiator 將結(jié)果匯總，發(fā)給客戶(hù)端；

GLOBAL JOIN 可以看做一個(gè)不完整的Broadcast JOIN實(shí)現(xiàn)。如果JOIN的右表數(shù)據(jù)量較大，就會(huì)占用大量網(wǎng)絡(luò)帶寬，導(dǎo)致查詢(xún)性能降低。

如圖所示，假設(shè)執(zhí)行的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為分布式表，對(duì)應(yīng)的本地表名為a_local, b_local。則改SQL在分布式執(zhí)行的時(shí)序?yàn)椋?/p>

1. initiator 收到查詢(xún)請(qǐng)求
2. initiator 和集群其他節(jié)點(diǎn)均執(zhí)行SELECT b_.i, b_.t FROM b_local AS b_即左表分布式表更改為本地表名。該SQL在集群范圍內(nèi)并行執(zhí)行。匯總結(jié)果，記錄為subquery。
3. initiator 將2)中subquery發(fā)送到集群中其他節(jié)點(diǎn)，并觸發(fā)分布式查詢(xún)：SELECT a_.i, a_.s, b_.t FROM a_local AS a_ JOIN subquery as b_ ON a_.i = b_.i其中subquery表示2)中執(zhí)行的結(jié)果
4. 各節(jié)點(diǎn)執(zhí)行完成JOIN計(jì)算后，向initiator節(jié)點(diǎn)發(fā)送數(shù)據(jù)

可以看出，GLOBAL JOIN 將右表的查詢(xún)?cè)趇nitiator節(jié)點(diǎn)上完成后，通過(guò)網(wǎng)絡(luò)發(fā)送到其他節(jié)點(diǎn)，避免其他節(jié)點(diǎn)重復(fù)計(jì)算，從而避免查詢(xún)放大。

3. 分布式JOIN最佳實(shí)踐

在清楚了ClickHouse 分布式JOIN查詢(xún)實(shí)現(xiàn)后，我們總結(jié)一些實(shí)際經(jīng)驗(yàn)。

• 一、盡量減少JOIN右表數(shù)據(jù)量

ClickHouse根據(jù)JOIN的右表數(shù)據(jù)，構(gòu)建HASH MAP，并將SQL中所需的列全部讀入內(nèi)存中。如果右表數(shù)據(jù)量過(guò)大，節(jié)點(diǎn)內(nèi)存無(wú)法容納后，無(wú)法完成計(jì)算。

在實(shí)際中，我們通常將較小的表作為右表，并盡可能增加過(guò)濾條件，降低進(jìn)入JOIN計(jì)算的數(shù)據(jù)量。

• 二、利用GLOBAL JOIN 避免查詢(xún)放大帶來(lái)性能損失

如果右表或者子查詢(xún)的數(shù)據(jù)量可控，可以使用GLOBAL JOIN來(lái)避免讀放大。需要注意的是，GLOBAL JOIN 會(huì)觸發(fā)數(shù)據(jù)在節(jié)點(diǎn)之間傳播，占用部分網(wǎng)絡(luò)流量。如果數(shù)據(jù)量較大，同樣會(huì)帶來(lái)性能損失。

• 三、數(shù)據(jù)預(yù)分布實(shí)現(xiàn)Colocate JOIN

當(dāng)JOIN涉及的表數(shù)據(jù)量都非常大時(shí)，讀放大，或網(wǎng)絡(luò)廣播都帶來(lái)巨大性能損失時(shí)，我們就需要采取另外一種方式來(lái)完成JOIN計(jì)算了。

根據(jù)“相同JOIN KEY必定相同分片”原理，我們將涉及JOIN計(jì)算的表，按JOIN KEY在集群維度作分片。將分布式JOIN轉(zhuǎn)為為節(jié)點(diǎn)的本地JOIN，極大減少了查詢(xún)放大問(wèn)題。

如果如下操作：

• 將涉及JOIN的表按JOIN KEY分片
• 根據(jù)2.2節(jié)描述，將JOIN預(yù)計(jì)中右表?yè)Q成相應(yīng)的本地表

如圖所示，執(zhí)行的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為分布式表，對(duì)應(yīng)的本地表名為a_local, b_local。則改SQL在分布式執(zhí)行的時(shí)序?yàn)椋?/p>

1. initiator 收到查詢(xún)請(qǐng)求
2. initiator 發(fā)起一次分布式查詢(xún)，本機(jī)以及其他節(jié)點(diǎn)執(zhí)行：SELECT a_.i, a_.s, b_.t FROM a_local AS a_ JOIN b_local as b_ ON a_.i = b_.i
3. 各節(jié)點(diǎn)執(zhí)行完成JOIN計(jì)算后，向initiator節(jié)點(diǎn)發(fā)送數(shù)據(jù)

由于數(shù)據(jù)以及預(yù)分區(qū)了，相同的JOIN KEY對(duì)應(yīng)的數(shù)據(jù)一定在一起，不會(huì)跨節(jié)點(diǎn)存在，所以無(wú)需對(duì)右表做分布式查詢(xún)，也能獲得正確結(jié)果。

4. 總結(jié)

本文介紹了ClickHouse JOIN實(shí)現(xiàn)原理，并根據(jù)原理介紹了相關(guān)的最佳實(shí)踐：

• 減少JOIN右表數(shù)據(jù)量
• 避免查詢(xún)放大帶來(lái)性能損失
• 數(shù)據(jù)預(yù)分布實(shí)現(xiàn)Colocate JOIN;