ApsaraDB for ClickHouse:Kafka からのデータ同期

制限事項

データは、ApsaraMQ for Kafka インスタンス、および ECS インスタンスにデプロイされたセルフマネージド Kafka クラスターからのみ同期できます。

前提条件

• ApsaraDB for ClickHouse ：

  • ApsaraMQ for Kafka インスタンスと同じリージョンおよび VPC 内にターゲットクラスターを作成しておく必要があります。詳細については、「クラスターの作成」をご参照ください。

  • ターゲットクラスター用に、必要な権限を持つデータベースアカウントを作成しておく必要があります。詳細については、「アカウント管理」をご参照ください。

• ApsaraMQ for Kafka ：

  • トピックを作成しておく必要があります。

  • コンシューマーグループを作成しておく必要があります。

注意事項

• ApsaraDB for ClickHouse の Kafka 外部テーブルがサブスクライブするトピックには、他のコンシューマーが存在しない必要があります。

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

手順

次の例では、ApsaraMQ for Kafka から ApsaraDB for ClickHouse Community-compatible Edition クラスターの default データベースにある kafka_table_distributed 分散テーブルにデータを同期します。

手順1：同期の仕組みの理解

ApsaraDB for ClickHouse は、Kafka テーブルエンジンとマテリアライズドビューを使用して、Kafka からリアルタイムでデータを消費および格納します。 データフローは次のとおりです：

• Kafka トピック：同期対象のソースデータ。

• ApsaraDB for ClickHouse の Kafka 外部テーブル (Kafka テーブルエンジンを使用するテーブル) ：指定された Kafka トピックからソースデータをプルします。

• マテリアライズドビュー： Kafka の外部テーブルからソースデータを読み取り、ApsaraDB for ClickHouse のローカルテーブルにデータを挿入します。

• ローカルテーブル：同期されたデータを格納します。

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 = 'value']

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 インスタンスのトピックにメッセージを送信します。

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

   2. インスタンスリスト ページで、対象インスタンスの名前をクリックします。

   3. [トピック] ページで、対象トピックを見つけ、操作 列で より > [メッセージ送信 (デモ)] を選択します。

   4. [クイック体験によるメッセージの送受信] ページで、Message Content を入力します。

      この例では、1,a および 2,b のメッセージを送信します。

   5. を決定 をクリックします。

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

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

   次のステートメントを使用してデータをクエリし、検証します。

   Enterprise Edition

   SELECT * FROM kafka_table_local; 

   Community-compatible Edition

   次のコードは、分散テーブルをクエリする例を示しています。

   • 宛先テーブルがローカルテーブルの場合は、クエリ内の 分散テーブル 名をローカルテーブル名に置き換える必要があります。

   • マルチノードクラスターである Community-compatible Edition クラスター を使用する場合は、分散テーブルをクエリすることを 強く推奨します。ローカルテーブルを直接クエリすると、単一ノードのデータのみが返され、結果セットが不完全になります。

   SELECT * FROM kafka_table_distributed; 

   クエリで結果が返されると、Kafka から ApsaraDB for ClickHouse へのデータ同期は成功です。

   クエリ結果は次のとおりです。

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

   クエリ結果が期待どおりでない場合は、「ステップ 7 (オプション)：Kafka 外部テーブルの消費ステータスの確認」に進み、問題のトラブルシューティングを行ってください。

select * from system.kafka_consumers;

SELECT * FROM system.kafka;

よくある質問

• MySQL、HDFS、Kafka などをソースとする外部テーブルにアクセスできないのはなぜですか？

• ClickHouse の行数が、Kafka からインポートしたメッセージ数と一致しないのはなぜですか？

• テーブル作成後、Kafka 外部テーブルに新しいデータが書き込まれないのはなぜですか？