Realtime Compute for Apache Flink：ClickHouse

背景資訊

ClickHouse是一個用於聯機分析（OLAP）的列式資料庫管理系統，詳情請參見What Is ClickHouse?。

特色功能

• 對於ClickHouse的分布式表，支援直接寫對應的本地表。

• 對於EMR的ClickHouse，提供Exactly Once的語義。

前提條件

• 已建立ClickHouse表，詳情請參見建立表。

• 已配置白名單。

  • 如果您使用的是阿里雲資料庫ClickHouse，配置白名單詳情請參見設定白名單。

  • 如果您使用的是阿里雲E-MapReduce的ClickHouse，配置白名單詳情請參見管理安全性群組。

  • 如果您使用的是阿里雲ECS上自建的ClickHouse，配置白名單詳情請參見安全性群組概述。

  • 如果為其他情況，請您自行配置ClickHouse所在機器的白名單讓其可被Flink所在機器訪問即可。

  說明

  如何查看Flink虛擬交換器的網段，請參見如何設定白名單？。

使用限制

• 暫不支援配置sink.parallelism參數。

• ClickHouse結果表保證At-Least-Once語義。

• 僅Flink計算引擎VVR 3.0.2及以上版本支援ClickHouse連接器。

• 僅Flink計算引擎VVR 3.0.3，VVR 4.0.7及以上版本支援ignoreDelete選項。

• 僅Flink計算引擎VVR 4.0.10及以上版本支援ClickHouse的Nested類型。

• 僅Flink計算引擎VVR 4.0.11及以上版本支援直接將資料寫入到ClickHouse分布式表對應的本地表。

• 僅Flink計算引擎VVR 4.0.11及以上版本提供寫EMR的ClickHouse的Exactly Once語義。但對EMR-3.45.1和EMR-5.11.1之後版本的ClickHouse，由於EMR ClickHouse產品能力變更，也不再提供Exactly Once語義。

• 僅Flink計算引擎VVR 8.0.7及以上版本支援使用balance的策略來均勻地將資料寫入ClickHouse的本地表。

• 僅ClickHouse社區相容版支援寫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，分別在192.XX.XX.1、192.XX.XX.2和192.XX.XX.3節點上。然後基於這三個本地表，建立了一個分布式表distributed_table_test。

  • 此時，如果您希望Flink可以直接寫本地表，並且可以按照某個key將相同key的資料寫到同一個節點的本地表中，則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;

  • 此時，如果您不想手動指定本地表的節點，可以讓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;

常見問題

• ClickHouse結果表是否支援回撤更新資料？

• 何時才可以在ClickHouse中看到寫入到結果表的資料?