clickhouse 九（同步kafka数据）

小灰灰 2021-05-10 11:18 714阅读 0赞

## 1 步骤 ##

kafka作为消息队列通常用来收集各个服务产生的数据，而下游各种数据服务订阅消费数据，本文通过使用clickhouse 自带的kafka 引擎，来同步消费数据。

同步步骤：

*  kafka中创建topic，创建消费者并消费该topic（查看消费情况）
 *  建立目标表(通常是MergeTree引擎系列），用来存储kafka中的数据；
 *  建立kafka引擎表，用于接入kafka数据源；
 *  创建Materialized View（物化视图）, 监听kafka中的数据并将数据同步到clickhouse的目标表中；

同步流程图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTAyNTM2Mg_size_16_color_FFFFFF_t_70]

## 2 创建测试数据源 ##

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafka-reader
    
    # 创建消费者指定topic
    ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning  --topic kafka-reader --group kafka-reader-group

## 3 创建数据储存目标表 ##

CREATE TABLE target(
    day Date,
    level String,
    message String
    ) ENGINE = SummingMergeTree(day, (day, level), 8192);

## 4 创建kafka消费表 ##

1 使用kafka引擎创建queue表来连接kafka并读取topic中的数据。该数据表订阅了名为kafka-reader的消息主题，且消费组的名称为kafka-reader-group，⽽消息的格式采⽤了JSONEachRow。  
2 在此之后，查询这张数据表就能够看到Kafka的数据了。但是再次查询这张便就会没有数据了，这是因为Kafka表引擎在执⾏查询之后就会删除表内的数据。

CREATE TABLE queue (
        timestamp DateTime,
        level String,
        message String
      )
      ENGINE = Kafka
    SETTINGS kafka_broker_list = '192.168.9.226:9092',
           kafka_topic_list = 'kafka-reader',
    	   kafka_row_delimiter = '\n',
           kafka_group_name = 'kafka-reader-group',
           kafka_format = 'JSONEachRow'

参数解析–必要参数：

*  kafka\_broker\_list – 以逗号分隔的kafka的brokers 列表 (192.168.9.226:9092)。
 *  kafka\_topic\_list – topic 列表 (kafka-reader)。
 *  kafka\_group\_name – Kafka 消费组名称 (kafka-reader-group)。如果不希望消息在集群中重复，请在每个分片中使用相同的组名。
 *  kafka\_format – 消息体格式。JSONEachRow也就是普通的json格式，使用与 SQL 部分的 FORMAT 函数相同表示方法。

参数解析–可选参数：

*  kafka\_row\_delimiter - 每个消息体之间的分隔符。
 *  kafka\_schema – 如果解析格式需要一个 schema 时，此参数必填。例如，普罗托船长 需要 schema - 文件路径以及根对象 schema.capnp:Message 的名字。
 *  kafka\_num\_consumers – 单个表的消费者数量。默认值是：1，如果一个消费者的吞吐量不足，则指定更多的消费者。消费者的总数不应该超过 topic 中分区的数量，因为每个分区只能分配一个消费者。

## 5 创建Materialized View（物化视图）传输数据 ##

创建好的物化视图，它将会在后台收集数据。可以持续不断地从 Kafka 收集数据并通过 SELECT 将数据转换为所需要的格式。

CREATE MATERIALIZED VIEW consumer TO target
     AS SELECT toDate(toDateTime(timestamp)) AS day, level,message
     FROM queue;

## 6 测试 ##

生产者添加数据：  
![在这里插入图片描述][20201230164205491.png]  
查询目标表，查看消费数据

SELECT *
    FROM target
    
    ┌────────day─┬─level─┬─message─┐
    │ 2020-12-01 │ 11    │ 不开心   │
    │ 2020-12-30 │ 13    │ 写博客   │
    │ 2020-12-31 │ 15    │ 买可乐   │
    │ 2020-12-31 │ 17    │ 真好喝   │
    └────────────┴───────┴─────────┘

查询consumer物化视图表，一般得到的数据和目标表差不多，除非实时数据很多，停止接收topic数据或更改转换逻辑需要停用物化视图，更改完之后再启用物化视图

# 停用
     DETACH TABLE consumer;
    # 启用  
     ATTACH TABLE consumer;

## 总结 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTAyNTM2Mg_size_16_color_FFFFFF_t_70 1]

简单来说，数据的流向如上图，Clickhouse消费kafka数据的过程中，通过kafka引擎表作为一个管道接收流入的数据，而物化视图负责将kafka引擎表的数据实时同步到目标表中，我们通过不同sql语句封装将kafka数据导入到不同目标表中，这是不错消费方式。

创作不易，喜欢的话可以点赞加关注哦！阿里嘎多民那桑！

参考文章： [clickhouse官方文档][clickhouse]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTAyNTM2Mg_size_16_color_FFFFFF_t_70]: /images/20210505/684c90e12acb4a70b906ce41b14aa2f2.png
[20201230164205491.png]: /images/20210505/44414f97b30940e8bba18b226047cc76.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTAyNTM2Mg_size_16_color_FFFFFF_t_70 1]: /images/20210505/503af94470e141c89d24d1562cd22d7a.png
[clickhouse]: https://clickhouse.tech/docs/zh/