本文介绍如何从消息队列Kafka版中同步数据至云数据库ClickHouse

注意事项

  • 消息队列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,a2,b为例。
    1. 登录消息队列kafka控制台
    2. 实例列表页面,单击目标实例名称。
    3. Topic管理页面,单击目标Topic操作列的更多 > 体验发送消息
    4. 快速体验消息收发页面,输入发送的消息内容
    5. 单击确定
  9. 查询云数据库ClickHouse目的表,查询语句如下。
    select * from kafka_table_distributed; 
    说明 如果您同步的目的表是本地表,请将查询语句中的分布式表名更换为本地表名,再进行查询。
    查询结果如下。
    ┌─id─┬─name─┐
    │  1   │  a     │  
    │  2   │  b     │ 
    └───┴────┘