ClickHouse要了解的騷氣join操作

1. ClickHouse單機JOIN實現(xiàn)

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算法實現(xiàn)比較簡單：

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

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

• Broadcast JOIN
• Shuffle Join
• Colocate JOIN

2.1 普通JOIN實現(xiàn)

• a. initiator 將SQL S中左表分布式表替換為對應的本地表，形成S'
• b. initiator 將a.中的S'分發(fā)到集群每個節(jié)點
• c. 集群節(jié)點執(zhí)行S'，并將結果匯總到initiator 節(jié)點
• d. initiator 節(jié)點將結果返回給客戶端

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在分布式執(zhí)行的時序為：

• 1）initiator 收到查詢請求
• 2) initiator 執(zhí)行分布式查詢，本節(jié)點和其他節(jié)點執(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é)點收到2）中SQL后，分析出右表時分布式表，則觸發(fā)一次分布式查詢：
  SELECT b_.i, b_.t FROM b_local AS b_ 集群各節(jié)點并發(fā)執(zhí)行，并合并結果，記為subquery.
• 4）集群節(jié)點完成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í)行的結果
• 5) 各節(jié)點執(zhí)行完成JOIN計算后，向initiator節(jié)點發(fā)送數(shù)據(jù)

2.2 GLOBAL JOIN 實現(xiàn)

• a. 若右表為子查詢，則initiator完成子查詢計算；
• b. initiator 將右表數(shù)據(jù)發(fā)送給集群其他節(jié)點；
• c. 集群節(jié)點將左表本地表與右表數(shù)據(jù)進行JOIN計算；
• d. 集群其他節(jié)點將結果發(fā)回給initiator節(jié)點；
• e. initiator 將結果匯總，發(fā)給客戶端；

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在分布式執(zhí)行的時序為：

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

3. 分布式JOIN最佳實踐

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

• 二、利用GLOBAL JOIN 避免查詢放大帶來性能損失

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

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

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在分布式執(zhí)行的時序為：

• 1）initiator 收到查詢請求
• 2）initiator 發(fā)起一次分布式查詢，本機以及其他節(jié)點執(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é)點執(zhí)行完成JOIN計算后，向initiator節(jié)點發(fā)送數(shù)據(jù)

4. 總結

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