Flink Connector

This is the official Apache Flink Sink Connector supported by ClickHouse. It is built using Flink’s AsyncSinkBase and the official ClickHouse java client.

The connector supports Apache Flink’s DataStream API. Table API support is planned for a future release.

Requirements

• Java 11+ (for Flink 1.17+) or 17+ (for Flink 2.0+)
• Apache Flink 1.17+

Flink version compatibility matrix

The connector is split into two artifacts to support both Flink 1.17+ and Flink 2.0+. Choose the artifact that matches your desired Flink version:

Installation & setup

Import as a dependency

For Flink 2.0+

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

dependencies {
    implementation("com.clickhouse.flink:flink-connector-clickhouse-2.0.0:{{ stable_version }}")
}

libraryDependencies += "com.clickhouse.flink" % "flink-connector-clickhouse-2.0.0" % {{ stable_version }} classifier "all"

For Flink 1.17+

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

dependencies {
    implementation("com.clickhouse.flink:flink-connector-clickhouse-1.17:{{ stable_version }}")
}

libraryDependencies += "com.clickhouse.flink" % "flink-connector-clickhouse-1.17" % {{ stable_version }} classifier "all"

Download the binary

The name pattern of the binary JAR is:

flink-connector-clickhouse-${flink_version}-${stable_version}-all.jar

where:

• flink_version is one of 2.0.0 or 1.17
• stable_version is a stable artifact release version

You can find all available released JAR files in the Maven Central Repository.

Using the DataStream API

Snippet

Let’s say you want to insert raw CSV data into ClickHouse:

public static void main(String[] args) {
    // Configure ClickHouseClient
    ClickHouseClientConfig clientConfig = new ClickHouseClientConfig(url, username, password, database, tableName);

    // Create an ElementConverter
    ElementConverter<String, ClickHousePayload> convertorString = new ClickHouseConvertor<>(String.class);

    // Create the sink and set the format using `setClickHouseFormat`
    ClickHouseAsyncSink<String> csvSink = new ClickHouseAsyncSink<>(
            convertorString,
            MAX_BATCH_SIZE,
            MAX_IN_FLIGHT_REQUESTS,
            MAX_BUFFERED_REQUESTS,
            MAX_BATCH_SIZE_IN_BYTES,
            MAX_TIME_IN_BUFFER_MS,
            MAX_RECORD_SIZE_IN_BYTES,
            clientConfig
    );

    csvSink.setClickHouseFormat(ClickHouseFormat.CSV);

    // Finally, connect your DataStream to the sink.
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Path csvFilePath = new Path(fileFullName);
    FileSource<String> csvSource = FileSource
            .forRecordStreamFormat(new TextLineInputFormat(), csvFilePath)
            .build();

    env.fromSource(
            csvSource,
            WatermarkStrategy.noWatermarks(),
            "GzipCsvSource"
    ).sinkTo(csvSink);
}

More examples and snippets can be found in our tests:

• flink-connector-clickhouse-1.17
• flink-connector-clickhouse-2.0.0

Quick start example

We have created maven-based example for an easy start with the ClickHouse Sink:

• Flink 1.17+
• Flink 2.0.0+

For more detailed instructions, see the Example Guide

DataStream API connection options

ClickHouse client options

options and serverSettings should be passed to the client as Map<String, String>. An empty map for either will use client or server defaults, respectively.

For example:

Map<String, String> javaClientOptions = Map.of(
    ClientConfigProperties.CA_CERTIFICATE.getKey(), "<my_CA_cert>",
    ClientConfigProperties.SSL_CERTIFICATE.getKey(), "<my_SSL_cert>",
    ClientConfigProperties.CLIENT_NETWORK_BUFFER_SIZE.getKey(), "30000",
    ClientConfigProperties.HTTP_MAX_OPEN_CONNECTIONS.getKey(), "5"
);

Map<String, String> serverSettings = Map.of(
    "insert_deduplicate", "1"
);

ClickHouseClientConfig clientConfig = new ClickHouseClientConfig(
    url,
    username,
    password,
    database,
    tableName,
    javaClientOptions,
    serverSettings,
    false // enableJsonSupportAsString
);

Sink options

The following options come directly from Flink’s AsyncSinkBase:

Supported data types

The table below provides a quick reference for converting data types when inserting from Flink into ClickHouse.

Inserting data from Flink into ClickHouse

Notes:

• A ZoneId must be provided when performing date operations.
• Precision and scale must be provided when performing decimal operations.
• In order for ClickHouse to parse a Java String as JSON, you need to enable enableJsonSupportAsString in ClickHouseClientConfig.
• The connector requires an ElementConvertor to map elements in the input DataStream to ClickHouse payloads. To this end, the connector provides ClickHouseConvertor and POJOConvertor, which you can use to implement this mapping using the DataWriter serialization methods above.

Supported input formats

You can find the list of available ClickHouse input formats in this documentation page and ClickHouseFormat.java.

To specify the format the connector should use to serialize your DataStream to ClickHouse payloads, use the setClickHouseFormat function. For example:

ClickHouseAsyncSink<String> csvSink = new ClickHouseAsyncSink<>(
        convertorString,
        MAX_BATCH_SIZE,
        MAX_IN_FLIGHT_REQUESTS,
        MAX_BUFFERED_REQUESTS,
        MAX_BATCH_SIZE_IN_BYTES,
        MAX_TIME_IN_BUFFER_MS,
        MAX_RECORD_SIZE_IN_BYTES,
        clientConfig
);
csvSink.setClickHouseFormat(ClickHouseFormat.CSV);

Metrics

The connector exposes the following additional metrics on top of Flink’s existing metrics:

Limitations

• The sink currently provides an at-least-once delivery guarantee. Work toward exactly-once semantics is being tracked here.
• The sink does not yet support a dead-letter queue (DLQ) for buffering unprocessable records. In the meantime, the connector will attempt to re-insert records that fail and drop them if unsuccessful. This feature is being tracked here.
• The sink does not yet support creation via Flink’s Table API or Flink SQL. This feature is being tracked here.

ClickHouse version compatibility and security

• The connector is tested against a range of recent ClickHouse versions, including latest and head, via a daily CI workflow. The tested versions are updated periodically as new ClickHouse releases become active. See here for the versions the connector is tested against daily.
• See the ClickHouse security policy for known security vulnerabilities and how to report a vulnerability.
• We recommend upgrading the connector continuously to not miss security fixes and new improvements.
• If you have an issue with migration, please create a GitHub issue and we will respond!

Advanced and recommended usage

• For optimal performance, ensure your DataStream element type is not a Generic type - see here for Flink’s type distinction. Non-generic elements will avoid the serialization overhead incurred by Kryo and improve throughput to ClickHouse.
• We recommend setting maxBatchSize to at least 1000 and ideally between 10,000 to 100,000. See this guide on bulk inserts for more information.
• To do OLTP-style deduplication or upsert to ClickHouse, refer to this documentation page. Note: this is not to be confused with batch deduplication that happens on retries.

Troubleshooting

CANNOT_READ_ALL_DATA

The following error may occur:

com.clickhouse.client.api.ServerException: Code: 33. DB::Exception: Cannot read all data. Bytes read: 9205. Bytes expected: 1100022.: (at row 9) : While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA)

Cause: Most commonly, the CANNOT_READ_ALL_DATA error means that your ClickHouse table schema has diverged from your Flink record schema. This can happen when one or the other is changed in a backwards incompatible way.

Solution: Update the schema in either (or both) your ClickHouse table or connector input data type so they are compatible. If needed, refer to the type mapping for how to resolve Java types to ClickHouse types. Note: if there are still records in flight, you will need to reset Flink state when you restart the connector.

Low throughput

You may experience that the connector’s throughput does not scale with the job’s parallelism (Flink task count) when writing to ClickHouse.

Cause: ClickHouse’s background part merging process may be slowing down inserts. This can happen when the configured batch size is too small, the connector is flushing too frequently, or a combination of the two.

Solution: Monitor the numRequestSubmitted and actualRecordsPerBatch metrics to help determine how to tune your batch size (maxBatchSize) and how frequently to flush. Also, see Advanced and recommended usage for batch sizing recommendations.

I am missing rows in my ClickHouse table

Cause: The batch(es) were dropped either because of a non-retryable failure or they could not be inserted in the configured number of retries (settable via ClickHouseClientConfig.setNumberOfRetries()). Note: by default, the connector will attempt to re-insert a batch up to 3 times before dropping it.

Solution: Inspect the TaskManager logs and/or stack trace(s) for the root cause.

Contributing and support

If you’d like to contribute to the project or report any issues, we welcome your input! Visit our GitHub repository to open an issue, suggest improvements, or submit a pull request.

Contributions are welcome! Please check the contribution guide in the repository before starting. Thank you for helping improve the ClickHouse Flink connector!