本文介绍如何从消息队列Kafka版中同步数据至云数据库ClickHouse。
注意事项
- 消息队列Kafka版为付费服务。计费详情,请参见计费概述。
- 请确保数据源消息队列Kafka版实例和目的云数据库ClickHouse集群在同一地域并使用相同的VPC。
前提条件
已创建消息队列Kafka版的Topic和Group。如何创建,请参见创建资源。
操作步骤
- 登录云数据库ClickHouse控制台。
- 在集群列表页面,单击目标集群ID。
- 单击右上方导航栏的登录数据库。
- 在登录实例页面,输入数据库账号和密码,单击登录。
- 创建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。 - 创建云数据库ClickHouse目的表。
- 创建本地表。
create table default.kafka_table_local ON CLUSTER default ( id Int32, name String ) ENGINE = MergeTree() ORDER BY (id);
- 创建分布式表。说明 如果您只需要同步数据至本地表,可跳过此步骤。
CREATE TABLE kafka_table_distributed ON CLUSTER default AS default.kafka_table_local ENGINE = Distributed(default, default, kafka_table_local, id);
- 创建本地表。
- 创建VIEW将Kafka消费表消费到的数据同步到云数据库ClickHouse目的表。如果您同步的目的表是本地表,请将分布式表名更换为本地表名,再进行同步。
CREATE MATERIALIZED VIEW consumer ON CLUSTER default TO kafka_table_distributed AS SELECT * FROM kafka_src_table;
- 在消息队列Kafka版的Topic端发送消息。本文以发送消息
1,a
和2,b
为例。- 登录消息队列kafka控制台。
- 在实例列表页面,单击目标实例名称。
- 在Topic管理页面,单击目标Topic操作列的更多 > 体验发送消息。
- 在快速体验消息收发页面,输入发送的消息内容。
- 单击确定。
- 查询云数据库ClickHouse目的表,查询语句如下。
select * from kafka_table_distributed;
说明 如果您同步的目的表是本地表,请将查询语句中的分布式表名更换为本地表名,再进行查询。查询结果如下。┌─id─┬─name─┐ │ 1 │ a │ │ 2 │ b │ └───┴────┘