Realtime Compute for Apache Flink:ClickHouse コネクタ

背景情報

ClickHouse は、オンライン分析処理 (OLAP) 向けの列指向データベース管理システムです。 詳細については、「ClickHouse とは」をご参照ください。

機能

• ClickHouse 分散テーブルのローカルテーブルにデータを直接書き込みます。

• Alibaba Cloud E-MapReduce (EMR) 上の ClickHouse に書き込む際に、exactly-once セマンティクスを実現します。

前提条件

• ClickHouse テーブルを作成します。 詳細については、「新しいテーブルの作成」をご参照ください。

• ホワイトリストを設定します。

  • ApsaraDB for ClickHouse を使用する場合は、「ホワイトリストの設定」をご参照ください。

  • Alibaba Cloud E-MapReduce (EMR) 上の ClickHouse を使用する場合は、「セキュリティグループの管理」をご参照ください。

  • ECS インスタンス上のセルフマネージド ClickHouse クラスターを使用する場合は、「セキュリティグループの概要」をご参照ください。

  • その他のすべての設定では、ClickHouse がデプロイされているマシンでホワイトリストを設定し、Realtime Compute for Apache Flink デプロイメントからのアクセスを許可する必要があります。

  説明

  Realtime Compute for Apache Flink の vSwitch ネットワークセグメントを表示するには、「ホワイトリストの設定方法」をご参照ください。

制限事項

• sink.parallelism パラメータはサポートされていません。

• デフォルトでは、ClickHouse シンクは at-least-once セマンティクスを提供します。

• ClickHouse コネクタは、Ververica Runtime (VVR) 3.0.2 以降でのみサポートされています。

• ignoreDelete オプションは、VVR 3.0.3、VVR 4.0.7 以降のバージョンでのみサポートされています。

• ClickHouse Nested データ型は、VVR 4.0.10 以降でのみサポートされています。

• 分散テーブルのローカルテーブルへの直接書き込みは、VVR 4.0.11 以降でのみサポートされています。

• EMR 上の ClickHouse への書き込みにおける exactly-once セマンティクスは、VVR 4.0.11 以降で利用可能です。ただし、製品機能の変更により、このセマンティクスは、ClickHouse on EMR のバージョンが v3.45.1 または v5.11.1 より後の場合は利用できません。

• ローカルテーブルノード間でデータを均等に分散する balance 書き込みモードは、VVR 8.0.7 以降でのみ利用可能です。

• ClickHouse ローカルテーブルへの書き込みは、ApsaraDB for ClickHouse Community-compatible Edition でのみサポートされています。

• ClickHouse カタログを登録する際、デフォルトのデータベース名にハイフン (-) が含まれていると、JDBC URL の検証が失敗します。

構文

CREATE TABLE clickhouse_sink (
  id INT,
  name VARCHAR,
  age BIGINT,
  rate FLOAT
) WITH (
  'connector' = 'clickhouse',
  'url' = '<yourUrl>',
  'userName' = '<yourUsername>',
  'password' = '<yourPassword>',
  'tableName' = '<yourTablename>',
  'maxRetryTimes' = '3',
  'batchSize' = '8000',
  'flushIntervalMs' = '1000',
  'ignoreDelete' = 'true',
  'shardWrite' = 'false',
  'writeMode' = 'partition',
  'shardingKey' = 'id'
);

WITH パラメータ

データ型のマッピング

-- ClickHouse
CREATE TABLE visits (
  StartDate Date,
  Goals Nested
  (
    ID UInt32,
    OrderID String
  )
  ...
);

-- Flink
CREATE TABLE visits (
  StartDate DATE,
  `Goals.ID` ARRAY<LONG>,
  `Goals.OrderID` ARRAY<STRING>
);

例

• 例 1：シングルノードテーブルへの書き込み。

  CREATE TEMPORARY TABLE clickhouse_source (
    id INT,
    name VARCHAR,
    age BIGINT,
    rate FLOAT
  ) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '50'
  );
  
  CREATE TEMPORARY TABLE clickhouse_output (
    id INT,
    name VARCHAR,
    age BIGINT,
    rate FLOAT
  ) WITH (
    'connector' = 'clickhouse',
    'url' = '<yourUrl>',
    'userName' = '<yourUsername>',
    'password' = '<yourPassword>',
    'tableName' = '<yourTablename>'
  );
  
  INSERT INTO clickhouse_output
  SELECT
    id,
    name,
    age,
    rate
  FROM clickhouse_source;

• 例 2：分散テーブルへの書き込み。

  ノード 192.XX.XX.1、192.XX.XX.2、192.XX.XX.3 にある local_table_test という名前の 3 つのローカルテーブルから、distributed_table_test という名前の分散テーブルが作成されていると仮定します。

  • Flink でローカルテーブルにデータを直接書き込み、キーによってデータをパーティション分割するには、次の DDL を使用します。

    CREATE TEMPORARY TABLE clickhouse_source (
      id INT,
      name VARCHAR,
      age BIGINT,
      rate FLOAT
    ) WITH (
      'connector' = 'datagen',
      'rows-per-second' = '50'
    );
    
    CREATE TEMPORARY TABLE clickhouse_output (
      id INT,
      name VARCHAR,
      age BIGINT,
      rate FLOAT
    ) WITH (
      'connector' = 'clickhouse',
      'url' = 'jdbc:clickhouse://192.XX.XX.1:3002,192.XX.XX.2:3002,192.XX.XX.3:3002/default',
      'userName' = '<yourUsername>',
      'password' = '<yourPassword>',
      'tableName' = 'local_table_test',
      'shardWrite' = 'true',
      'writeMode' = 'partition',
      'shardingKey' = 'name'
    );
    
    INSERT INTO clickhouse_output
    SELECT
      id,
      name,
      age,
      rate
    FROM clickhouse_source;

  • URL で手動で指定する代わりに、Flink にローカルテーブルノードを自動的に検出させるには、次の DDL を使用します。

    CREATE TEMPORARY TABLE clickhouse_source (
      id INT,
      name VARCHAR,
      age BIGINT,
      rate FLOAT
    ) WITH (
      'connector' = 'datagen',
      'rows-per-second' = '50'
    );
    
    CREATE TEMPORARY TABLE clickhouse_output (
      id INT,
      name VARCHAR,
      age BIGINT,
      rate FLOAT
    ) WITH (
      'connector' = 'clickhouse',
      'url' = 'jdbc:clickhouse://192.XX.XX.1:3002/default', -- クラスター内のノードの JDBC URL。
      'userName' = '<yourUsername>',
      'password' = '<yourPassword>',
      'tableName' = 'distributed_table_test', -- 分散テーブルの名前。
      'shardWrite' = 'true',
      'inferLocalTable' = 'true', -- inferLocalTable を true に設定します。
      'writeMode' = 'partition',
      'shardingKey' = 'name'
    );
    
    INSERT INTO clickhouse_output
    SELECT
      id,
      name,
      age,
      rate
    FROM clickhouse_source;

よくある質問

• ApsaraDB for ClickHouse の結果テーブルから更新されたデータをリトラクトできますか？

• ClickHouse シンクテーブルでのデータの可視性