Clickhouse作為一個列式存儲分析型數(shù)據(jù)庫，提供了很多集成其他組件的表引擎數(shù)據(jù)同步方案。

官網(wǎng)介紹

?一 Kafka 表引擎

使用Clickhouse集成的Kafka表引擎消費Kafka寫入Clickhouse表中。?

1.1 流程圖

1.2 建表

根據(jù)上面的流程圖需要建立三張表，分別Clickhouse存儲數(shù)據(jù)表、Kafka數(shù)據(jù)消費引擎表、物化視圖。

（1）Clickhouse存儲數(shù)據(jù)表

create table if not exists my_test (`id` Int64 comment '主鍵ID',`name` String comment '名稱',`create_time` DateTime comment '創(chuàng)建時間'
)ENGINE = MergeTree()
PARTITION BY toInt64(toYYYYMMDD(create_time))
PRIMARY KEY id
ORDER BY (id, create_time)
SETTINGS index_granularity = 8192;

（2）Kafka數(shù)據(jù)消費引擎表

create table if not exists kafka_my_test (`id` Int64 comment '主鍵ID',`name` String comment '名稱',`create_time` DateTime comment '創(chuàng)建時間'
)ENGINE = Kafka()
SETTINGS
kafka_broker_list = '127.0.0.1:9092',
kafka_topic_list = 'topic_m_test',
kafka_group_name = 'group_id_test',
kafka_format = 'JSONEachRow';

必要參數(shù)：

• kafka_broker_list?– 以逗號分隔的 brokers 列表 (localhost:9092)。
• kafka_topic_list?– topic 列表 (my_topic)。
• kafka_group_name?– Kafka 消費組名稱 (group1)。如果不希望消息在集群中重復，請在每個分片中使用相同的組名。
• kafka_format?– 消息體格式。使用與 SQL 部分的?FORMAT?函數(shù)相同表示方法，例如?JSONEachRow。了解詳細信息，請參考?Formats?部分。

可選參數(shù)：

• kafka_row_delimiter?- 每個消息體（記錄）之間的分隔符。
• kafka_schema?– 如果解析格式需要一個 schema 時，此參數(shù)必填。例如，普羅托船長?需要 schema 文件路徑以及根對象?schema.capnp:Message?的名字。
• kafka_num_consumers?– 單個表的消費者數(shù)量。默認值是：1，如果一個消費者的吞吐量不足，則指定更多的消費者。消費者的總數(shù)不應該超過 topic 中分區(qū)的數(shù)量，因為每個分區(qū)只能分配一個消費者。

（3）物化視圖

CREATE MATERIALIZED VIEW IF NOT EXISTS view_m_test TO m_test AS SELECT id, name, create_time FROM kafka_m_test;

?1.3 數(shù)據(jù)模擬

#使用命令生產(chǎn)數(shù)據(jù)
bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic topic_m_test#發(fā)送一下數(shù)據(jù)
{"id": 12345666,"name":"test","age":12,"create_time":"2024-04-05 12:23:34"}#查詢
select * from m_test limit 10;

官網(wǎng)介紹?