Realtime Compute for Apache Flink:ClickHouse コネクタ

前提条件

開始する前に、以下の条件を満たしていることを確認してください。

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

• ClickHouse クラスターのホワイトリストが設定されていること：

  • ApsaraDB for ClickHouse クラスター：「ホワイトリストの設定」

  • Alibaba Cloud E-MapReduce (EMR) ClickHouse クラスター：「セキュリティグループの管理」

  • Elastic Compute Service (ECS) 上の自社管理型 ClickHouse：「概要」

  • その他のデプロイメント: Realtime Compute for Apache Flink で使用される vSwitch の CIDR ブロックを、ClickHouse クラスターを実行しているマシンの許可リストに追加します。CIDR ブロックを確認するには、「ワークスペースおよび名前空間の管理と操作に関するよくある質問」をご参照ください。

コネクタの機能

制限事項

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

• 結果テーブルは、デフォルトで at-least-once セマンティクスをサポートします。

• Ververica Runtime (VVR) 3.0.2 以降が必要です。

• ignoreDelete パラメーターは、VVR 3.0.3、VVR 4.0.7、またはそれ以降のマイナーバージョンを必要とします。

• NESTED データの型は、VVR 4.0.10 以降が必要です。

• 分散テーブルに対応する ClickHouse のローカルテーブルへの直接書き込みには、VVR 4.0.11 以降が必要です。これは、ApsaraDB for ClickHouse Community 互換エディションクラスターでのみサポートされます。

• EMR ClickHouse クラスターにおける 1 回限りのセマンティクスは、VVR 4.0.11 以降が必要です。ただし、EMR ClickHouse の機能変更により、EMR V3.45.1 クラスターおよび EMR V5.11.1 より後のマイナーバージョンでは利用できなくなりました。

• balance 書き込みモードは、VVR 8.0.7 以降が必要です。

構文

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 句のパラメーター

書き込みモードの選択

writeMode パラメーターは、shardWrite=true で ClickHouse の分散テーブルに書き込む際、レコードをローカルテーブルノード間でどのように分散するかを制御します。

writeMode=partition は ignoreDelete=true を必要とし、exactlyOnce=true とは併用できません。

データの型のマッピング

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

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

サンプル

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;

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;

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', -- 分散テーブルをホストする任意のノードの URL
  'userName'       = '<yourUsername>',
  'password'       = '<yourPassword>',
  'tableName'      = 'distributed_table_test',                    -- 分散テーブルの名前
  'shardWrite'     = 'true',
  'inferLocalTable' = 'true',                                     -- ローカルテーブルノードを自動検出
  'writeMode'      = 'partition',
  'shardingKey'    = 'name'
);

INSERT INTO clickhouse_output
SELECT id, name, age, rate
FROM clickhouse_source;

よくある質問

ApsaraDB for ClickHouse の結果テーブルから更新済みデータを撤回（削除）できますか？

ignoreDelete=false を設定し、DDL 文でプライマリキーを定義します。その後、コネクタは撤回メッセージを受信した際に ALTER DELETE 文を発行します。ただし、この機能を有効化すると、書き込みスループットが大幅に低下し、writeMode=partition や exactlyOnce=true との併用はできません。詳細については、「ApsaraDB for ClickHouse の結果テーブルから更新済みデータを撤回できますか？」をご参照ください。

書き込んだデータは、ClickHouse でいつから表示可能になりますか？

ClickHouse は非同期のパートマージを使用するため、書き込み完了直後にはデータが即座に表示されない場合があります。遅延時間は、ClickHouse クラスターで設定されたマージ間隔によって異なります。詳細については、「ClickHouse コンソールで ClickHouse sink テーブルに書き込まれたデータをいつ確認できますか？」をご参照ください。