本篇文章轉(zhuǎn)自：https://blog.csdn.net/weixin_41461992/article/details/106790507

起因

由于需要做各種數(shù)據(jù)庫(kù)擺渡到kafka的組件研究。
其中clickhouse和kafka間的數(shù)據(jù)擺渡，根據(jù)官方給出的kafka引擎文檔，便有了我這篇實(shí)踐記錄。
相應(yīng)的，該配置也非常簡(jiǎn)單。

官方傳送門： kafka engine clickhouse

這邊對(duì)數(shù)據(jù)庫(kù)和kafka環(huán)境不再累述。

一、開發(fā)環(huán)境

kafka 2.4
zookeeper 3.4.5
clickhouse 20.4.5.36
centos7

二、 介紹

clickhouse支持kafka的表雙向同步，其中提供的為Kafka引擎。

其大致情況為如下情況：Kafka主題中存在對(duì)應(yīng)的數(shù)據(jù)格式，Clickhouse創(chuàng)建一個(gè)Kafka引擎表（即相當(dāng)于一個(gè)消費(fèi)者），當(dāng)主題有消息進(jìn)入時(shí)，獲取該消息，將其進(jìn)行消費(fèi)，然后物化視圖同步插入到MergeTree表中。

該引擎還支持反向?qū)懭氲終afka中，即往Kafka引擎表中插入數(shù)據(jù)，可以同步到Kafka中（同樣可以使用物化視圖將不同引擎需要的表數(shù)據(jù)同步插入到Kafka引擎表中）。

下面為Kafka Engine的一些配置：
老版本格式為：

Kafka(kafka_broker_list, kafka_topic_list, kafka_group_name, kafka_format[, kafka_row_delimiter, kafka_schema, kafka_num_consumers])

新版本格式為：

Kafka SETTINGSkafka_broker_list = 'localhost:9092',kafka_topic_list = 'topic1,topic2',kafka_group_name = 'group1',kafka_format = 'JSONEachRow',kafka_row_delimiter = '\n',kafka_schema = '',kafka_num_consumers = 2

必填參數(shù)（例如topic、kafka集群、消費(fèi)者組等）：

kafka_broker_list – 以逗號(hào)分隔的 brokers 列表 (例如kafka1:9092,kafka2:9092,kafka3:9092)。

kafka_topic_list – topic 列表 (你的topic名字，也可以多個(gè))。

kafka_group_name – Kafka 消費(fèi)組名稱 (group1)。如果不希望消息在集群中重復(fù)，請(qǐng)?jiān)诿總€(gè)分片中使用相同的組名。

kafka_format – 消息體格式。使用與 SQL 部分的 FORMAT 函數(shù)相同表示方法，例如 JSONEachRow、CSV、XML等。Formats格式傳送門

非必填的參數(shù)：

kafka_row_delimiter - 每個(gè)消息體（記錄）之間的分隔符。

kafka_schema – 如果解析格式需要一個(gè) schema 時(shí)，此參數(shù)必填。

kafka_num_consumers – 單個(gè)表的消費(fèi)者數(shù)量。默認(rèn)值是：1，如果一個(gè)消費(fèi)者的吞吐量不足，則指定更多的消費(fèi)者。眾所周知消費(fèi)者的總數(shù)不應(yīng)該超過(guò) topic 中分區(qū)的數(shù)量，因?yàn)槊總€(gè)分區(qū)只能分配一個(gè)消費(fèi)者。

三、實(shí)踐

創(chuàng)建，使用引擎創(chuàng)建一個(gè)Kafka消費(fèi)者并作為一條數(shù)據(jù)流。

CREATE TABLE queue (q_date Date,level String,message String) ENGINE = Kafka SETTINGS kafka_broker_list = 'k1:9092,k2:9092,k3:9092',kafka_topic_list = 'my_topic',kafka_group_name = 'kafka_group_test',kafka_format = 'CSV',kafka_num_consumers = 4;

消費(fèi)的消息會(huì)被自動(dòng)追蹤，因此每個(gè)消息在不同的消費(fèi)組里只會(huì)記錄一次。如果希望獲得兩次數(shù)據(jù)，則使用另一個(gè)組名創(chuàng)建副本。

消費(fèi)組可以靈活配置并且在集群之間同步。例如，如果集群中有10個(gè)主題和5個(gè)表副本，則每個(gè)副本將獲得2個(gè)主題。如果副本數(shù)量發(fā)生變化，主題將自動(dòng)在副本中重新分配。

創(chuàng)建一個(gè)結(jié)構(gòu)表

CREATE TABLE daily (day Date,level String,message String ) ENGINE = MergeTree(day, (day, level), 8192);

創(chuàng)建物化視圖，該視圖會(huì)在后臺(tái)轉(zhuǎn)換引擎中的數(shù)據(jù)并將其放入之前創(chuàng)建的表中。

CREATE MATERIALIZED VIEW consumer TO dailyAS SELECT q_date AS day, level, messageFROM queue;

其中AS后面的語(yǔ)句是自己根據(jù)實(shí)際需求進(jìn)行調(diào)整的。
為了提高性能，接受的消息被分組為max_insert_block_size大小的塊。如果未在stream_flush_interval_ms毫秒內(nèi)形成塊，則不關(guān)心塊的完整性，都會(huì)將數(shù)據(jù)刷新到表中。

停止或者修改轉(zhuǎn)換邏輯
detach物化視圖

DETACH TABLE consumer; ATTACH TABLE consumer;

在創(chuàng)建表前，需要在kafka創(chuàng)建一個(gè)topic用于測(cè)試。

./bin/kafka-topic.sh --create --topic my_topic --partitions 3 --replication-factor 3

此處的topic需要與之前的kafka_topic_list對(duì)應(yīng)，既然該參數(shù)為list，則可以配置多個(gè)topic

生產(chǎn)者往topic中生產(chǎn)數(shù)據(jù)

./bin/kafka-console-producer.sh --topic my_topic --broker-list k1:9092

按照CSV格式，生產(chǎn)數(shù)據(jù)輸入

2020-06-28 level1 message 2020-06-28 level2 message 2020-06-28 level3 message

此時(shí)查看表daily,數(shù)據(jù)已同步

select * from daily |--- day --- level --- message ---| |--- 2020-06-28 --- level1 --- message ---| |--- 2020-06-28 --- level2 --- message ---| |--- 2020-06-28 --- level3 --- message ---|

同理，向kafka引擎表中插入數(shù)據(jù)，也可以在my_topic中可以消費(fèi)到插入數(shù)據(jù)。

數(shù)據(jù)量測(cè)試
在該版本，我在kafka生產(chǎn)者客戶端進(jìn)行for循環(huán)2000w數(shù)據(jù)進(jìn)行測(cè)試，延遲不高，基本在10秒內(nèi)同步完成，不知道在表字段數(shù)量和復(fù)雜sql語(yǔ)句時(shí)情況如何。還需要進(jìn)一步進(jìn)行實(shí)際使用

所遇問(wèn)題

表結(jié)構(gòu)變更
對(duì)于數(shù)據(jù)同步問(wèn)題，其中一個(gè)就是表同步之間的結(jié)構(gòu)對(duì)應(yīng)問(wèn)題。
由于表創(chuàng)建的時(shí)候，已經(jīng)固定，所以ck的kafka引擎在遇到字段改變的時(shí)候，依然需要?jiǎng)h表重建，或者修改物化視圖進(jìn)行不同的sql操作。

延遲問(wèn)題
在之前的版本中，社區(qū)有人提出該同步延遲太高，特別是數(shù)據(jù)量大的時(shí)候，但是在我實(shí)際測(cè)試中，大約2000w簡(jiǎn)單單表同步延遲可以接受。具體性能需要進(jìn)一步測(cè)試。

format格式
對(duì)于自己規(guī)定的格式，一定要正確，比如csv就是csv，json就是json格式，不然會(huì)報(bào)錯(cuò)。

消費(fèi)問(wèn)題
前面說(shuō)過(guò)，kafka引擎其實(shí)是一個(gè)或者多個(gè)消費(fèi)者進(jìn)行topic的消費(fèi)，那必然就涉及到消費(fèi)問(wèn)題，如何重新消費(fèi)，如何在需要修改業(yè)務(wù)的時(shí)候重新連接。重置偏移量。
場(chǎng)景：丟失數(shù)據(jù)，重新消費(fèi)

首先，我們可以將daily中的消息干掉，手動(dòng)導(dǎo)致消息丟失（自己整一個(gè)消息丟失）TRUNCATE TABLE daily此時(shí)，daily數(shù)據(jù)被刪除了，同步的數(shù)據(jù)丟失。 接下來(lái)，我們停止kafka引擎表，在clickhouse中執(zhí)行DETACH TABLE queue最后在kafka中，執(zhí)行重置偏移量的命令./bin/kafka-consumer-groups --bootstrap-server k1:9092 --topic my_topic --group kafka_group_test --reset-offsets --to-earliest --excute然后，重連queue表ATTACH TABLE queue

這樣，就開始重新消費(fèi)啦

錯(cuò)誤日志
clickhouse日志查看為ck目錄下的log文件夾的clickhouse-server.log中。

與50位技術(shù)專家面對(duì)面20年技術(shù)見證，附贈(zèng)技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的Clickhouse Engine kafka 将kafka数据同步clickhouse的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。