Realtime Compute for Apache Flink:ClickHouse

Prerequisites

Before you begin, ensure that you have:

• A ClickHouse table. For more information, see Create a new table.

• A whitelist configured for your ClickHouse cluster:

  • ApsaraDB for ClickHouse cluster: Configure the whitelist

  • Alibaba Cloud E-MapReduce (EMR) ClickHouse cluster: Manage security groups

  • Self-managed ClickHouse on Elastic Compute Service (ECS): Overview

  • Other deployments: Add the CIDR block of the vSwitch used by Realtime Compute for Apache Flink to the allowlist of the machines running your ClickHouse cluster. To find the CIDR block, see FAQ about workspace and namespace management and operations.

Connector capabilities

Limitations

• The sink.parallelism parameter is not supported.

• Result tables support at-least-once semantics by default.

• Requires Ververica Runtime (VVR) 3.0.2 or later.

• The ignoreDelete parameter requires VVR 3.0.3, VVR 4.0.7, or a later minor version of either.

• The NESTED data type requires VVR 4.0.10 or later.

• Writing directly to a ClickHouse local table that corresponds to a distributed table requires VVR 4.0.11 or later. Only ApsaraDB for ClickHouse Community-compatible Edition clusters support this.

• Exactly-once semantics for EMR ClickHouse clusters requires VVR 4.0.11 or later. This is no longer available for EMR V3.45.1 clusters or minor versions later than EMR V5.11.1, due to a capability change in EMR ClickHouse.

• The balance write mode requires VVR 8.0.7 or later.

Syntax

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'
);

Parameters in the WITH clause

Choose a write mode

The writeMode parameter controls how records are distributed across local table nodes when writing to a ClickHouse distributed table with shardWrite=true.

writeMode=partition requires ignoreDelete=true and is incompatible with exactlyOnce=true.

Data type mappings

CREATE TABLE visits (
  StartDate Date,
  Goals Nested
  (
    ID UInt32,
    OrderID String
  )
);

CREATE TABLE visits (
  StartDate DATE,
  `Goals.ID`      ARRAY<LONG>,
  `Goals.OrderID` ARRAY<STRING>
);

Examples

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;

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;

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 of any node hosting the distributed table
  'userName'       = '<yourUsername>',
  'password'       = '<yourPassword>',
  'tableName'      = 'distributed_table_test',                    -- Name of the distributed table
  'shardWrite'     = 'true',
  'inferLocalTable' = 'true',                                     -- Auto-discover local table nodes
  'writeMode'      = 'partition',
  'shardingKey'    = 'name'
);

INSERT INTO clickhouse_output
SELECT id, name, age, rate
FROM clickhouse_source;

FAQ

Can I retract (delete) updated data from an ApsaraDB for ClickHouse result table?

Set ignoreDelete=false and define a primary key in the DDL statement. The connector then issues ALTER DELETE statements when it receives retract messages. Note that this significantly reduces write throughput, and it cannot be used together with writeMode=partition or exactlyOnce=true. For more details, see Can I retract the updated data from an ApsaraDB for ClickHouse result table?

When does written data become visible in ClickHouse?

ClickHouse uses asynchronous part merging, so data may not be immediately visible after a write completes. The delay depends on the merge interval configured in your ClickHouse cluster. For more details, see When can I view the data that is written to a ClickHouse sink table in the ClickHouse console?