Flink Connector

## 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")

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Java Type ClickHouse Type Supported Serialization Method|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `byte`/`Byte` `Int8` ✅ `DataWriter.writeInt8`| `short`/`Short` `Int16` ✅ `DataWriter.writeInt16`| `int`/`Integer` `Int32` ✅ `DataWriter.writeInt32`| `long`/`Long` `Int64` ✅ `DataWriter.writeInt64`| `BigInteger` `Int128` ✅ `DataWriter.writeInt128`| `BigInteger` `Int256` ✅ `DataWriter.writeInt256`| `short`/`Short` `UInt8` ✅ `DataWriter.writeUInt8`| `int`/`Integer` `UInt8` ✅ `DataWriter.writeUInt8` | `int`/`Integer` `UInt16` ✅ `DataWriter.writeUInt16`| `long`/`Long` `UInt32` ✅ `DataWriter.writeUInt32`| `long`/`Long` `UInt64` ✅ `DataWriter.writeUInt64`| `BigInteger` `UInt64` ✅ `DataWriter.writeUInt64`| `BigInteger` `UInt128` ✅ `DataWriter.writeUInt128`| `BigInteger` `UInt256` ✅ `DataWriter.writeUInt256`| `BigDecimal` `Decimal` ✅ `DataWriter.writeDecimal`| `BigDecimal` `Decimal32` ✅ `DataWriter.writeDecimal`| `BigDecimal` `Decimal64` ✅ `DataWriter.writeDecimal`| `BigDecimal` `Decimal128` ✅ `DataWriter.writeDecimal`| `BigDecimal` `Decimal256` ✅ `DataWriter.writeDecimal`| `float`/`Float` `Float` ✅ `DataWriter.writeFloat32`| `double`/`Double` `Double` ✅ `DataWriter.writeFloat64`| `boolean`/`Boolean` `Boolean` ✅ `DataWriter.writeBoolean`| `String` `String` ✅ `DataWriter.writeString`| `String` `FixedString` ✅ `DataWriter.writeFixedString`| `LocalDate` `Date` ✅ `DataWriter.writeDate`| `LocalDate` `Date32` ✅ `DataWriter.writeDate32`| `LocalDateTime` `DateTime` ✅ `DataWriter.writeDateTime`| `ZonedDateTime` `DateTime` ✅ `DataWriter.writeDateTime`| `LocalDateTime` `DateTime64` ✅ `DataWriter.writeDateTime64`| `ZonedDateTime` `DateTime64` ✅ `DataWriter.writeDateTime64`| `int`/`Integer` `Time` ❌ N/A|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `long`/`Long` `Time64` ❌ N/A|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `byte`/`Byte` `Enum8` ✅ `DataWriter.writeInt8`| `int`/`Integer` `Enum16` ✅ `DataWriter.writeInt16`| `java.util.UUID` `UUID` ✅ `DataWriter.writeIntUUID`| `String` `JSON` ✅ `DataWriter.writeJSON`| `Array<Type>` `Array<Type>` ✅ `DataWriter.writeArray`| `Map<K,V>` `Map<K,V>` ✅ `DataWriter.writeMap`| `Tuple<Type,..>` `Tuple<T1,T2,..>` ✅ `DataWriter.writeTuple`| `Object` `Variant` ❌ N/A | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | |

Notes:

- A `ZoneId` must be provided when performing date operations.
- [Precision and scale](https://clickhouse.com/docs/sql-reference/data-types/decimal#decimal-value-ranges) 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[​](#supported-input-formats "Direct link to Supported input formats")

You can find the list of available ClickHouse input formats in [this documentation page](https://clickhouse.com/docs/interfaces/formats#formats-overview) and [ClickHouseFormat.java](https://github.com/ClickHouse/clickhouse-java/blob/main/clickhouse-data/src/main/java/com/clickhouse/data/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,
        clickHouseClientConfig
);
csvSink.setClickHouseFormat(ClickHouseFormat.CSV);
```

Note