Flink 连接器 - ClickHouse Documentation

这是由 ClickHouse 官方支持的 Apache Flink Sink Connector。它基于 Flink 的 AsyncSinkBase 和官方 ClickHouse Java 客户端构建。该连接器支持 Apache Flink 的 DataStream API。对 Table API 的支持计划在后续 release 中提供。

要求

Java 11+ (用于 Flink 1.17+) 或 17+ (用于 Flink 2.0+)
Apache Flink 1.17+

Flink 版本兼容性矩阵

该连接器分为两个制品，以同时支持 Flink 1.17+ 和 Flink 2.0+。请选择与所用 Flink 版本对应的制品：

Flink 版本	制品	ClickHouse Java 客户端版本	所需 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+

该连接器尚未针对早于 Flink 1.17.2 的版本进行测试。

安装与设置

作为依赖引入

对于 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"

适用于 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"

下载二进制包

二进制 JAR 的命名规则如下：

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

其中：

flink_version 为 2.0.0 或 1.17
stable_version 是稳定版本制品的发布版本

你可以在 Maven Central 仓库中找到所有已发布的 JAR 文件。

使用 DataStream API

代码示例

假设你想将原始 CSV 数据插入到 ClickHouse：

Java

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

    // 创建一个 ElementConverter
    ElementConverter<String, ClickHousePayload> convertorString = new ClickHouseConvertor<>(String.class);

    // 创建 sink，并使用 `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);

    // 最后，将 DataStream 连接到 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);
}

更多示例和代码片段可在我们的测试中找到：

快速入门示例

我们提供了基于 Maven 的示例，方便您快速开始使用 ClickHouse Sink：

如需更详细的说明，请参阅示例指南

DataStream API 连接选项

ClickHouse 客户端选项

Parameters	Description	Default Value	Required
`url`	完整的 ClickHouse URL	不适用	是
`username`	ClickHouse 数据库用户名	不适用	是
`password`	ClickHouse 数据库密码	不适用	是
`database`	ClickHouse 数据库名称	不适用	是
`table`	ClickHouse 表名	不适用	是
`options`	Java 客户端配置选项的映射	空映射	否
`serverSettings`	ClickHouse 服务端会话设置的映射	空映射	否
`enableJsonSupportAsString`	用于让 ClickHouse 服务端期望 JSON 数据类型采用 JSON 格式 `String` 的服务端设置	true	否

options 和 serverSettings 应以 Map<String, String> 的形式传递给客户端。如果其中任一项为空映射，则分别使用客户端或服务端的默认值。

所有可用的 Java 客户端选项均列在 ClientConfigProperties.java 和此文档页面中。所有可用的服务端会话设置均列在此文档页面中。

例如：

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 选项

以下选项直接来自 Flink 的 AsyncSinkBase：

参数	描述	默认值	必填
`maxBatchSize`	单个批次中可插入的最大记录数	N/A	是
`maxInFlightRequests`	在 sink 开始施加背压之前，允许的最大进行中请求数	N/A	是
`maxBufferedRequests`	在 sink 开始施加背压之前，可在其中缓冲的最大记录数	N/A	是
`maxBatchSizeInBytes`	一个批次允许达到的最大大小 (以字节为单位) 。所有发送的批次都将小于或等于该大小	N/A	是
`maxTimeInBufferMS`	记录在被刷新前可在 sink 中停留的最长时间	N/A	是
`maxRecordSizeInBytes`	sink 可接受的最大记录大小，超过该大小的记录会被自动拒绝	N/A	是

支持的数据类型

下表快速列出了将数据从 Flink 插入 ClickHouse 时的数据类型转换对应关系。

将 Flink 中的数据插入 ClickHouse

Java 类型	ClickHouse 类型	是否支持	序列化方法
`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

注意：

执行日期操作时，必须提供 ZoneId。
执行 decimal 操作时，必须提供精度和标度。
要让 ClickHouse 将 Java String 解析为 JSON，需要在 ClickHouseClientConfig 中启用 enableJsonSupportAsString。
该连接器需要一个 ElementConvertor，用于将输入 DataStream 中的元素映射为 ClickHouse 载荷。为此，连接器提供了 ClickHouseConvertor 和 POJOConvertor，你可以结合上述 DataWriter 序列化方法使用它们来实现这种映射。

支持的输入格式

你可以在此文档页面和 ClickHouseFormat.java 中查看可用的 ClickHouse 输入格式列表。要指定连接器用于将 DataStream 序列化为 ClickHouse 载荷的格式，请使用 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);

默认情况下，如果在 ClickHouseClientConfig 中显式将 setSupportDefault 设置为 true 或 false，连接器将分别使用 RowBinaryWithDefaults 或 RowBinary。

指标

该连接器在 Flink 现有指标的基础上，还额外暴露了以下指标：

指标	描述	类型	状态
`numBytesSend`	请求载荷中发送到 ClickHouse 的总字节数。注意：该指标衡量的是通过网络发送的序列化数据大小，可能与 ClickHouse 的 `system.query_log` 中的 `written_bytes` 不同；后者反映的是数据经过处理后实际写入存储的字节数	计数器	✅
`numRecordSend`	发送到 ClickHouse 的记录总数	计数器	✅
`numRequestSubmitted`	已发送的请求总数 (即实际执行的 flush 次数)	计数器	✅
`numOfDroppedBatches`	因不可重试失败而丢弃的批次总数	计数器	✅
`numOfDroppedRecords`	因不可重试失败而丢弃的记录总数	计数器	✅
`totalBatchRetries`	因可重试失败而进行的批次重试总数	计数器	✅
`writeLatencyHistogram`	成功写入延迟分布的直方图 (毫秒)	直方图	✅
`writeFailureLatencyHistogram`	写入失败延迟分布的直方图 (毫秒)	直方图	✅
`triggeredByMaxBatchSizeCounter`	因达到 `maxBatchSize` 而触发的 flush 总数	计数器	✅
`triggeredByMaxBatchSizeInBytesCounter`	因达到 `maxBatchSizeInBytes` 而触发的 flush 总数	计数器	✅
`triggeredByMaxTimeInBufferMSCounter`	因达到 `maxTimeInBufferMS` 而触发的 flush 总数	计数器	✅
`actualRecordsPerBatch`	实际批次大小分布的直方图	直方图	✅
`actualBytesPerBatch`	每个批次实际字节数分布的直方图	直方图	✅

局限性

该 sink 当前提供至少一次交付保证。实现精确一次语义的相关工作可在此处跟踪。
该 sink 目前尚不支持用于缓冲无法处理记录的死信队列 (DLQ) 。在此之前，连接器会尝试重新 insert 失败的记录；如果仍然失败，则会将其丢弃。此功能可在此处跟踪。
该 sink 目前尚不支持通过 Flink 的 Table API 或 Flink SQL 创建。此功能可在此处跟踪。

ClickHouse 版本兼容性与安全

该连接器通过每日 CI 工作流针对一系列较新的 ClickHouse 版本进行测试，包括 latest 和 head。随着新的 ClickHouse 发行版进入活跃状态，测试版本也会定期更新。有关该连接器每日测试的版本，请参见此处。
有关已知安全漏洞以及如何报告漏洞，请参见 ClickHouse 安全策略。
我们建议持续升级该连接器，以免错过安全修复和新改进。
如果你在迁移过程中遇到问题，请创建 GitHub issue，我们会回复！

高级和推荐用法

为获得最佳性能，请确保你的 DataStream 元素类型不是 Generic 类型——请参阅这篇关于 Flink 类型区分的说明。非 Generic 元素可避免 Kryo 引入的序列化开销，并提升写入 ClickHouse 的吞吐量。
我们建议将 maxBatchSize 设置为至少 1000，理想范围是 10,000 到 100,000。更多信息请参阅这篇关于批量 insert 的指南。
如需在 ClickHouse 中执行 OLTP 风格的去重或 upsert，请参阅此文档页面。注意：不要将其与发生在 retries 时的批次去重混淆。

故障排查

CANNOT_READ_ALL_DATA

可能会发生以下错误：

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)

原因：最常见的情况是，CANNOT_READ_ALL_DATA 错误表示你的 ClickHouse 表 schema 与 Flink 记录 schema 不一致。这通常发生在其中一方以不向后兼容的方式发生变更时。 解决方案：更新 ClickHouse 表 schema 或连接器输入数据类型中的 schema (或同时更新两者) ，使其相互兼容。如有需要，请参阅type mapping，了解如何将 Java 类型映射为 ClickHouse 类型。注意：如果仍有记录在传输过程中，则在重启连接器时需要重置 Flink 状态。

吞吐量低

将数据写入 ClickHouse 时，你可能会发现连接器的吞吐量不会随着作业并行度 (Flink task 数量) 提升而线性扩展。原因：ClickHouse 的后台分片合并过程可能会拖慢 insert。在配置的批次大小过小、连接器 flush 过于频繁，或两者同时存在时，都可能出现这种情况。 解决方案：监控 numRequestSubmitted 和 actualRecordsPerBatch 指标，以帮助判断如何调整批次大小 (maxBatchSize) 以及 flush 频率。另请参阅高级和推荐用法中的批次大小建议。

我的 ClickHouse 表中有行缺失

原因：这些批次被丢弃了，原因可能是发生了不可重试的故障，或者在配置的重试次数内仍无法插入 (可通过 ClickHouseClientConfig.setNumberOfRetries() 设置) 。注意：默认情况下，连接器会在丢弃某个批次之前最多尝试重新插入 3 次。 解决方案：检查 TaskManager 日志和/或堆栈跟踪，以定位根本原因。

贡献与支持

如果您想为该项目贡献力量或报告任何问题，欢迎向我们反馈！请访问我们的 GitHub 仓库提交 issue、提出改进建议，或提交拉取请求。欢迎贡献！开始之前，请先查阅仓库中的贡献指南。感谢您帮助改进 ClickHouse Flink 连接器！

​要求

​Flink 版本兼容性矩阵

​安装与设置

​作为依赖引入

​对于 Flink 2.0+

​适用于 Flink 1.17 及以上版本

​下载二进制包

​使用 DataStream API

​代码示例

​快速入门示例

​DataStream API 连接选项

​ClickHouse 客户端选项

​Sink 选项

​支持的数据类型

​将 Flink 中的数据插入 ClickHouse

​支持的输入格式

​指标

​局限性

​ClickHouse 版本兼容性与安全

​高级和推荐用法

​故障排查

​CANNOT_READ_ALL_DATA

​吞吐量低

​我的 ClickHouse 表中有行缺失

​贡献与支持

要求

Flink 版本兼容性矩阵

安装与设置

作为依赖引入

对于 Flink 2.0+

适用于 Flink 1.17 及以上版本

下载二进制包

使用 DataStream API

代码示例

快速入门示例

DataStream API 连接选项

ClickHouse 客户端选项

Sink 选项

支持的数据类型

将 Flink 中的数据插入 ClickHouse

支持的输入格式

指标

局限性

ClickHouse 版本兼容性与安全

高级和推荐用法

故障排查

CANNOT_READ_ALL_DATA

吞吐量低

我的 ClickHouse 表中有行缺失

贡献与支持