Realtime Compute for Apache Flink:ClickHouseコネクタ

背景情報

ClickHouseは、オンライン分析処理（OLAP）に使用される列指向データベース管理システムです。詳細については、「What Is ClickHouse?」をご参照ください。

機能

• ClickHouse分散テーブルの場合、データはClickHouse分散テーブルに対応するClickHouseローカルテーブルに直接書き込まれます。

• Alibaba Cloud E-MapReduce（EMR）にデプロイされているClickHouseクラスタの場合、exactly-onceセマンティクスを使用できます。

前提条件

• ClickHouseテーブルが作成されていること。詳細については、「Create a New table」をご参照ください。

• ClickHouseクラスタのホワイトリストが構成されていること。

  • Alibaba Cloud ApsaraDB for ClickHouseクラスタを使用する場合は、「ホワイトリストの構成」の手順に従ってホワイトリストを構成します。

  • Alibaba Cloud EMRにデプロイされているClickHouseクラスタを使用する場合は、「セキュリティグループの管理」の手順に従ってホワイトリストを構成します。

  • Elastic Compute Service（ECS）インスタンスでホストされているセルフマネージドClickHouseクラスタを使用する場合は、「セキュリティグループの概要」の手順に従ってホワイトリストを構成します。

  • その他の場合、ClickHouseクラスタがデプロイされているマシンのホワイトリストを構成して、Realtime Compute for Apache FlinkがデプロイされているマシンからClickHouseクラスタにアクセスできるようにします。

  説明

  Realtime Compute for Apache Flink が属する vSwitch の CIDR ブロックを表示する方法の詳細については、「コンソール操作」をご参照ください。

制限事項

• ClickHouseコネクタは、sink.parallelismパラメータをサポートしていません。

• ClickHouse結果テーブルは、at-least-onceセマンティクスをサポートしています。

• Ververica Runtime（VVR） 3.0.2以降を使用するRealtime Compute for Apache Flinkのみが、ClickHouseコネクタをサポートしています。

• VVR 3.0.3またはVVR 4.0.7、あるいはそれらの以降のマイナーバージョンを使用するRealtime Compute for Apache Flinkのみが、WITH句のignoreDeleteパラメータをサポートしています。

• VVR 4.0.10以降を使用するRealtime Compute for Apache Flinkのみが、ClickHouseのNESTEDデータ型をサポートしています。

• VVR 4.0.11以降を使用するRealtime Compute for Apache Flinkのみが、ClickHouse分散テーブルに対応するClickHouseローカルテーブルにデータを書き込むことができます。

• VVR 4.0.11以降を使用するRealtime Compute for Apache Flinkのみが、Alibaba Cloud EMRにデプロイされているClickHouseクラスタのテーブルにデータを書き込むためのexactly-onceセマンティクスを提供します。EMR ClickHouseの機能変更により、EMR V3.45.1以降のマイナーバージョン、またはEMR V5.11.1以降のマイナーバージョンのClickHouseクラスタのテーブルにデータを書き込む際に、exactly-onceセマンティクスは使用できなくなりました。

• VVR 8.0.7以降を使用するRealtime Compute for Apache Flinkでのみ、writeModeパラメータをbalanceに設定して、ClickHouseローカルテーブルに均等にデータを書き込むことができます。

• ApsaraDB for ClickHouse Community 互換エディションクラスタのみが、ClickHouseローカルテーブルへのデータ書き込みを許可します。

構文

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：データはノードのClickHouseローカルテーブルに書き込まれます。

  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：データはClickHouse分散テーブルに書き込まれます。

  local_table_testという名前の3つのClickHouseローカルテーブルが、192.XX.XX.1、192.XX.XX.2、192.XX.XX.3ノードに存在します。ClickHouseローカルテーブルに基づいて、distributed_table_testという名前のClickHouse分散テーブルが作成されます。

  • 同じキーを持つデータを特定のノードにある同じClickHouseローカルテーブルに直接書き込む場合は、次のステートメントを実行します。

    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パラメータで、ClickHouseローカルテーブルにデータが書き込まれるノードを手動で指定したくない場合は、次のステートメントを実行して、システムがClickHouseローカルテーブルのノードを自動的に推測できるようにします。

    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パラメータをClickHouse分散テーブルが属するノードのJDBC URLに設定します。
      'userName' = '<yourUsername>',
      'password' = '<yourPassword>',
      'tableName' = 'distributed_table_test', -- tableNameパラメータをClickHouse分散テーブルの名前に設定します。
      'shardWrite' = 'true',
      'inferLocalTable' = 'true', -- inferLocalTableパラメータをtrueに設定します。
      'writeMode' = 'partition',
      'shardingKey' = 'name'
    );
    
    INSERT INTO clickhouse_output
    SELECT
      id,
      name,
      age,
      rate
    FROM clickhouse_source;

FAQ

• ApsaraDB for ClickHouse結果テーブルから更新されたデータを取り消すことはできますか？

• ClickHouseシンクテーブルに書き込まれたデータは、いつClickHouseコンソールで表示できますか？