Flink Connector

[Edit this page](https://github.com/ClickHouse/clickhouse-docs/blob/main/docs/integrations/data-ingestion/apache-flink/flink-connector.md)

ClickHouse Supported

This is the official [Apache Flink Sink Connector](https://github.com/ClickHouse/flink-connector-clickhouse) supported by ClickHouse. It is built using Flink's [AsyncSinkBase](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) and the official ClickHouse [java client](https://github.com/ClickHouse/clickhouse-java).

The connector supports Apache Flink's DataStream API. Table API support is [planned for a future release](https://github.com/ClickHouse/flink-connector-clickhouse/issues/42).

- [Requirements](#requirements)- [Flink version compatibility matrix](#flink-compatibility-matrix)- [Installation & setup](#installation--setup)
      - [Import as a dependency](#import-as-a-dependency)- [Download the binary](#download-the-binary)- [Using the DataStream API](#using-the-datastream-api)
        - [Snippet](#datastream-snippet)- [Quick start example](#datastream-quick-start)- [DataStream API connection options](#datastream-api-connection-options)- [Supported data types](#supported-data-types)
          - [Inserting data from Flink into ClickHouse](#inserting-data-from-flink-into-clickhouse)- [Supported input formats](#supported-input-formats)- [Metrics](#metrics)- [Limitations](#limitations)- [ClickHouse version compatibility and security](#compatibility-and-security)- [Advanced and recommended usage](#advanced-and-recommended-usage)- [Troubleshooting](#troubleshooting)
                      - [CANNOT\_READ\_ALL\_DATA](#cannot_read_all_data)- [Low throughput](#low_throughput)- [I am missing rows in my ClickHouse table](#missing_rows)- [Contributing and support](#contributing-and-support)

## Requirements[​](#requirements "Direct link to Requirements")

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

## Flink version compatibility matrix[​](#flink-compatibility-matrix "Direct link to 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:

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Flink Version Artifact ClickHouse Java Client Version Required Java|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | latest flink-connector-clickhouse-2.0.0 0.9.5 Java 17+|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | 2.0.1 flink-connector-clickhouse-2.0.0 0.9.5 Java 17+|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | 2.0.0 flink-connector-clickhouse-2.0.0 0.9.5 Java 17+|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | 1.20.2 flink-connector-clickhouse-1.17 0.9.5 Java 11+|  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | 1.19.3 flink-connector-clickhouse-1.17 0.9.5 Java 11+|  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | | 1.18.1 flink-connector-clickhouse-1.17 0.9.5 Java 11+|  |  |  |  | | --- | --- | --- | --- | | 1.17.2 flink-connector-clickhouse-1.17 0.9.5 Java 11+ | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | |

Note

The connector has not been tested against Flink versions earlier than 1.17.2

## Installation & setup[​](#installation--setup "Direct link to Installation & setup")

### Import as a dependency[​](#import-as-a-dependency "Direct link to Import as a dependency")

#### For Flink 2.0+[​](#flink-2 "Direct link to For Flink 2.0+")

- Maven- Gradle- SBT

```
<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+[​](#flink-117 "Direct link to For Flink 1.17+")

- Maven- Gradle- SBT

```
<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[​](#download-the-binary "Direct link to 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](https://github.com/ClickHouse/flink-connector-clickhouse/releases)

You can find all available released JAR files in the [Maven Central Repository](https://repo1.maven.org/maven2/com/clickhouse/flink/).

## Using the DataStream API[​](#using-the-datastream-api "Direct link to Using the DataStream API")

### Snippet[​](#datastream-snippet "Direct link to Snippet")

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

- Java

```
public static void main(String[] args) {
    // Configure ClickHouseClient
    ClickHouseClientConfig clickHouseClientConfig = 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,
            clickHouseClientConfig
    );

    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](https://github.com/ClickHouse/flink-connector-clickhouse/tree/main/flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink)
- [flink-connector-clickhouse-2.0.0](https://github.com/ClickHouse/flink-connector-clickhouse/tree/main/flink-connector-clickhouse-2.0.0/src/test/java/org/apache/flink/connector/clickhouse/sink)

### Quick start example[​](#datastream-quick-start "Direct link to Quick start example")

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