設(shè)計理念
Everything is table（萬物皆表），數(shù)據(jù)表就是ClickHouse和外部交互的接口弧满。在數(shù)據(jù)表背后無論連接的是本地文件、HDFS解虱、Zookeeper還是其它服務(wù)隙咸，終端用戶始終只需要面對數(shù)據(jù)表铃肯，只需要用SQL查詢語言威恼。
處理各種數(shù)據(jù)場景走內(nèi)部集成的路線竹祷，既與外部系統(tǒng)（其它數(shù)據(jù)庫擅编，消息中間件或者是服務(wù)接口）的集成直接在數(shù)據(jù)庫內(nèi)部實現(xiàn)攀细。
面向表編程，一張數(shù)據(jù)表最終能夠提供哪些功能爱态、擁有哪些特性谭贪、數(shù)據(jù)會議什么格式被保存以及數(shù)據(jù)會怎樣被加載，這些都將由它的表引擎決定锦担。
表引擎特性
| 表引擎 | 概念 | 引擎 |
|---|---|---|
| 合并樹類型 | MergeTree提供主鍵索引俭识，數(shù)據(jù)分區(qū)，數(shù)據(jù)副本和數(shù)據(jù)采樣等基本能力洞渔，其它表引擎在MergeTree上各有所長套媚；ReplacingMergeTree具有刪除重復(fù)數(shù)據(jù)的特性；SummingMergeTree會按照排序鍵自動聚合數(shù)據(jù)磁椒；Replicated系列支持?jǐn)?shù)據(jù)副本 | MergeTree,SummingMergeTree ... |
| 外部存儲類型 | 外部存儲表引擎直接從其他的存儲系統(tǒng)讀取數(shù)據(jù)堤瘤，例如直接讀取HDFS的文件或者MySQL數(shù)據(jù)庫的表。這些引擎只負責(zé)元數(shù)據(jù)管理和數(shù)據(jù)查詢浆熔，而它們自身通常并不負責(zé)數(shù)據(jù)的寫入本辐，數(shù)據(jù)文件直接由外部系統(tǒng)提供。 | HDFS,MySQL,JDBC,Kafka,File |
| 內(nèi)存類型 | 面向內(nèi)存查詢，數(shù)據(jù)從內(nèi)存中直接訪問慎皱。除了Memory引擎之外老虫，其它幾款都會將數(shù)據(jù)寫入磁盤以防數(shù)據(jù)丟失。在數(shù)據(jù)被加載時宝冕，數(shù)據(jù)會被全部加載到內(nèi)存张遭、以供查詢使用。全部加載到內(nèi)存：意味著有更好的查詢性能地梨；但是如果加載的數(shù)據(jù)量過大菊卷，就會有極大的內(nèi)存消耗和負擔(dān) | Memory,Set,Join,Buffer, |
| 日志類型 | 數(shù)據(jù)量很小（100萬以下）宝剖，同時查詢場景也比較簡單洁闰，并且是“一次”寫入多次查詢方式，可以使用日志類型万细；共性：不支持索引和分區(qū)等高級特性扑眉；不支持并發(fā)讀寫，當(dāng)針對一張日志寫入數(shù)據(jù)時赖钞，針對這張表的查詢會被阻塞腰素，直至寫入動作結(jié)束；擁有物理存儲雪营，數(shù)據(jù)會被保存到本地文件中 | TinyLog,StripeLog,Log |
| 接口類型 | 本身不保存數(shù)據(jù)弓千，而是像粘合劑一樣整合其它數(shù)據(jù)表；使用時献起，不用關(guān)心底層復(fù)雜性洋访，像接口一樣為用戶提供統(tǒng)一訪問頁面 | Merge,Dictionary,Distribute |
| 其它類型 | 擴充ClickHouse的能力邊界 | Live View,NULL,URL |
引擎使用場景
引擎使用場景
表引擎使用
SummingMergeTree（MergeTree家族）
使用
數(shù)據(jù)聚合
-
聚合前數(shù)據(jù)
聚合前數(shù)據(jù) -
聚合后數(shù)據(jù)
聚合后數(shù)據(jù)
使用說明
PRIMARY KEY可與ORDER BY不同，PRIMARY KEY作為主鍵索引谴餐，ORDER BY作為聚合條件姻政。
PRIMARY KEY是ORDER BY的前綴，索引和數(shù)據(jù)仍然具有對應(yīng)關(guān)系岂嗓；因為數(shù)據(jù)以O(shè)RDER BY排序汁展，索引以PRIMARY KEY排序，PRIMARY KEY是ORDER BY的前綴摄闸，那么索引也是ORDER BY有序的善镰，同一個排序標(biāo)準(zhǔn)，產(chǎn)生相同的數(shù)據(jù)順序年枕，所以炫欺，索引和數(shù)據(jù)仍然具有對應(yīng)關(guān)系。
ORDER BY可以可以減少熏兄，GROUP BY(A, B, C, D) --> GROUP BY(A, B); ORDER BY可以增加新的列品洛。
工作中使用
- SummingMergeTree聚合ClickHouse Shard中數(shù)據(jù)树姨，同時通過Distribute表向外提供服務(wù)
- PARTITION BY toStartOfHour(event_date) + PRIMARY KEY (event_date, network_id)
- ORDER BY (event_date, network_id, a, c, d) ...
- TTL event_date + INTERVAL 35 DAY DELETE
- SETTINGS index_granularity = 8192, replicated_deduplication_window = 0
原理
- 用ORDER BY排序鍵作為聚合數(shù)據(jù)的條件Key
- 只有在合并分區(qū)的時候才會觸發(fā)匯總的邏輯
- 以數(shù)據(jù)分區(qū)為單位來聚合數(shù)據(jù)。當(dāng)分區(qū)合并時桥状，同一個數(shù)據(jù)分區(qū)內(nèi)聚合Key相同的數(shù)據(jù)會被合并匯總帽揪，而不同分區(qū)之間的數(shù)據(jù)不會被匯總
- 如果在定義引擎時指定了columns匯總列（非主鍵的數(shù)值類型字段），則SUM匯總這些列字段辅斟；如果未指定转晰，則聚合所有非主鍵的數(shù)值類型字段
- 在進行數(shù)據(jù)匯總時，因為分區(qū)內(nèi)的數(shù)據(jù)已經(jīng)基于ORDER BY排序士飒，所以能夠找到相鄰且擁有相同聚合Key的數(shù)據(jù)
- 在匯總數(shù)據(jù)時查邢，同一個分區(qū)內(nèi)，相同聚合Key的多行數(shù)據(jù)會合并成一行酵幕。其中扰藕，匯總字段進行SUM計算；對于那些非匯總地段芳撒，則會使用第一行數(shù)據(jù)的取值邓深。
- 支持嵌套結(jié)構(gòu)，但列字段名稱必須以Map后綴結(jié)尾笔刹。
Kafka（外部存儲類型）
使用
kafka環(huán)境準(zhǔn)備
生產(chǎn)消息
數(shù)據(jù)data.json：
{ "id": "A001", "city": "wuhan", "v1": 10, "v2": 20, "create_time": "2019-08-10 17:00:00" }
{ "id": "A001", "city": "wuhan", "v1": 20, "v2": 30, "create_time": "2019-08-20 17:00:00" }
{ "id": "A001", "city": "zhuhai", "v1": 20, "v2": 30, "create_time": "2019-08-10 17:00:00" }
{ "id": "A001", "city": "wuhan", "v1": 10, "v2": 20, "create_time": "2019-02-10 09:00:00" }
{ "id": "A002", "city": "wuhan", "v1": 60, "v2": 50, "create_time": "2019-10-10 17:00:00" }
生產(chǎn)
kafka-console-producer --topic test --bootstrap-server localhost:9092 < data.json
創(chuàng)建kafka表
CREATE TABLE test_kafka
(
`id` String,
`city` String,
`v1` UInt32,
`v2` Float64,
`create_time` DateTime
)
ENGINE = Kafka()
SETTINGS kafka_broker_list = '172.18.0.3:9092', kafka_topic_list = 'test', kafka_group_name = 'test', kafka_format = 'JSONEachRow', kafka_skip_broken_messages = 100
消費消息
SELECT * FROM test_kafka
查看kafka消費日志
cat /var/log/clickhouse-server/clickhouse-server.log | grep kafka
2021.06.30 16:23:03.749646 [ 75 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Debug> executeQuery: (from 127.0.0.1:34022) select * from default.test_kafka;
2021.06.30 16:23:03.750176 [ 75 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Debug> StorageKafka (test_kafka): Starting reading 1 streams
2021.06.30 16:23:07.262678 [ 91 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Trace> StorageKafka (test_kafka): Polled batch of 98 messages. Offsets position: [ test[0:98] ]
2021.06.30 16:23:07.282639 [ 91 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Warning> StorageKafka (test_kafka): Parsing of message (topic: test, partition: 0, offset: 77) return no rows.
2021.06.30 16:23:07.285524 [ 91 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Trace> StorageKafka (test_kafka): Polled offset 98 (topic: test, partition: 0)
2021.06.30 16:23:07.339373 [ 91 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Trace> StorageKafka (test_kafka): Committed offset 98 (topic: test, partition: 0)
原理
- 只負責(zé)元數(shù)據(jù)的管理和數(shù)據(jù)查詢芥备，不存儲數(shù)據(jù)（外部存儲引擎共性），支持從kafa消費消息舌菜，也可以向kafka中插入數(shù)據(jù)（Demo）
- 默認(rèn)情況下门躯，Kafka表引擎每隔500毫秒會拉取一次數(shù)據(jù)，時間由stream_poll_timeout_ms參數(shù)控制酷师，數(shù)據(jù)首先會被放入緩存，在時機成熟時染乌，緩存數(shù)據(jù)會被刷新到數(shù)據(jù)表
- 滿足下列條件之一山孔，觸發(fā)刷新動作：
a. 當(dāng)完成一個數(shù)據(jù)塊兒的寫入（數(shù)據(jù)塊兒的大小由kafka_max_block_size參數(shù)控制，默認(rèn)情況下65536)
b. 等待時間超過7500毫秒（stream_flush_interval_ms參數(shù)控制荷憋，默認(rèn)7500ms）
Join（內(nèi)存類型）
使用
CREATE TABLE join_tb1 (
id UInt8,
name String,
time DateTime
) ENGINE = Log
INSERT INTO TABLE join_tb1 VALUES(1, 'ClickHouse', '2019-05-01 12:00:00'),
(2, 'Spark', '2019-05-01 12:30:00'), (3, 'ElasticSearch', '2019-05-01 13:00:00');
CREATE TABLE id_join_tb1 (
id UInt8,
price UInt32,
time DateTime
) ENGINE = Join(ANY, LEFT, id)
INSERT INTO TABLE id_join_tb1 VALUES (1, 100, '2019-05-01 11:55:00'),
(1, 105, '2019-05-01 11:10:00'),
(2, 90, '2019-05-01 12:01:00'),
(3, 80, '2019-05-01 11:55:00'),
(5, 70, '2019-05-01 11:55:00'),
(6, 60, '2019-05-01 11:55:00');
SELECT id, name, price FROM join_tb1 LEFT JOIN id_join_tb1 USING(id);
SELECT joinGet('id_join_tb1', 'price', toUInt8(1));
原理
- ENGINE = Join(join_strictness, join_type, key1[, key2, ...])
- join_strictness: 連接的精度台颠，它決定了JOIN查詢在連接數(shù)據(jù)時所使用的策略，目前支持ALL勒庄，ANY和ASOF三種類型
- join_type: 連接左右兩個數(shù)據(jù)集合的策略串前；交集，并集实蔽，笛卡爾積或其他形式荡碾，目前支持INNER，OUTER和CROSS局装；當(dāng)join_type類型為ANY時坛吁，在數(shù)據(jù)寫入時劳殖，join_key重復(fù)的數(shù)據(jù)會被自動忽略
- join_key: 連接鍵，決定使用哪個列字段進行關(guān)聯(lián)
Merge（接口類型）
使用
# 數(shù)據(jù)以年分表拨脉，使用Merge引擎進行粘合
CREATE TABLE test_table_2018(
id String,
create_time DateTime,
code String
)ENGINE = MergeTree()
PARTITION BY toYYYYMM(create_time)
ORDER BY id
CREATE TABLE test_table_2019(
id String,
create_time DateTime,
code String
)ENGINE = Log()
PARTITION BY toYYYYMM(create_time)
ORDER BY id
CREATE TABLE test_table_all as test_table_2018
ENGINE = MergeTree(currentDatebase(), '^test_table_')
原理
- 不存儲數(shù)據(jù)哆姻，而是像粘合劑一樣可以整合其他的數(shù)據(jù)表
- 被代理查詢的數(shù)據(jù)表要在同一個數(shù)據(jù)庫內(nèi)，且擁有相同的表結(jié)構(gòu)玫膀，但是它們可以使用不同的表引擎以及不同的分區(qū)定義（對于MergeTree而言）
URL（其它類型）
使用
/* GET users listing. */
router.get('/users', func (req, res, next) {
var result = ''
for(let i=0; i<5; i++){
result += '{"name":"nauu'+i+'"}\n';
}
res.send(result)
})
/* POST user. */
router.post('/users', func (req, res) {
res.sendStatus(200)
})
CREATE TABLE url_table (
name String
)
ENGINE = URL('http://localhost:9688/users', JSONEachRow)
SELECT * FROM url_table
INSERT INTO TABLE url_table VALUES('nauu-insert')
原理
- URL表引擎等價于HTTP客戶端矛缨，它可以通過HTTP/HTTPS協(xié)議，直接訪問遠端的REST服務(wù)帖旨。
- SELECT查詢會被底層轉(zhuǎn)換為GET請求
- INSERT查詢會被轉(zhuǎn)換為POST請求
綜合使用例子
Kafka + MATERIALIZED VIEW + ReplicateSummingMergeTree + Distributed
Kafka Engine Table： 外部存儲表箕昭，消費kafka消息
Materialize View: 當(dāng)數(shù)據(jù)插入到kafka表時，執(zhí)行select語句將數(shù)據(jù)進行transform后碉就，插入到To表
SummingMergeTree: MergeTree家族表盟广，支持partition summing，主鍵索引瓮钥，數(shù)據(jù)分區(qū)筋量，replica和數(shù)據(jù)采樣；
Distribute 表： 進行數(shù)據(jù)粘合碉熄，為用戶提供統(tǒng)一的數(shù)據(jù)視圖
引用
Docker HDFS
Docker HDFS 2
apt install netcat
Docker中容器之間通訊方式
安裝ifconfig apt install net-tools
查看docker容器ip地址 docker inspect kafka-docker_clickhouse-server_1 | grep IP
Kafka引擎
ClickHouse原理解析與應(yīng)用實踐
