实时计算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中看到写入到结果表的数据?