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

About The Project

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

• DataStreamAPI
• Table API (This feature is not implemented yet and is planned for a future release)

Supported Flink Versions

Installation

For Flink 2.0.0+

Maven

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

For Flink 1.17+

Maven

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

DataStream API

Snippet

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

Example

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

Table API

Table API is planned for a future release. This section will be updated once available.

Snippet

Planned for a future release — this section will provide a usage snippet for configuring the Table API.

Example

Planned for a future release — a complete end-to-end example will be added once the Table API becomes available.

Supported ClickHouse Types

• 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.

Configuration Options

Client configuration

Sink configuration

Our Sink is built on top of Flink’s AsyncSinkBase

Sink Metrics

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

Limitations

• Currently the sink does not support exactly-once semantics

Compatibility

• All projects in this repo are tested with all active LTS versions of ClickHouse.
• 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!

Contributing

Please see our contributing guide.