实时计算Flink版：ClickHouse

背景信息

ClickHouse是一个用于联机分析（OLAP）的列式数据库管理系统，详情请参见What Is ClickHouse?。

特色功能

• 对于ClickHouse的分布式表，支持直接写对应的本地表。

• 对于EMR的ClickHouse，提供Exactly Once的语义。

前提条件

• 已创建ClickHouse表，详情请参见创建表。

• 已配置白名单。

  • 如果您使用的是阿里云数据库ClickHouse，配置白名单详情请参见设置白名单。

  • 如果您使用的是阿里云E-MapReduce的ClickHouse，配置白名单详情请参见管理安全组。

  • 如果您使用的是阿里云ECS上自建的ClickHouse，配置白名单详情请参见安全组概述。

  • 如果为其他情况，请您自行配置ClickHouse所在机器的白名单让其可被Flink所在机器访问即可。

  说明

  如何查看Flink全托管交换机的网段，请参见如何设置白名单？。

使用限制

• 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语义。

语法结构

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:8123,192.XX.XX.2:8123,192.XX.XX.3:8123/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;

常见问题

ClickHouse结果表是否支持回撤更新数据？