Flink Connector

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:

- [Flink 1.17+](https://github.com/ClickHouse/flink-connector-clickhouse/tree/main/examples/maven/flink-v1.7/covid)
- [Flink 2.0.0+](https://github.com/ClickHouse/flink-connector-clickhouse/tree/main/examples/maven/flink-v2/covid)

For more detailed instructions, see the [Example Guide](https://github.com/ClickHouse/flink-connector-clickhouse/blob/main/examples/README.md)

### DataStream API connection options[​](#datastream-api-connection-options "Direct link to DataStream API connection options")

#### ClickHouse client options[​](#client-options "Direct link to ClickHouse client options")

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Parameters Description Default Value Required|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `url` Fully qualified Clickhouse URL N/A Yes|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `username` ClickHouse database username N/A Yes|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `password` ClickHouse database password N/A Yes|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `database` ClickHouse database name N/A Yes|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `table` ClickHouse table name N/A Yes|  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `options` Map of Java client configuration options Empty map No|  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | | `serverSettings` Map of ClickHouse server session settings Empty map No|  |  |  |  | | --- | --- | --- | --- | | `enableJsonSupportAsString` ClickHouse server setting to expect a JSON formatted String for the [JSON data type](https://clickhouse.com/docs/sql-reference/data-types/newjson) true No | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | |

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

Note

All available Java client options are listed in [ClientConfigProperties.java](https://github.com/ClickHouse/clickhouse-java/blob/main/client-v2/src/main/java/com/clickhouse/client/api/ClientConfigProperties.java) and [this documentation page](https://clickhouse.com/docs/integrations/language-clients/java/client#configuration).

All available server session settings are listed in [this documentation page](https://clickhouse.com/docs/operations/settings/settings).

For example:

- Java

```
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 clickHouseClientConfig = new ClickHouseClientConfig(
    url,
    username,
    password,
    database,
    tableName,
    javaClientOptions,
    serverSettings,
    false // enableJsonSupportAsString
);
```

#### Sink options[​](#sink-options "Direct link to Sink options")

The following options come directly from Flink's `AsyncSinkBase`:

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Parameters Description Default Value Required|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `maxBatchSize` Maximum number of records inserted in a single batch N/A Yes|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `maxInFlightRequests` The maximum number of in flight requests allowed before the sink applies backpressure N/A Yes|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `maxBufferedRequests` The maximum number of records that may be buffered in the sink before backpressure is applied N/A Yes|  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `maxBatchSizeInBytes` The maximum size (in bytes) a batch may become. All batches sent will be smaller than or equal to this size N/A Yes|  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | | `maxTimeInBufferMS` The maximum time a record may stay in the sink before being flushed N/A Yes|  |  |  |  | | --- | --- | --- | --- | | `maxRecordSizeInBytes` The maximum record size that the sink will accept, records larger than this will be automatically rejected N/A Yes | | | | | | | | | | | | | | | | | | | | | | | | | | | |

## Supported data types[​](#supported-data-types "Direct link to 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[​](#inserting-data-from-flink-into-clickhouse "Direct link to Inserting data from Flink into ClickHouse")