文章目錄

• Distributed引擎
• • 分布式寫入流程
  • • 數據寫入分片
    • 副本復制數據
  • 分布式查詢流程
  • • 多副本的路由規則
    • 多分片查詢的流程
    • 使用Global優化分布式子查詢

---

Distributed引擎

Distributed表引擎是分布式表的代名詞，它自身不存儲任何數據，而是作為數據分片的透明代理，能夠自動路由數據至集群中的各個節點，所以Distributed表引擎需要和其他數據表引擎一起協同工作。

ClickHouse并不像其他分布式系統那樣，擁有高度自動化的分片功能。ClickHouse提供了本地表（Local Table） 與 分布式表（Distributed Table） 的概念

由Distributed表將數據寫入多個分片

• 本地表：通常以_local為后綴進行命名。本地表是承接數據的載體，可以使用非Distributed的任意表引擎，一張本地表對應了一個數據分片。
• 分布式表：通常以_all為后綴進行命名。分布式表只能使用Distributed表引擎，它與本地表形成一對多的映射關系，日后將通過分布式表代理操作多張本地表。

分布式寫入流程

在向集群內的分片寫入數據時，通常有兩種思路

• 借助外部計算系統，事先將數據均勻分片，再借由計算系統直接將數據寫入ClickHouse集群的各個本地表。
• 通過Distributed表引擎代理寫入分片數據。

第一種方案通常擁有更好的寫入性能，因為分片數據是被并行點對點寫入的。但是這種方案的實現主要依賴于外部系統，而不在于ClickHouse自身，所以這里主要會介紹第二種思路。為了便于理解整個過程，這里會將分片寫入、副本復制拆分成兩個部分進行講解。

數據寫入分片

由Distributed表將數據寫入多個分片

在第一個分片節點寫入本地分片數據：首先在CH5節點，對分布式表test_shard_2_all執行INSERT查詢，嘗試寫入10、30、200和55四行數據。執行之后分布式表主要會做兩件事情：

• 根據分片規則劃分數據

• 將屬于當前分片的數據直接寫入本地表test_shard_2_local。

第一個分片建立遠端連接，準備發送遠端分片數據：將歸至遠端分片的數據以分區為單位，分別寫入/test_shard_2_all存儲目錄下的臨時bin文件，接著，會嘗試與遠端分片節點建立連接。

第一個分片向遠端分片發送數據：此時，會有另一組監聽任務負責監聽/test_shard_2_all目錄下的文件變化，這些任務負責將目錄數據發送至遠端分片，其中，每份目錄將會由獨立的線程負責發送，數據在傳輸之前會被壓縮。

第二個分片接收數據并寫入本地：CH6分片節點確認建立與CH5的連接，在接收到來自CH5發送的數據后，將它們寫入本地表。

由第一個分片確認完成寫入：最后，還是由CH5分片確認所有的數據發送完畢。

可以看到，在整個過程中，Distributed表負責所有分片的寫入工作。本著誰執行誰負責的原則，在這個示例中，由CH5節點的分布式表負責切分數據，并向所有其他分片節點發送數據。

在由Distributed表負責向遠端分片發送數據時，有異步寫和同步寫兩種模式：

• 如果是異步寫，則在Distributed表寫完本地分片之后，INSERT查詢就會返回成功寫入的信息；
• 如果是同步寫，則在執行INSERT查詢之后，會等待所有分片完成寫入。

副本復制數據

如果在集群的配置中包含了副本，那么除了剛才的分片寫入流程之外，還會觸發副本數據的復制流程。數據在多個副本之間，有兩種復制實現方式：

• Distributed表引擎：副本數據的寫入流程與分片邏輯相同，所以Distributed會同時負責分片和副本的數據寫入工作。但在這種實現方案下，它很有可能會成為寫入的單點瓶頸，所以就有了接下來將要說明的第二種方案。

• ReplicatedMergeTree表引擎：如果使用ReplicatedMergeTree作為本地表的引擎，則在該分片內，多個副本之間的數據復制會交由ReplicatedMergeTree自己處理，不再由Distributed負責，從而為其減負。

使用Distributed與ReplicatedMergeTree分發副本數據的對比示意圖

分布式查詢流程

與數據寫入有所不同，在面向集群查詢數據的時候，只能通過Distributed表引擎實現。當Distributed表接收到SELECT查詢的時候，它會依次查詢每個分片的數據，再合并匯總返回，流程如下：

多副本的路由規則

在查詢數據的時候，如果集群中的某一個分片有多個副本，此時Distributed引擎就會通過負載均衡算法從眾多的副本中選取一個，負載均衡算法有以下四種。

在ClickHouse的服務節點中，擁有一個全局計數器errors_count，當服務發生任何異常時，該計數累積加1。

random（默認）：random算法會選擇errors_count錯誤數量最少的副本，如果多個副本的errors_count計數相同，則在它們之中隨機選擇一個。

nearest_hostname：nearest_hostname可以看作random算法的變種，首先它會選擇errors_count錯誤數量最少的副本，如果多個副本的errors_count計數相同，則選擇集群配置中host名稱與當前host最相似的一個。而相似的規則是以當前host名稱為基準按字節逐位比較，找出不同字節數最少的一個。

in_order：in_order同樣可以看作random算法的變種，首先它會選擇errors_count錯誤數量最少的副本，如果多個副本的errors_count計數相同，則按照集群配置中replica的定義順序逐個選擇。

first_or_random：first_or_random可以看作in_order算法的變種，首先它會選擇errors_count錯誤數量最少的副本，如果多個副本的errors_count計數相同，它首先會選擇集群配置中第一個定義的副本，如果該副本不可用，則進一步隨機選擇一個其他的副本。

多分片查詢的流程

分布式查詢與分布式寫入類似，同樣本著誰執行誰負責的原則，它會由接收SELECT查詢的Distributed表，并負責串聯起整個過程。

首先它會將針對分布式表的SQL語句，按照分片數量將查詢拆分成若干個針對本地表的子查詢，然后向各個分片發起查詢，最后再匯總各個分片的返回結果。

--查詢分布式表 SELECT * FROM distributor_table--轉換為查詢本地表，并將該命令推送到各個分片節點上執行 SELECT * FROM local_table

如下圖

對分布式表執行查詢的執行計劃

查詢各個分片數據：One和Remote步驟是并行執行的，它們分別負責了本地和遠端分片的查詢動作。

合并返回結果：多個分片數據均查詢返回后，在執行節點將所有數據union合并

使用Global優化分布式子查詢

如果現在有一項查詢需求，例如要求找到同時擁有兩個倉庫的用戶，應該如何實現？對于這類交集查詢的需求，可以使用IN子查詢，此時你會面臨兩難的選擇：IN查詢的子句應該使用本地表還是分布式表？（使用JOIN面臨的情形與IN類似）。

使用本地表的問題（可能查詢不到結果）

如果在IN查詢中使用本地表時，如下列語句

SELECT uniq(id) FROM distributed_table WHERE repo = 100 AND id IN (SELECT id FROM local_table WHERE repo = 200)

在實際執行時，分布式表在接收到查詢后會將上述SQL替換成本地表的形式，再發送到每個分片進行執行，此時，每個分片上實際執行的是以下語句

SELECT uniq(id) FROM local_table WHERE repo = 100 AND id IN (SELECT id FROM local_table WHERE repo = 200)

那么此時查詢的最終結果就有可能是錯誤的，因為在單個分片上只保存了部分的數據，這就導致該SQL語句可能沒有匹配到任何數據，如下圖

使用本地表作為IN查詢子句的執行邏輯

使用分布式表的問題（查詢請求被放大N^2倍，N為節點數量）

如果在IN查詢中使用本地表時，如下列語句

SELECT uniq(id) FROM distributed_table WHERE repo = 100 AND id IN (SELECT id FROM distributed_table WHERE repo = 200)

對于此次查詢，每個分片節點不僅需要查詢本地表，還需要再次向其他的分片節點再次發起遠端查詢，如下圖

IN查詢子句查詢放大原因示意

因此可以得出結論，在IN查詢子句使用分布式表的時候，雖然查詢的結果得到了保證，但是查詢請求會被放大N的平方倍，其中N等于集群內分片節點的數量，假如集群內有10個分片節點，則在一次查詢的過程中，會最終導致100次的查詢請求，這顯然是不可接受的。

使用GLOBAL優化查詢

為了解決查詢放大的問題，我們可以使用GLOBAL IN或GLOBAL JOIN進行優化，下面就簡單介紹一下GLOBAL的執行流程

SELECT uniq(id) FROM distributed_table WHERE repo = 100 AND id GLOBAL IN (SELECT id FROM distributed_table WHERE repo = 200)

使用GLOBAL IN查詢的流程示意圖

如上圖，主要有以下五個步驟

將IN子句單獨提出，發起了一次分布式查詢。

將分布式表轉local本地表后，分別在本地和遠端分片執行查詢。

將IN子句查詢的結果進行匯總，并放入一張臨時的內存表進行保存。

將內存表發送到遠端分片節點。

將分布式表轉為本地表后，開始執行完整的SQL語句，IN子句直接使用臨時內存表的數據。

在使用GLOBAL修飾符之后，ClickHouse使用內存表臨時保存了IN子句查詢到的數據，并將其發送到遠端分片節點，以此到達了數據共享的目的，從而避免了查詢放大的問題。由于數據會在網絡間分發，所以需要特別注意臨時表的大小，IN或者JOIN子句返回的數據不宜過大。如果表內存在重復數據，也可以事先在子句SQL中增加DISTINCT以實現去重。

總結

以上是生活随笔為你收集整理的ClickHouse 分布式原理：Distributed引擎的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。