ApsaraDB for ClickHouse:Sync from Kafka

Limitations

Only the following Kafka sources can sync to ApsaraDB for ClickHouse:

• ApsaraMQ for Kafka instances

• Self-managed Kafka clusters deployed on Elastic Compute Service (ECS) instances

Prerequisites

Before you begin, ensure that you have:

• An ApsaraDB for ClickHouse cluster in the same region and virtual private cloud (VPC) as the source ApsaraMQ for Kafka instance. See Create a cluster.

• A database account with permissions to perform operations on the destination database. See Account management.

• A topic created in the ApsaraMQ for Kafka instance. See Step 1: Create a topic.

• A consumer group created in the ApsaraMQ for Kafka instance. See Step 2: Create a group.

Usage notes

• A topic subscribed by a ClickHouse Kafka external table cannot be consumed by any other consumer simultaneously.

• The field types in the Kafka external table, the materialized view, and the local table must all match.

How it works

ApsaraDB for ClickHouse syncs data from Kafka through a three-layer pipeline:

• Kafka external table — pulls messages from a specified Kafka topic using the built-in Kafka table engine. Direct SELECT queries against this table are not supported by default.

• Materialized view — triggered automatically on each batch from the Kafka external table. It reads the batch and inserts the rows into the local table.

• Local table — the final storage destination. Query this table (or a distributed table built on top of it) for your data.

Step 1: Log on to an ApsaraDB for ClickHouse database

See Connect to a ClickHouse cluster using DMS.

Step 2: Create a Kafka external table

The Kafka external table connects ClickHouse to a Kafka topic and consumes messages. It does not store data.

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`
(
    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';

Step 3: Create a destination table

The destination table stores the synchronized data. The table engine depends on your cluster edition.

For more information on CREATE TABLE syntax, see CREATE TABLE.

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

Step 4: Create a materialized view

The materialized view reads each batch from the Kafka external table and inserts the rows into the destination table.

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 5: Verify data synchronization

1. Send test messages to the Kafka topic:

   1. Log on to the ApsaraMQ for Kafka console.

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

   3. On the Topics page, find the topic and choose More > Send Message in the Actions column.

   4. In the Send to Produce and Consume Message panel, enter the message content. This example sends 1,a and 2,b.

   5. Click OK.

2. Log on to the ApsaraDB for ClickHouse database and query the destination table. See Connect to a ClickHouse cluster using DMS

   Enterprise Edition

   SELECT * FROM kafka_table_local;

   Community-compatible Edition

   On multi-node Community-compatible Edition clusters, always query the distributed table. Querying the local table directly returns data from only one node and produces incomplete results.

   SELECT * FROM kafka_table_distributed;

   Expected output:

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

   If the result matches, data synchronization is working. If not, proceed to Step 6 to check the consumer status.

Step 6 (Optional): Check consumer status

If synchronized data does not match what was sent, query the system tables to inspect the Kafka external table's consumer state.

SELECT * FROM system.kafka_consumers;

SELECT * FROM system.kafka;

FAQ

• ApsaraDB for ClickHouse FAQ

• ApsaraDB for ClickHouse FAQ

• ApsaraDB for ClickHouse FAQ