ClickHouse connector

Background information

ClickHouse is a column-oriented database management system for Online Analytical Processing (OLAP). For more information, see What Is ClickHouse?.

Features

• Write data directly to the local tables of a ClickHouse distributed table.

• Achieve exactly-once semantics when writing to ClickHouse on Alibaba Cloud E-MapReduce (EMR).

Prerequisites

• Create a ClickHouse table. For more information, see Create a New table.

• Configure a whitelist.

  • If you use ApsaraDB for ClickHouse, see Configure a whitelist.

  • If you use ClickHouse on Alibaba Cloud E-MapReduce (EMR), see Manage security groups.

  • If you use a self-managed ClickHouse cluster on an ECS instance, see Security group overview.

  • For all other setups, configure the whitelist on the machine where ClickHouse is deployed to allow access from the Realtime Compute for Apache Flink deployment.

  Note

  To view the vSwitch network segment for Realtime Compute for Apache Flink, see How do I configure a whitelist?.

Limitations

• The sink.parallelism parameter is not supported.

• By default, the ClickHouse sink provides at-least-once semantics.

• The ClickHouse connector is supported only in Ververica Runtime (VVR) 3.0.2 and later.

• The ignoreDelete option is supported only in VVR 3.0.3, VVR 4.0.7, and later versions.

• The ClickHouse Nested data type is supported only in VVR 4.0.10 and later.

• Direct writes to the local tables of a distributed table are supported only in VVR 4.0.11 and later.

• The exactly-once semantic for writing to ClickHouse on EMR is available only in VVR 4.0.11 and later. However, due to product capability changes, this semantic is unavailable for ClickHouse on EMR v3.45.1 and versions later than v5.11.1.

• The balance write mode, which evenly distributes data across local table nodes, is available only in VVR 8.0.7 and later.

• Writing to a ClickHouse local table is supported only for ApsaraDB for ClickHouse Community-compatible Edition.

• When you register a ClickHouse catalog, the default database name cannot contain a hyphen (-). Otherwise, the JDBC URL validation fails.

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

WITH parameters

Data type mapping

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

Examples

• Example 1: Write to a single-node table.

  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;

• Example 2: Write to a distributed table.

  Assume a distributed table named distributed_table_test is created from three local tables named local_table_test, which are located on nodes 192.XX.XX.1, 192.XX.XX.2, and 192.XX.XX.3.

  • If you want Flink to write data directly to the local tables and partition the data by a key, use the following 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;

  • If you want Flink to automatically discover the local table nodes instead of specifying them manually in the URL, use the following 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', -- The JDBC URL of a node in the cluster.
      'userName' = '<yourUsername>',
      'password' = '<yourPassword>',
      'tableName' = 'distributed_table_test', -- The name of the distributed table.
      'shardWrite' = 'true',
      'inferLocalTable' = 'true', -- Set inferLocalTable to true.
      'writeMode' = 'partition',
      'shardingKey' = 'name'
    );
    
    INSERT INTO clickhouse_output
    SELECT
      id,
      name,
      age,
      rate
    FROM clickhouse_source;

FAQ

• Can I retract the updated data from an ApsaraDB for ClickHouse result table?

• Data visibility in ClickHouse sink tables