Flink Connector

## Advanced and recommended usage[​](#advanced-and-recommended-usage "Direct link to Advanced and recommended usage")

- For optimal performance, ensure your DataStream element type is **not** a Generic type - see [here for Flink's type distinction](https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/dev/datastream/fault-tolerance/serialization/types_serialization/#flinks-typeinformation-class). 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](/docs/optimize/bulk-inserts) for more information.
- To do OLTP-style deduplication or upsert to ClickHouse, refer to [this documentation page](/docs/guides/developer/deduplication#options-for-deduplication). *Note: this is not to be confused with batch deduplication that happens on retries.*

## Troubleshooting[​](#troubleshooting "Direct link to Troubleshooting")

### CANNOT\_READ\_ALL\_DATA[​](#cannot_read_all_data "Direct link to 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](#inserting-data-from-flink-into-clickhouse) 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[​](#low_throughput "Direct link to 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](https://clickhouse.com/docs/merges) 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](#advanced-and-recommended-usage) for batch sizing recommendations.

### I am missing rows in my ClickHouse table[​](#missing_rows "Direct link to 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[​](#contributing-and-support "Direct link to Contributing and support")

If you'd like to contribute to the project or report any issues, we welcome your input!
Visit our [GitHub repository](https://github.com/ClickHouse/flink-connector-clickhouse) to open an issue, suggest
improvements, or submit a pull request.

Contributions are welcome! Please check the [contribution guide](https://github.com/ClickHouse/flink-connector-clickhouse/blob/main/CONTRIBUTING.md) in the repository before starting.
Thank you for helping improve the ClickHouse Flink connector!

- [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)

Was this page helpful?