ClickHouse Flink Connector

Table of Contents

• About The Project
• Supported Flink Versions
• Installation
• DataStream API
  • Snippets
  • Examples
• Table API
  • Snippets
  • Examples
• Supported ClickHouse Types
• Configuration Options
  • Client Configuration
  • Sink Configuration
  • Sink Metrics
• Limitations
• Contributing

This is a repo of ClickHouse official Apache Flink Connector supported by the ClickHouse team. The connector supports two main Apache Flink APIs:

• DataStream API
• Table API / Flink SQL (inserts only, no updates or deletes)

Maven

<dependency>
    <groupId>com.clickhouse.flink</groupId>
    <artifactId>flink-connector-clickhouse-2.0.0</artifactId>
    <version>0.2.0</version>
    <classifier>all</classifier>
</dependency>

Maven

<dependency>
    <groupId>com.clickhouse.flink</groupId>
    <artifactId>flink-connector-clickhouse-1.17</artifactId>
    <version>0.2.0</version>
    <classifier>all</classifier>
</dependency>

Configure ClickHouseClient

ClickHouseClientConfig clickHouseClientConfig = new ClickHouseClientConfig(url, username, password, database, tableName);

If you are planning to insert RAW CSV data as is

Create a ClickHouseConvertor

ClickHouseConvertor<String> convertorString = new ClickHouseConvertor<>(String.class);

Build the sink (optional knobs have sensible defaults — set only what you need):

ClickHouseAsyncSink<String> csvSink = ClickHouseAsyncSink.<String>builder()
        .setElementConverter(convertorString)
        .setClickHouseClientConfig(clickHouseClientConfig)
        .setClickHouseFormat(ClickHouseFormat.CSV)
        .setMaxBatchSize(MAX_BATCH_SIZE)
        .setMaxInFlightRequests(MAX_IN_FLIGHT_REQUESTS)
        .setMaxBufferedRequests(MAX_BUFFERED_REQUESTS)
        .setMaxBatchSizeInBytes(MAX_BATCH_SIZE_IN_BYTES)
        .setMaxTimeInBufferMS(MAX_TIME_IN_BUFFER_MS)
        .setMaxRecordSizeInBytes(MAX_RECORD_SIZE_IN_BYTES)
        .build();

Finally, connect your DataStream to the sink.

data.sinkTo(csvSink);

More examples and snippets can be found in our tests flink-connector-clickhouse-1.17 and flink-connector-clickhouse-2.0.0

We have created maven based example for easy start with ClickHouse Sink Different versions for Flink

Java (Maven)

• Flink 1.17+
• Flink 2.0.0+

Scala (sbt)

• Flink 1.17+
• Flink 2.0.0+

For more detailed instructions, see the Example Guide

The connector registers itself as the clickhouse SQL connector: create a table with 'connector' = 'clickhouse' and INSERT INTO it from Flink SQL or the Table API (TableDescriptor.forConnector("clickhouse")). The sink is insert-only: append queries work, update-producing queries (e.g. a plain GROUP BY aggregation) are rejected by the planner — use ReplacingMergeTree/AggregatingMergeTree plus FINAL/argMax for upsert-style needs.

At planning time the connector reads the real column types from the target ClickHouse table and validates the Flink schema against them, so typos, type mismatches, narrowing and unsupported types fail at job submission with a precise message instead of at the first flush. Columns are matched by name, case-sensitively; ClickHouse columns you leave out of the Flink schema get their server-side DEFAULT, or the type's default value (0/''/empty) if they have none — the latter is logged as a warning at planning. A nullable Flink column can only target a Nullable(...) ClickHouse column — declare columns NOT NULL when the target column isn't Nullable (in Flink SQL, columns and collection elements are nullable unless declared otherwise).

An INSERT INTO t (a, b) column list is honoured: the columns the statement leaves out stay out of the request, so the server applies their DEFAULT instead of receiving the planner's padding NULL. This needs Flink 1.18 or newer — Flink 1.17 does not report the column list to the sink, so a partial insert there still writes an explicit NULL into every omitted column.

MATERIALIZED, ALIAS and EPHEMERAL columns take no value from an insert. You may still declare them in the Flink schema as long as every statement omits them through a column list — declare them nullable, since Flink refuses to leave a NOT NULL column out of one. A statement that would write such a column fails at planning and names it.

CREATE TABLE ch_events (
    event_id   BIGINT NOT NULL,
    amount     DECIMAL(18, 4) NOT NULL,
    created_at TIMESTAMP(3) NOT NULL
) WITH (
    'connector' = 'clickhouse',
    'url'       = 'http://localhost:8123',
    'username'  = 'default',
    'password'  = '',
    'database'  = 'analytics',
    'table'     = 'events'
    -- + optional batching/retry options, see below
);

INSERT INTO ch_events SELECT event_id, amount, created_at FROM kafka_src;

Connection options (required unless noted): url, username, password (defaults to ''), database, table.

Batching / backpressure

Reliability

Type / compatibility

Passthrough: clickhouse.client.<key> options are forwarded to the ClickHouse client, clickhouse.server.<key> become per-query server settings. The connector sends print_pretty_type_names = 0 with its schema introspection only, so the type names it reads are canonical (named Tuple columns are otherwise pretty-printed across lines and rejected in the insert header) — this makes ClickHouse 23.10 the SQL sink's minimum server version, as older servers reject the unknown setting — and pins input_format_null_as_default, input_format_defaults_for_omitted_fields and (for JSON columns) input_format_binary_read_json_as_string on every insert; a user copy of any of these settings, in either spelling, is rejected. The client option async is pinned per insert the same way, so it is rejected too.

Flink STRING into a ClickHouse JSON column works out of the box — the connector enables the client's JSON-as-string mode automatically exactly when a JSON column is mapped.

Lossless widening is implicit; a pair that would round a value (DOUBLE into Float32, a narrower Decimal scale or timestamp precision) is rejected at planning. A pair whose values may not all fit the column (the "Notes" below) is accepted and every value is checked at write time, failing the job naming the column, value and range; with 'sink.strict-numeric-mapping' = 'true' the numeric pairs among them are rejected at planning instead, since a wider column can always take their place. Any other pair fails at planning naming the column and both types.

Unsupported: BINARY/VARBINARY, TIME, TIMESTAMP WITH TIME ZONE, INTERVAL; ClickHouse Enum (#43), Variant (#60), Time (#91), IPv4/6, Dynamic, geo — exclude such columns and let server defaults fill them.

Everywhere: nullable Flink columns need Nullable(...) targets — except Nullable(UInt8/16/32/64), blocked until issue #144 — and composites must be NOT NULL. LowCardinality is transparent; SimpleAggregateFunction(f, T) matches as T, top-level only.

Notes for SQL users:

• Operator names are planner-generated (e.g. Sink: ch_events[3]), which affects metric identifiers; numRecordsSend counts at flush, so retried batches double-count versus SELECT count().
• A compiled plan does not freeze the ClickHouse mapping: an ALTER TABLE between planning and execution fails at the first flush (same exposure as the DataStream path).
• The community itinycheng connector also registers the clickhouse identifier; having both jars on the classpath fails with Flink's "multiple factories" error — keep only one.
• SQL gateways without ClickHouse network access can deploy in application mode, where planning runs on the JobManager inside the data plane.

See the round-trip integration test ClickHouseTableApiIntegrationTests for a complete DDL + INSERT INTO example against a real ClickHouse.

This is the DataStream (Java value) view; for the Flink SQL / Table API pairing rules see Type mapping.

• A ZoneId must also be provided when performing date operations.
• Precision and scale must also be provided when performing decimal operations.
• To use JSON type as a string, you need to enable enableJsonSupportAsString in ClickHouseClientConfig .
• SimpleAggregateFunction(f, T) is wire-encoded exactly as its inner type T, so bind the Java value for T and ignore the aggregate function f. Any T supported above works, including Nullable, LowCardinality, Decimal, Array, Map and Tuple.

Our Sink is built on top of Flink’s AsyncSinkBase

Our Sink exposes additional metrics on top of Flink's existing metrics:

• Currently the sink does not support exactly-once semantics

• All projects in this repo are tested with all active LTS versions of ClickHouse.
• The Table API / SQL sink requires ClickHouse 23.10 or newer (see "Passthrough" under Connector options); the DataStream API has no such floor.
• Support policy
• We recommend upgrading the connector continuously to not miss security fixes and new improvements
  • If you have an issue with migration - create and issue and we will respond!

Please see our contributing guide.