ApsaraDB for ClickHouse:ApsaraMQ for Kafka インスタンスからのデータの同期

制限事項

ApsaraMQ for Kafka インスタンスまたは Elastic Compute Service (ECS) インスタンスにデプロイされた自己管理 Kafka クラスタからのデータのみを ApsaraDB for ClickHouse インスタンスに同期できます。

前提条件

• 同期先 ApsaraDB for ClickHouse クラスター:

  • 同期先クラスターが作成されていること。ソース ApsaraMQ for Kafka インスタンスと同期先クラスターは同じリージョンにデプロイされ、同じ VPC を使用していること。詳細については、「ApsaraDB for ClickHouse クラスターを作成する」をご参照ください。

  • 同期先クラスターのデータベースにログインするためのアカウントが作成されており、データベースに対する操作を実行する権限を持っていること。詳細については、「アカウント管理」をご参照ください。

• ソース ApsaraMQ for Kafka インスタンス:

  • トピックが作成されていること。詳細については、「ステップ 1: トピックを作成する」をご参照ください。

  • コンシューマーグループが作成されていること。詳細については、「ステップ 2: グループを作成する」をご参照ください。

使用上の注意

• トピックが ApsaraDB for ClickHouse Kafka 外部テーブルによってサブスクライブされている場合、そのトピックは他のコンシューマーによって消費されることはできません。

• Kafka 外部テーブル、マテリアライズドビュー、およびローカルテーブルを作成する場合は、3 つのテーブルのフィールドタイプが一致している必要があります。

手順

この例では、ApsaraMQ for Kafka インスタンスから、Community 互換 Edition を実行する ApsaraDB for ClickHouse クラスターのデフォルトデータベース内の kafka_table_distributed 分散テーブルにデータを同期する方法を示します。

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

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;

ステップ 6: データが同期されているかを確認する

1. ApsaraMQ for Kafka インスタンスの Topic にメッセージを送信します。

   1. ApsaraMQ for Kafka コンソール にログインします。

   2. [インスタンス] ページで、管理するインスタンスの名前をクリックします。

   3. [トピック] ページで、必要なトピックを見つけ、[アクション] 列で [詳細] > [メッセージの送信] を選択します。

   4. [メッセージの作成と消費への送信] パネルで、送信するメッセージの内容を指定します。

      この例では、1,a と 2,b というメッセージが送信されます。

   5. [OK] をクリックします。

2. ApsaraDB for ClickHouse データベースにログインし、分散テーブルをクエリして、データが同期されているかどうかを確認します。

   ApsaraDB for ClickHouse データベースへのログイン方法の詳細については、「DMS を使用して ClickHouse に接続する」をご参照ください。

   データ検証用のサンプル文:

   Enterprise Edition

   SELECT * FROM kafka_table_local; 

   Community 互換 Edition

   次の文は、分散テーブルをクエリする方法を示しています。

   • デスティネーションテーブルがローカルテーブルの場合、文中の分散テーブル名をローカルテーブル名に置き換える必要があります。

   • 複数のノードを持つ Community 互換 Edition クラスタを使用する場合は、分散テーブルからデータをクエリすることをお勧めします。そうしないと、クラスタ内の 1 つのノードからのみデータがクエリされ、結果が不完全になります。

   SELECT * FROM kafka_table_distributed; 

   クエリを実行して成功の結果を受け取った場合、ApsaraMQ for Kafka インスタンスから ApsaraDB for ClickHouse クラスタにデータが同期されていることを示します。

   サンプル結果:

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

   クエリ結果が期待値と異なる場合は、「(オプション) ステップ 7: Kafka 外部テーブルの消費ステータスを表示する」に記載されている手順に従って、問題のトラブルシューティングを行うことができます。

select * from system.kafka_consumers;

SELECT * FROM system.kafka;

よくある質問

• ApsaraDB for ClickHouse クラスターの外部テーブルを MySQL、Hadoop 分散ファイルシステム (HDFS)、Kafka などのプラットフォーム上のテーブルに接続できないのはなぜですか。

• Kafka インスタンスのテーブルと ApsaraDB for ClickHouse クラスターの外部テーブル間でデータを同期した後、テーブル間の行数が一致しないのはなぜですか。

• Kafka 外部テーブルのデータを格納するローカルテーブルのデータ量が、外部テーブルからデータを書き込んだ後も変更されないのはなぜですか。