从Kafka同步

注意事项

• 消息队列Kafka版为付费服务。计费详情，请参见计费概述。
• 请确保数据源消息队列Kafka版实例和目的云数据库ClickHouse集群在同一地域并使用相同的VPC。

前提条件

已创建消息队列Kafka版的Topic和Group。如何创建，请参见创建资源。

操作步骤

1. 登录云数据库ClickHouse控制台。
2. 在集群列表页面，单击目标集群ID。
3. 单击右上方导航栏的登录数据库。
4. 在登录实例页面，输入数据库账号和密码，单击登录。
5. 创建Kafka消费表，示例语句如下。

   CREATE TABLE default.kafka_src_table ON CLUSTER default
   (   //定义表结构的字段
       id Int32,
       name String               
   ) ENGINE = Kafka()
   SETTINGS
       kafka_broker_list = 'alikafka-post-cn-tl32i5sc****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-tl32i5sc****-2-vpc.alikafka.aliyuncs.com:9092',
       kafka_topic_list = 'test',
       kafka_group_name = 'test',
       kafka_format = 'CSV',
       kafka_num_consumers = 1,
       kafka_max_block_size = 65536,
       kafka_skip_broken_messages = 0,
       kafka_auto_offset_reset = 'latest';

   说明

   • Kafka消费表不能直接作为结果表使用。
   • Kafka消费表只是用来消费Kafka数据，没有真正的存储数据。

   常用参数说明如下。

   名称	是否必选	说明
   kafka_broker_list	是	以英文逗号分隔的Kafka的接入点地址列表。如何查看接入点，请参见查看接入点。
   kafka_topic_list	是	以英文逗号分隔的Topic名称列表。
   kafka_group_name	是	Kafka的消费组名称。更多信息，请参见创建Group。
   kafka_format	是	云数据库ClickHouse支持处理的消息体格式。 说明 云数据库ClickHouse支持的消息体格式，具体请参见Formats for Input and Output Data。
   kafka_row_delimiter	否	行分隔符，用于分割不同的数据行。默认为“\n”，您也可以根据数据写入的实际分割格式进行设置。
   kafka_num_consumers	否	单个表的消费者数量，默认值为1。 说明 一个消费者的吞吐量不足时，需要指定更多的消费者。 消费者的总数不应超过Topic中的分区数，因为每个分区只能分配一个消费者。
   kafka_max_block_size	否	Kafka消息的最大批次大小，单位：Byte，默认值为65536 Byte。
   kafka_skip_broken_messages	否	kafka消息解析器对于脏数据的容忍度，默认值为0。如果kafka_skip_broken_messages=N，则引擎将跳过N条无法解析的Kafka消息（一条消息等于一行数据）。
   kafka_commit_every_batch	否	执行Kafka commit的频率，取值说明如下。 0：完全写入一整个Block数据块的数据后才执行commit。 1：每写完一个Batch批次的数据就执行一次commit。
   kafka_ auto_offset_reset	否	从哪个offset开始读取Kafka数据，取值说明如下。 earliest：从最早的offset开始读取Kafka数据。 latest：从最晚的offset开始读取Kafka数据。 说明 21.8版本的云数据库ClickHouse集群不支持该参数。

   说明 更多参数说明，请参见Kafka。
6. 创建云数据库ClickHouse目的表。
   1. 创建本地表。

      create table default.kafka_table_local ON CLUSTER default (
        id Int32,
        name String
      ) ENGINE = MergeTree()
          ORDER BY (id);

   2. 创建分布式表。
      说明 如果您只需要同步数据至本地表，可跳过此步骤。

      CREATE TABLE kafka_table_distributed ON CLUSTER default AS default.kafka_table_local
      ENGINE = Distributed(default, default, kafka_table_local, id);

7. 创建VIEW将Kafka消费表消费到的数据同步到云数据库ClickHouse目的表。
   如果您同步的目的表是本地表，请将分布式表名更换为本地表名，再进行同步。

   CREATE MATERIALIZED VIEW consumer ON CLUSTER default TO kafka_table_distributed AS SELECT * FROM kafka_src_table;

8. 在消息队列Kafka版的Topic端发送消息。
   本文以发送消息1,a和2,b为例。
   1. 登录消息队列kafka控制台。
   2. 在实例列表页面，单击目标实例名称。
   3. 在Topic管理页面，单击目标Topic操作列的更多 > 体验发送消息。
   4. 在快速体验消息收发页面，输入发送的消息内容。
   5. 单击确定。
9. 查询云数据库ClickHouse目的表，查询语句如下。

   select * from kafka_table_distributed; 

   说明 如果您同步的目的表是本地表，请将查询语句中的分布式表名更换为本地表名，再进行查询。

   查询结果如下。

   ┌─id─┬─name─┐
   │  1   │  a     │  
   │  2   │  b     │ 
   └───┴────┘