Sync data from Kafka

Limitations

You can sync data only from Message Queue for Apache Kafka instances and self-managed Kafka clusters that are deployed on ECS instances.

Prerequisites

• Alibaba Cloud ClickHouse:

  • A destination cluster is created. The cluster and the Message Queue for Apache Kafka instance must be in the same region and use the same VPC. For more information, see Create a cluster.

  • A database account is created for the destination cluster and has permissions to perform operations on the database. For more information, see Account management.

• Message Queue for Apache Kafka:

  • A topic is created.

  • A consumer group is created.

Usage notes

• The topic to which an Alibaba Cloud ClickHouse Kafka external table subscribes cannot have other consumers.

• When you create a Kafka external table, a materialized view, and a local table, ensure that their field types match.

Procedure

This topic provides an example of how to sync data from Message Queue for Apache Kafka to a distributed table named kafka_table_distributed in the default database of an Alibaba Cloud ClickHouse Community-compatible Edition cluster.

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_consumer = 1,]
    [kafka_max_block_size = 0,]
    [kafka_skip_broken_messages = N,]
    [kafka_commit_every_batch = 0,]
    [kafka_auto_offset_reset = N]

CREATE TABLE default.kafka_src_table ON CLUSTER `default`
(-- Define the table structure fields
    id Int32,
    name String               
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'alikafka-post-cn-****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-****1-2-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-****-3-vpc.alikafka.aliyuncs.com:9092',
    kafka_topic_list = 'testforCK',
    kafka_group_name = 'GroupForTestCK',
    kafka_format = 'CSV';

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

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_local AS SELECT * FROM kafka_src_table;

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

Step 6: Verify the data sync

1. Send messages to the topic in Message Queue for Apache Kafka.

   1. Log on to the Message Queue for Apache Kafka console.

   2. On the Instances page, click the name of the destination instance.

   3. On the Topic Management page, find the destination topic and choose More > Experience Sending Message from the Actions column.

   4. In the Quickly Experience Message Sending and Receiving dialog box, enter the Message Content.

      This topic uses the messages 1,a and 2,b as examples.

   5. Click OK.

2. Log on to Alibaba Cloud ClickHouse and query the distributed table to check whether the data is synced.

   For information about how to log on to Alibaba Cloud ClickHouse, see Connect to a ClickHouse cluster by using DMS.

   Use the following query statement to verify the data:

   Enterprise Edition

   SELECT * FROM kafka_table_local; 

   Community-compatible Edition

   The following statement provides an example of how to query a distributed table.

   • If the destination table is a local table, you must replace the distributed table name with the local table name in the query.

   • If you use a multi-node Community-compatible Edition cluster, we strongly recommend that you query the distributed table. If you query a local table directly, you will retrieve data from only one node, which can result in an incomplete dataset.

   SELECT * FROM kafka_table_distributed; 

   If the query returns a result, the data has been successfully synced from Kafka to Alibaba Cloud ClickHouse.

   The query result is as follows:

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

   If the query returns an unexpected result, you can troubleshoot the issue by following the instructions in Step 7 (Optional): View the consumption status of the Kafka external table.

select * from system.kafka_consumers;

SELECT * FROM system.kafka;

FAQ

• Why can't I connect to external tables such as MySQL, HDFS, and Kafka?

• Why does the number of rows in ClickHouse not match the number of rows imported from Kafka?

• Why does the data in a Kafka external table not increase after the table is created?