ApsaraDB for ClickHouse:Synchronize data from an ApsaraMQ for Kafka instance

Limits

Only data from ApsaraMQ for Kafka instances or self-managed Kafka clusters deployed on Elastic Compute Service (ECS) instances can be synchronized to ApsaraDB for ClickHouse instances.

Prerequisites

• Destination ApsaraDB for ClickHouse cluster:

  • A destination cluster is created. A souce ApsaraMQ for Kafka instance and the destination cluster are deployed in the same region and use the same virtual private cloud (VPC). For more information, see Create an ApsaraDB for ClickHouse cluster.

  • An account is created to log on to a database in the destination cluster and has the permissions to perform operations on the database. For more information, see Account management.

• Source ApsaraMQ for Kafka instance:

  • A topic is created. For more information, see Step 1: Create a topic.

  • A consumer group is created. For more information, see Step 2: Create a group.

Usage notes

• If a topic is subscribed by an ApsaraDB for ClickHouse Kafka external table, the topic cannot be consumed by other consumers.

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

Procedure

This example shows how to synchronize data from an ApsaraMQ for Kafka instance to the kafka_table_distributed distributed table in the default database of an ApsaraDB for ClickHouse cluster that runs Community-compatible Edition.

Step 1: Understand the synchronization principle

ApsaraDB for ClickHouse uses a built-in Kafka table engine and the materialized view mechanism to synchronize data from ApsaraMQ for Kafka instances. This achieves real-time data consumption and storage. The following figure shows the synchronization process.

• Kafka topic: specifies the source data to be synchronized.

• ApsaraDB for ClickHouse Kafka external table: pulls source data from a specified Kafka topic.

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

• 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`
(-- Fields that define 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: Check whether data is synchronized

1. Send messages to a topic of the ApsaraMQ for Kafka instance.

   1. Log on to the ApsaraMQ for Kafka console.

   2. On the Instances page, click the name of the instance that you want to manage.

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

   4. In the Send to Produce and Consume Message panel, specify the message content that you want to send.

      In this example, the messages 1,a and 2,b are sent.

   5. Click OK.

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

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

   Sample statements for verifying data:

   Enterprise Edition

   SELECT * FROM kafka_table_local; 

   Community-compatible Edition

   The following statement shows 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 statement.

   • If you use a Community-compatible Edition cluster with multiple nodes, we recommend that you query data from the distributed table. Otherwise, data is queried only from one node in the cluster, leading to incomplete results.

   SELECT * FROM kafka_table_distributed; 

   When you execute a query and receive a successful result, it indicates that data is synchronized from the ApsaraMQ for Kafka instance to the ApsaraDB for ClickHouse cluster.

   Sample result:

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

   If the query result is different from what you expect, you can further troubleshoot the issue by following the instructions described in (Optional) Step 7: View the consumption status of the Kafka external table.

select * from system.kafka_consumers;

SELECT * FROM system.kafka;

FAQ

• Why am I unable to connect an external table in an ApsaraDB for ClickHouse cluster to a table on platforms such as MySQL, Hadoop Distributed File System (HDFS), and Kafka?

• Why is the number of rows inconsistent between a table in a Kafka instance and an external table in an ApsaraDB for ClickHouse cluster after data is synchronized between the tables?

• Why does the volume of data in the local table that stores the data of a Kafka external table remain unchanged after I write data from the external table?