Synchronize data from Kafka

Limitations

You can synchronize data only from ApsaraMQ for Kafka instances and self-managed Kafka clusters that are deployed on ECS instances.

Prerequisites

• ApsaraDB for ClickHouse:

  • A destination cluster has been created in the same region and VPC as the ApsaraMQ for Kafka instance. For more information, see Create a cluster.

  • A database account with the required permissions has been created for the destination cluster. For more information, see Account Management.

• ApsaraMQ for Kafka:

  • A topic has been created.

  • A consumer group has been created.

Usage notes

• The topic that the ApsaraDB for ClickHouse Kafka external table subscribes to must not have other consumers.

• When you create the Kafka external table, materialized view, and local table, the field types of the three tables must match.

Procedure

The following example synchronizes data from ApsaraMQ for Kafka to the kafka_table_distributed distributed table in the default database of a Community-compatible Edition cluster of ApsaraDB for ClickHouse.

Step 1: Understand how synchronization works

ApsaraDB for ClickHouse uses the Kafka table engine and a materialized view to consume and store data from Kafka in real time. The data flow is as follows.

• Kafka topic: The source data to be synchronized.

• ApsaraDB for ClickHouse Kafka external table (a table that uses the Kafka table engine): pulls source data from a specified Kafka topic.

• Materialized view: reads source data from the external table for Kafka and inserts the data into a local table in ApsaraDB for ClickHouse.

• Local table: stores the synchronized 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`
(
    -- Define the fields of the table schema.
    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 synchronization

1. Send messages to the topic in the ApsaraMQ for Kafka instance.

   1. Log on to the ApsaraMQ for Kafka console.

   2. On the Instance list page, click the name of the destination instance.

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

   4. On the Send and Consume Message with Quick Experience page, enter the Message Content.

      This example sends the messages 1,a and 2,b.

   5. Click OK.

2. Log on to the ApsaraDB for ClickHouse cluster, query the distributed table, and check whether the data is synchronized.

   For more information about how to log on to an ApsaraDB for ClickHouse cluster, see Connect to an ApsaraDB for ClickHouse cluster by using DMS.

   Use the following statements to query and verify data:

   Enterprise edition

   SELECT * FROM kafka_table_local; 

   Community-compatible edition

   The following code 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 in the query with the local table name.

   • If you use a Community-compatible Edition cluster that is a multi-node cluster, we strongly recommend that you query the distributed table. If you query a local table directly, it will only return data from a single node, resulting in an incomplete result set.

   SELECT * FROM kafka_table_distributed; 

   If the query returns results, the data synchronization from Kafka to ApsaraDB for ClickHouse is successful.

   The query results are as follows.

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

   If the query results are not as expected, proceed to Step 7 (Optional): Check the consumption status of the Kafka external table to further troubleshoot the issue.

select * from system.kafka_consumers;

SELECT * FROM system.kafka;

FAQ

• Why are external tables for sources such as MySQL, HDFS, and Kafka unreachable?

• Why does the row count in ClickHouse not match the number of messages imported from Kafka?

• Why is no new data written to the Kafka external table after the table is created?