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

前提条件

  • 已创建消息队列Kafka版实例以及创建资源Topic和Group。如何创建,请参见公网和VPC接入创建资源
  • 已创建目标云数据库ClickHouse集群和登录数据库的账号和密码。如何创建,请参见新建集群创建账号
  • 已具备登录数据库的操作权限。具体信息,请参见准备权限
  • 请确保数据源消息队列Kafka版实例和目标云数据库ClickHouse集群在同一地域并使用相同的VPC。

操作步骤

  1. 登录云数据库ClickHouse控制台
  2. 在页面左上角,选择目标集群所在的地域。
  3. 登录数据库。
    1. 集群列表页面,选择默认实例列表,单击目标集群ID。
    2. 集群信息页面,单击右上方导航栏的登录数据库
    3. 在DMS数据管理服务控制台的登录实例页面,输入数据库账号和密码,单击登录
  4. 创建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_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_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(默认值):完全写入一整个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';
  5. 创建云数据库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);
  6. 创建视图将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,a2,b为例。
    5. 单击确定
  2. 查询云数据库ClickHouse分布式表,确认数据是否同步成功,查询语句如下。
    Select * from kafka_table_distributed; 
    说明 如果您同步的目的表是本地表,请将查询语句中的分布式表名更换为本地表名,再进行查询。
    查询结果如下。
    ┌─id─┬─name─┐
    │  1 │  a   │  
    │  2 │  b   │ 
    └────┴──────┘
    说明 当您执行查询语句并成功返回结果时,说明数据已从Kafka同步至 云数据库ClickHouse

通过查询系统表验证同步结果

通过查询系统表 system.kafka查看kafka消费表的消费状态,查询语句如下。
Select * from system.kafka
查询结果如下。
┌─database─┬──────────table──────────────┬─topic─┬─consumer_group─┬─last_read_message_count─┬───────status──────┬─exception─┐
│  default │  kafka_table_distributed    │ test  │   test         │          2              │   attach_view     │           │  
└──────────┴─────────────────────────────┴───────┴────────────────┴─────────────────────────┴───────────────────┴───────────┘
查询结果说明如下。
名称 说明
database kafka消费表的数据库名称。
table kafka消费表的表名。
topic kafka消费表的topic名称。
consumer_group kafka消费表的group名称。
last_read_message_count 拉取到的kafka消费表的消息数量。
status kafka消费表的消费状态。取值说明:
  • no_view:kafka消费表没有创建视图。
  • attach_view:kafka消费表创建了视图。
  • normal:正常状态。
    说明 当kafka消费表有消费数据时,kafka消费表的消费状态为 normal
  • skip_parse:跳过错误解析。
  • error:消费异常。
exception 异常详情。
说明status取值为 error时,该参数返回异常详情。

常见问题

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