Clickhouse Engine kafka 将kafka数据同步clickhouse

清疚 2023-02-18 08:15 239阅读 0赞

## 起因 ##

由于需要做各种数据库摆渡到kafka的组件研究。  
其中clickhouse和kafka间的数据摆渡，根据官方给出的kafka引擎文档，便有了我这篇实践记录。  
相应的，该配置也非常简单。

官方传送门： [kafka engine clickhouse][]

这边对数据库和kafka环境不再累述。

## 一、开发环境 ##

kafka 2.4  
zookeeper 3.4.5  
clickhouse 20.4.5.36  
centos7

## 二、 介绍 ##

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

其大致情况为如下情况：Kafka主题中存在对应的数据格式，Clickhouse创建一个Kafka引擎表（即相当于一个消费者），当主题有消息进入时，获取该消息，将其进行消费，然后物化视图同步插入到MergeTree表中。

![Kafka同步到CK流程][Kafka_CK]  
该引擎还支持反向写入到Kafka中，即往Kafka引擎表中插入数据，可以同步到Kafka中（同样可以使用物化视图将不同引擎需要的表数据同步插入到Kafka引擎表中）。

![反向同步到Kafka][Kafka]  
**下面为Kafka Engine的一些配置：**

老版本格式为：

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

新版本格式为：

Kafka SETTINGS
      kafka_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

**必填参数（例如topic、kafka集群、消费者组等）：**

1.  kafka\_broker\_list – 以逗号分隔的 brokers 列表 (例如kafka1:9092,kafka2:9092,kafka3:9092)。
2.  kafka\_topic\_list – topic 列表 (你的topic名字，也可以多个)。
3.  kafka\_group\_name – Kafka 消费组名称 (group1)。如果不希望消息在集群中重复，请在每个分片中使用相同的组名。
4.  kafka\_format – 消息体格式。使用与 SQL 部分的 FORMAT 函数相同表示方法，例如 JSONEachRow、CSV、XML等。[Formats格式传送门][Formats]

**非必填的参数：**

1.  kafka\_row\_delimiter - 每个消息体（记录）之间的分隔符。
2.  kafka\_schema – 如果解析格式需要一个 schema 时，此参数必填。
3.  kafka\_num\_consumers – 单个表的消费者数量。默认值是：1，如果一个消费者的吞吐量不足，则指定更多的消费者。众所周知消费者的总数不应该超过 topic 中分区的数量，因为每个分区只能分配一个消费者。

## 三、 实践 ##

1.  创建，使用引擎创建一个Kafka消费者并作为一条数据流

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;

消费的消息会被自动追踪，因此每个消息在不同的消费组里只会记录一次。如果希望获得两次数据，则使用另一个组名创建副本。  
消费组可以灵活配置并且在集群之间同步。例如，如果群集中有 10 个主题和 5 个表副本，则每个副本将获得 2 个主题。 如果副本数量发生变化，主题将自动在副本中重新分配。

1.  创建一个结构表

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

1.  创建物化视图，该视图会在后台转换引擎中的数据并将其放入之前创建的表中。

CREATE MATERIALIZED VIEW consumer TO daily
        AS SELECT q_date AS day, level, message
        FROM queue;

其中AS后面的语句是自己根据实际需求进行调整的。  
为了提高性能，接受的消息被分组为 max\_insert\_block\_size 大小的块。如果未在 stream\_flush\_interval\_ms 毫秒内形成块，则不关心块的完整性，都会将数据刷新到表中。

1.  停止或者修改转换逻辑  
    detach物化视图

DETACH TABLE consumer;
      ATTACH TABLE consumer;

1.  在创建表前，需要在kafka创建一个topic用于测试

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

此处的topic需要与之前的 kafka\_topic\_list对应，既然该参数为list，则可以配置多个topic

1.  生产者往topic中生产数据

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

按照CSV格式，生产数据输入

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

此时查看表daily，数据已同步

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

同理，向kafka引擎表中插入数据，也可以在my\_topic中可以消费到插入数据。

1.  数据量测试  
    在该版本，我在kafka生产者客户端进行for循环2000w数据进行测试，延迟不高，基本在10秒内同步完成，不知道在表字段数量和复杂sql语句时情况如何。还需要进一步进行实际使用

# 所遇问题 #

1.  表结构变更  
    对于数据同步问题，其中一个就是表同步之间的结构对应问题。  
    由于表创建的时候，已经固定，所以ck的kafka引擎在遇到字段改变的时候，依然需要删表重建，或者修改物化视图进行不同的sql操作。
2.  延迟问题  
    在之前的版本中，社区有人提出该同步延迟太高，特别是数据量大的时候，但是在我实际测试中，大约2000w简单单表同步延迟可以接受。具体性能需要进一步测试。
3.  format格式  
    对于自己规定的格式，一定要正确，比如csv就是csv，json就是json格式，不然会报错。
4.  消费问题  
    前面说过，kafka引擎其实是一个或者多个消费者进行topic的消费，那必然就涉及到消费问题，如何重新消费，如何在需要修改业务的时候重新连接。重置偏移量。  
    场景：丢失数据，重新消费：

首先，我们可以将daily中的消息干掉，手动导致消息丢失（自己整一个消息丢失）
    
    TRUNCATE TABLE daily
    
    此时，daily数据被删除了，同步的数据丢失。
    接下来，我们停止kafka引擎表，在clickhouse中执行
    
    DETACH TABLE queue
    
    最后在kafka中，执行重置偏移量的命令
    
    ./bin/kafka-consumer-groups --bootstrap-server k1:9092 --topic my_topic --group kafka_group_test --reset-offsets --to-earliest --excute
    
    然后，重连queue表
    
    ATTACH TABLE queue

这样，就开始重新消费啦

1.  错误日志  
    clickhouse日志查看为ck目录下的log文件夹的clickhouse-server.log中。

[kafka engine clickhouse]: https://clickhouse.tech/docs/zh/engines/table-engines/integrations/kafka/
[Kafka_CK]: https://img-blog.csdnimg.cn/20200617171930828.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg==,size_16,color_FFFFFF,t_70
[Kafka]: https://img-blog.csdnimg.cn/20200617172452963.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg==,size_16,color_FFFFFF,t_70
[Formats]: https://clickhouse.tech/docs/en/interfaces/formats/