从Kafka同步数据

前提条件

• 已创建目标云数据库ClickHouse集群，且保证消息队列Kafka和目标云数据库ClickHouse集群在同一地域并使用相同的VPC。如何创建，请参见新建集群。

• 目标云数据库ClickHouse集群已创建登录数据库的账号且已具备数据库的操作权限。如何创建，请参见创建账号和准备权限。

操作步骤

1. 登录云数据库ClickHouse控制台。

2. 在页面左上角，选择目标集群所在的地域。

3. 在集群列表页面，选择目标集群对应类型的实例列表，单击目标集群ID。

4. 在集群信息页面，单击右上方导航栏的登录数据库。

5. 在DMS数据管理服务控制台的登录实例页面，输入数据库账号和密码，单击登录。

6. 创建Kafka消费表。

   说明

   • 创建Kafka消费表是为了将Kafka消费表的数据同步到云数据库ClickHouse的表中。

   • Kafka消费表不能直接使用。

   • Kafka消费表只是用来消费Kafka数据，没有真正地存储数据，最终消费数据需要实现消息批处理后进行消费。

   建表语法如下。

   CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
   (
       name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
       name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
       ...
   ) ENGINE = Kafka()
   SETTINGS
       kafka_broker_list = 'host:port1,host:port2,host:port3',
       kafka_topic_list = 'topic_name1,topic_name2,...',
       kafka_group_name = 'group_name',
       kafka_format = 'data_format'[,]
       [kafka_row_delimiter = 'delimiter_symbol',]
       [kafka_num_consumers = N,]
       [kafka_thread_per_consume = 1,]
       [kafka_max_block_size = 0,]
       [kafka_skip_broken_messages = N,]
       [kafka_commit_every_batch = 0,]
       [kafka_auto_offset_reset = N]

   常用参数说明如下。

   名称	是否必选	说明
   kafka_broker_list	是	以英文逗号（,）分隔的Kafka的接入点地址列表。如何查看接入点，请参见查看接入点。
   kafka_topic_list	是	以英文逗号（,）分隔的Topic名称列表。如何查看Topic名称，请参见创建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_thread_per_consumer	否	指定每个消费者是否启动独立线程进行消费，默认值为0。取值说明如下。 0：表示所有消费者共同使用1个线程消费。 1：表示每个消费者启动独立线程进行消费。 提升消费速度更多信息请参见Kafka性能调优。
   kafka_max_block_size	否	Kafka消息的最大批次大小，单位：Byte，默认值为65536。
   kafka_skip_broken_messages	否	Kafka消息解析器对于脏数据的容忍度，默认值为0。如果kafka_skip_broken_messages=N，则引擎将跳过N条无法解析的Kafka消息（一条消息等于一行数据）。
   kafka_commit_every_batch	否	执行Kafka Commit的频率，默认值为0，取值说明如下： 0：完全写入一整个Block数据块的数据后才执行Commit。 1：每写完一个Batch批次的数据就执行一次Commit。
   kafka_auto_offset_reset	否	消息的偏移量，从哪个offset开始读取Kafka数据，取值说明如下： earliest（默认值）：从最早的offset开始读取Kafka数据。 latest：从最晚的offset开始读取Kafka数据。 说明 21.8版本的云数据库ClickHouse集群不支持该参数。

   说明

   更多参数说明，请参见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,alikafka-post-cn-tl32i5sc****-3-vpc.alikafka.aliyuncs.com:9092',
       kafka_topic_list = 'test',
       kafka_group_name = 'test',
       kafka_format = 'CSV';

7. 创建云数据库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);

8. 创建视图将Kafka消费表的数据同步到云数据库ClickHouse的分布式表。

   说明

   如果您同步的目的表是本地表，请将分布式表名更换为本地表名，再进行同步。

   创建视图语法如下。

   CREATE MATERIALIZED VIEW [view.name] ON CLUSTER default TO [dest_table] AS SELECT * FROM [src_table];

   示例语句如下。

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

验证同步结果

您可以选择以下任意一种方式验证同步结果。

查询云数据库ClickHouse分布式表验证同步结果

1. 在云消息队列Kafka版的Topic端发送消息。

   1. 登录消息队列Kafka控制台。

   2. 在实例列表页面，单击目标实例名称。

   3. 在Topic管理页面，单击目标Topic操作列的更多 > 体验发送消息。

   4. 在快速体验消息收发页面，输入发送的消息内容。

      本文以发送消息1,a和2,b为例。

   5. 单击确定。

2. 查询云数据库ClickHouse分布式表，确认数据是否同步成功，查询语句如下。

   SELECT * FROM kafka_table_distributed; 

   说明

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

   查询结果如下。

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

   说明

   当您执行查询语句并成功返回结果时，说明数据已从Kafka同步至云数据库ClickHouse。

查询系统表验证同步结果

SELECT * FROM system.kafka

┌─database─┬──────────table──────────────┬─topic─┬─consumer_group─┬─last_read_message_count─┬───────status──────┬─exception─┐
│  default │  kafka_table_distributed    │ test  │   test         │          2              │   attach_view     │           │  
└──────────┴─────────────────────────────┴───────┴────────────────┴─────────────────────────┴───────────────────┴───────────┘

常见问题

从Kafka同步数据到云数据库ClickHouse的常见问题及处理方法，请参见常见问题。