本文會稍微講述分散式系統的基本定義和概念，沒基礎的也會看得懂，所以請各位放心食用（小孩愛吃）。

隨著資料量從 TB 級別成長至 PB 級別，單體式架構顯然無法應付現代資料分析與即時查詢的需求。ClickHouse 透過 Distributed Table 與分布式查詢架構，讓資料能夠橫向擴展到數十、數百台節點，並在龐大資料量下依然維持秒級查詢回應。

什麼是 Distributed Table？

Distributed Table 並不是實際存放資料的表，而是 ClickHouse 提供的 查詢路由代理 (Query Router)。它的作用是：

• 接收分布式查詢請求。
• 將查詢根據分片 (Shard) 規則分派至對應的資料節點 (Shard / Replica)。
• 收集各分片節點的查詢結果後，進行合併 (Merge) 回傳給使用者。

核心概念：

自己做的

語法與配置

1. 建立 Remote Table (實際儲存資料的表)

CREATE TABLE cluster_shard.local_visits
(
    Date Date,
    UserID UInt64,
    PageViews UInt32
) ENGINE = MergeTree()
ORDER BY (Date, UserID);

2. 建立 Distributed Table (分布式查詢路由表)

在 ClickHouse Cloud 中，若要建立 distributed table，是不支援 Distributed(...) 的，請改使用 remote/remoteSecure 來實作。

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster] AS [db2.]name2 ENGINE = Distributed(cluster, database, table[, sharding_key[, policy_name]]) [SETTINGS name=value, ...]

CREATE TABLE cluster_shard.distributed_visits
AS cluster_shard.local_visits
ENGINE = Distributed(cluster_name, 'cluster_shard', local_visits, rand());

參數說明：

分片策略設計

設計建議：

• 維持分片數量與資料量均衡 (避免資料傾斜)。
• 分片鍵選擇應盡量與查詢 WHERE 條件一致，提升 Data Skipping 效率。
• 盡可能避免查詢跨所有 Shard (如無分片條件的全表掃描)。

分片與副本架構

分布式查詢的運作流程

1. Client 發起查詢 (Distributed Table)。
2. Distributed Table 根據 Sharding Key (或 rand()) 決定分派至哪些 Shard。
3. 每個 Shard 中的 Replica Node 會由 ClickHouse 根據負載自動選擇 (或讀取任意 Replica)。
4. 每個節點執行查詢，將部分結果回傳至 Distributed Table 節點。
5. Distributed Table 節點將來自各 Shard 的結果進行合併 (Merge)，並回傳給 Client。

Replica 設計策略

優化方向

額外補充

由於內容較多，若有興趣可查看官方文件，以下是一些額外補充：

1. INSERT, SELECT

2. Distributed Tables 中的 local 節點優先讀取 (prefer_localhost_replica)

• 如果 Distributed Table 與 remote_servers 中的 Replica 節點為同一台機器，則查詢預設會優先從 Local 節點讀取，減少網路傳輸。
• 這行為由 prefer_localhost_replica 控制，可強制或禁止此優先行為。

3. Data Skipping with Distributed Table

• 若 Distributed Table 中的 Remote Table (如 MergeTree) 設計了 Partition / Primary Key 索引，ClickHouse 仍能在分布式查詢時利用這些索引進行 Data Skipping (裁剪)。
• 前提是 Distributed Table 查詢條件需能對應到分片鍵或分區條件，否則會掃描所有 Shard。

4. insert_distributed_sync 設定 (強制同步寫入)

• 預設 Distributed Table 的 INSERT 操作是非同步的，當資料送至 Remote Shard 後即視為成功。
• 若希望寫入時能等待所有 Shard 回應後才返回成功，可將 insert_distributed_sync 設為 1，確保資料一致性。
• 這會影響寫入延遲，但能保證寫入成功時資料已在所有節點完成寫入。

5. 支持分佈式 DDL (ON CLUSTER 語法)

• 若 clusters.xml 中設定 allow_distributed_ddl_queries = true，可使用：

  CREATE TABLE my_table ON CLUSTER my_cluster ...
  

  讓表結構自動同步到叢集內所有節點。

• 適用於建立 Distributed Table 需要跨節點一致的場景。

6. Distributed_aggregation_memory_efficient

• 分布式查詢的聚合 (Group By) 過程中，若資料量極大會導致記憶體壓力過高。
• 開啟 distributed_aggregation_memory_efficient = 1 後，ClickHouse 會讓各 Shard 返回部分聚合結果，再分批進行聚合 Merge，降低 Driver Node 的記憶體負擔。