背景信息

ClickHouse是一個用于聯(lián)機分析（OLAP）的列式數(shù)據(jù)庫管理系統(tǒng)，詳情請參見什么是ClickHouse。

特色功能

• 對于ClickHouse的分布式表，支持直接寫對應(yīng)的本地表。

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

前提條件

• 已創(chuàng)建ClickHouse表，詳情請參見創(chuàng)建表。

• 已配置白名單。

  • 如果您使用的是阿里云數(shù)據(jù)庫ClickHouse，配置白名單詳情請參見設(shè)置白名單。

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

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

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

  說明

  如何查看Flink虛擬交換機的網(wǎng)段，請參見如何設(shè)置白名單？。

使用限制

• 暫不支持配置sink.parallelism參數(shù)。

• ClickHouse結(jié)果表保證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及以上版本支持直接將數(shù)據(jù)寫入到ClickHouse分布式表對應(yīng)的本地表。

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

• 僅Flink計算引擎VVR 8.0.7及以上版本支持使用balance的策略來均勻地將數(shù)據(jù)寫入ClickHouse的本地表。

• 僅ClickHouse社區(qū)兼容版支持寫ClickHouse本地表。

語法結(jié)構(gòu)

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參數(shù)

類型映射

-- 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單節(jié)點表。

  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分布式表。

  假設(shè)您已經(jīng)有三個本地表，表名為local_table_test，分別在192.XX.XX.1、192.XX.XX.2和192.XX.XX.3節(jié)點上。然后基于這三個本地表，創(chuàng)建了一個分布式表distributed_table_test。

  • 此時，如果您希望Flink可以直接寫本地表，并且可以按照某個key將相同key的數(shù)據(jù)寫到同一個節(jié)點的本地表中，則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;

  • 此時，如果您不想手動指定本地表的節(jié)點，可以讓Flink來自動推測本地表節(jié)點，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', -- 分布式表所在節(jié)點對應(yīng)的JDBC URL。
      'userName' = '<yourUsername>',
      'password' = '<yourPassword>',
      'tableName' = 'distributed_table_test', --為分布式表的名字。
      'shardWrite' = 'true',
      'inferLocalTable' = 'true', --需設(shè)置inferLocalTable為true。
      'writeMode' = 'partition',
      'shardingKey' = 'name'
    );
    
    INSERT INTO clickhouse_output
    SELECT
      id,
      name,
      age,
      rate
    FROM clickhouse_source;

常見問題

• ClickHouse結(jié)果表是否支持回撤更新數(shù)據(jù)？

• 何時才可以在ClickHouse中看到寫入到結(jié)果表的數(shù)據(jù)?