Clickhouse 分布式表&本地表

CK 分布式表和本地表

ck的表分為兩種:

• 分布式表

  一個邏輯上的表, 可以理解為數據庫中的視圖, 一般查詢都查詢分布式表. 分布式表引擎會將我們的查詢請求路由本地表進行查詢, 然后進行匯總最終返回給用戶.

• 本地表:

  實際存儲數據的表

1. 不寫分布式表的原因

1. 分布式表接收到數據后會將數據拆分成多個parts, 並轉發數據到其它服務器, 會引起服務器間網絡流量增加、服務器merge的工作量增加, 導致寫入速度變慢, 並且增加了Too many parts的可能性.
2. 數據的一致性問題, 先在分布式表所在的機器進行落盤, 然后異步的發送到本地表所在機器進行存儲，中間沒有一致性的校驗, 而且在分布式表所在機器時如果機器出現down機, 會存在數據丟失風險.
3. 數據寫入默認是異步的，短時間內可能造成不一致.
4. 對zookeeper的壓力比較大(待驗證). 沒經過正式測試, 只是看到了有人提出.

2. Replication & Sharding

ClickHouse依靠ReplicatedMergeTree引擎族與ZooKeeper實現了復制表機制, 成為其高可用的基礎.

ClickHouse像ElasticSearch一樣具有數據分片(shard)的概念, 這也是分布式存儲的特點之一, 即通過並行讀寫提高效率. ClickHouse依靠Distributed引擎實現了分布式表機制, 在所有分片（本地表）上建立視圖進行分布式查詢.

3. Replicated Table & ReplicatedMergeTree Engines

不同於HDFS的副本機制(基於集群實現), Clickhouse的副本機制是基於表實現的. 用戶在創建每張表的時候, 可以決定該表是否高可用.

Local_table

CREATE TABLE IF NOT EXISTS {local_table} ({columns}) 
ENGINE = ReplicatedMergeTree('/clickhouse/tables/#_tenant_id_#/#__appname__#/#_at_date_#/{shard}/hits', '{replica}')
partition by toString(_at_date_) sample by intHash64(toInt64(toDateTime(_at_timestamp_)))
order by (_at_date_, _at_timestamp_, intHash64(toInt64(toDateTime(_at_timestamp_))))

支持復制表的引擎都是ReplicatedMergeTree引擎族, 具體可以查看官網:

Data Replication

ReplicatedMergeTree引擎族接收兩個參數:

• ZK中該表相關數據的存儲路徑, ClickHouse官方建議規范化, 例如: /clickhouse/tables/{shard}/[database_name]/[table_name].
• 副本名稱, 一般用{replica}即可.

ReplicatedMergeTree引擎族非常依賴於zookeeper, 它在zookeeper中存儲了大量的數據:

表結構信息、元數據、操作日志、副本狀態、數據塊校驗值、數據part merge過程中的選主信息...

同時, zookeeper又在復制表急之下扮演了三種角色:

元數據存儲、日志框架、分布式協調服務

可以說當使用了ReplicatedMergeTree時, zookeeper壓力特別重, 一定要保證zookeeper集群的高可用和資源.

3.1. 數據同步的流程

1. 寫入到一個節點
2. 通過interserver HTTP port端口同步到其他實例上
3. 更新zookeeper集群記錄的信息

3.2. 重度依賴Zookeeper導致的問題

ck的replicatedMergeTree引擎方案有太多的信息存儲在zk上, 當數據量增大, ck節點數增多, 會導致服務非常不穩定, 目前我們的ck集群規模還小, 這個問題還不嚴重, 但依舊會出現很多和zk有關的問題(詳見遇到的問題).

實際上 ClickHouse 把 ZK 當成了三種服務的結合, 而不僅把它當作一個 Coordinate service(協調服務), 可能這也是大家使用 ZK 的常用用法。ClickHouse 還會把它當作 Log Service(日志服務)，很多行為日志等數字的信息也會存在 ZK 上；還會作為表的 catalog service(元數據存儲)，像表的一些 schema 信息也會在 ZK 上做校驗，這就會導致 ZK 上接入的數量與數據總量會成線性關系。

目前針對這個問題, clickhouse社區提出了一個mini checksum方案, 但是這並沒有徹底解決 znode 與數據量成線性關系的問題. 目前看到比較好的方案是字節的:

我們就基於 MergeTree 存儲引擎開發了一套自己的高可用方案。我們的想法很簡單，就是把更多 ZK 上的信息卸載下來，ZK 只作為 coordinate Service。只讓它做三件簡單的事情：行為日志的 Sequence Number 分配、Block ID 的分配和數據的元信息，這樣就能保證數據和行為在全局內是唯一的。

關於節點，它維護自身的數據信息和行為日志信息，Log 和數據的信息在一個 shard 內部的副本之間，通過 Gossip 協議進行交互。我們保留了原生的 multi-master 寫入特性，這樣多個副本都是可以寫的，好處就是能夠簡化數據導入。圖 6 是一個簡單的框架圖。

以這個圖為例，如果往 Replica 1 上寫，它會從 ZK 上獲得一個 ID，就是 Log ID，然后把這些行為和 Log Push 到集群內部 shard 內部活着的副本上去，然后當其他副本收到這些信息之后，它會主動去 Pull 數據，實現數據的最終一致性。我們現在所有集群加起來 znode 數不超過三百萬，服務的高可用基本上得到了保障，壓力也不會隨着數據增加而增加。

4. Distributed Table & Distributed Engine

ClickHouse分布式表的本質並不是一張表, 而是一些本地物理表(分片)的分布式視圖,本身並不存儲數據. 分布式表建表的引擎為Distributed.

Distrbuted_table

CREATE TABLE IF NOT EXISTS {distributed_table} as {local_table} 
ENGINE = Distributed({cluster}, '{local_database}', '{local_table}', rand())

Distributed引擎需要以下幾個參數：

• 集群標識符
• 本地表所在的數據庫名稱
• 本地表名稱
• 分片鍵(sharding key) - 可選
  該鍵與config.xml中配置的分片權重(weight)一同決定寫入分布式表時的路由, 即數據最終落到哪個物理表上. 它可以是表中一列的原始數據(如site_id), 也可以是函數調用的結果, 如上面的SQL語句采用了隨機值rand(). 注意該鍵要盡量保證數據均勻分布, 另外一個常用的操作是采用區分度較高的列的哈希值, 如intHash64(user_id).

4.1. 數據查詢的流程

1. 各個實例之間會交換自己持有的分片的表數據
2. 匯總到同一個實例上返回給用戶

參考

Clickhouse Overview

ClickHouse復制表、分布式表機制與使用方法

最快開源 OLAP 引擎！ ClickHouse 在頭條的技術演進