ClickHouse与Flink实时集成：实现Flink SQL写入ClickHouse的Exactly-Once方案

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

ClickHouse与Flink实时集成：实现Flink SQL写入ClickHouse的Exactly-Once方案

在实现Flink SQL写入ClickHouse的Exactly-Once方案前，首先需要准备好运行环境并理解整体架构设计。Flink作为流处理框架，与ClickHouse这一高性能列式数据库结合，可实现实时数据的高效处理与存储。

环境准备包括：

• 安装部署ClickHouse集群，确保支持事务引擎（如MergeTree系列引擎）
• 配置Flink集群，版本建议1.13+以获得更好的ClickHouse集成支持
• 添加必要的连接器依赖（flink-connector-clickhouse）

架构设计上，我们采用以下模式：

• Flink作业从Kafka等消息队列消费数据
• 使用Flink的Checkpoint机制保存状态
• 将状态快照与数据变更一同写入ClickHouse
• 利用ClickHouse的事务支持保证数据一致性

Exactly-Once语义确保每条数据仅被处理一次，即使在发生故障的情况下也不会丢失或重复处理。在Flink与ClickHouse集成中，主要通过以下机制实现：

2.1 Flink Checkpoint机制

Flink通过分布式快照（Checkpoint）机制记录应用状态，并结合Barrier传播确保处理顺序。在启用Checkpoint后，Flink会在指定时间间隔生成Checkpoint点，这些点记录了所有算子的状态。

// 启用Checkpoint，设置间隔为1分钟
env.enableCheckpointing(60000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(600000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

2.2 ClickHouse事务支持

从20.4版本开始，ClickHouse原生支持事务操作，通过INSERT语句与事务控制可以保证数据写入的原子性。

— 开启事务
BEGIN TRANSACTION;
— 执行插入操作
INSERT INTO table_name VALUES (…), (…), (…);
— 提交事务
COMMIT;

2.3 两阶段提交协议

为实现Exactly-Once，我们需要两阶段提交协议协调Flink与ClickHouse：

下面详细介绍实现Exactly-Once写入ClickHouse的具体步骤：

3.1 创建ClickHouse目标表

首先需要在ClickHouse中创建支持事务的目标表：

CREATE TABLE target_table (
id UInt64,
data String,
timestamp DateTime,
event_time DateTime64(3)
) ENGINE = MergeTree()
ORDER BY (id, timestamp)
SETTINGS index_granularity = 8192;

3.2 配置Flink SQL连接器

// 创建ClickHouse连接参数
Properties properties = new Properties();
properties.setProperty("connector", "clickhouse");
properties.setProperty("url", "jdbc:clickhouse://host:port/database");
properties.setProperty("table", "target_table");
properties.setProperty("username", "default");
properties.setProperty("password", "");
properties.setProperty("write.flush.max.size", "80m");
properties.setProperty("write.flush.queue.size", "1");
properties.setProperty("write.batch.size", "80m");
properties.setProperty("write.max.retries", "3");
properties.setProperty("write.ignore.exceptions", "true");
properties.setProperty("lookup.cache.max-rows", "10000");
properties.setProperty("lookup.cache.ttl", "10min");
properties.setProperty("lookup.max-retries", "3");
// 创建表
tableEnv.executeSql(
"CREATE TABLE flink_clickhouse (" +
" id BIGINT," +
" data STRING," +
" event_time TIMESTAMP(3)," +
" watermark FOR event_time AS event_time – INTERVAL '5' SECOND" +
") WITH (" +
properties +
")"
);

3.3 实现Exactly-Once写入

// 执行Flink SQL作业
tableEnv.executeSql(
"INSERT INTO flink_clickhouse " +
"SELECT id, data, event_time " +
"FROM kafka_table"
);

3.4 监控与故障恢复

作业运行过程中，需要监控Checkpoint状态和ClickHouse写入情况。当作业失败重启时，Flink会从最近的Checkpoint恢复状态，配合ClickHouse的事务机制保证Exactly-Once语义。

4.1 批处理优化

将小批数据聚合为大批量写入，减少网络开销：

// 设置批处理大小
SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '1s';
SET 'table.exec.mini-batch.size' = '10000';

4.2 并行度调整

根据ClickHouse集群性能，合理设置Flink作业并行度：

// 设置并行度
SET 'parallelism.default' = '4';

4.3 表引擎选择

根据业务特点选择合适的ClickHouse表引擎：

• 高频写入场景：使用ReplacingMergeTree或SummingMergeTree
• 时序数据场景：使用TimeSeriesMergeTree

4.4 故障重试配置

// 设置重试策略
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, // 重试次数
Time.seconds(10) // 重试间隔
));

下面是mermaid流程图，展示Exactly-Once写入的流程：

#publish-mermaid-0-1789609772955{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#publish-mermaid-0-1789609772955 .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#publish-mermaid-0-1789609772955 .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#publish-mermaid-0-1789609772955 .error-icon{fill:#552222;}#publish-mermaid-0-1789609772955 .error-text{fill:#552222;stroke:#552222;}#publish-mermaid-0-1789609772955 .edge-thickness-normal{stroke-width:1px;}#publish-mermaid-0-1789609772955 .edge-thickness-thick{stroke-width:3.5px;}#publish-mermaid-0-1789609772955 .edge-pattern-solid{stroke-dasharray:0;}#publish-mermaid-0-1789609772955 .edge-thickness-invisible{stroke-width:0;fill:none;}#publish-mermaid-0-1789609772955 .edge-pattern-dashed{stroke-dasharray:3;}#publish-mermaid-0-1789609772955 .edge-pattern-dotted{stroke-dasharray:2;}#publish-mermaid-0-1789609772955 .marker{fill:#333333;stroke:#333333;}#publish-mermaid-0-1789609772955 .marker.cross{stroke:#333333;}#publish-mermaid-0-1789609772955 svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#publish-mermaid-0-1789609772955 p{margin:0;}#publish-mermaid-0-1789609772955 .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#publish-mermaid-0-1789609772955 .cluster-label text{fill:#333;}#publish-mermaid-0-1789609772955 .cluster-label span{color:#333;}#publish-mermaid-0-1789609772955 .cluster-label span p{background-color:transparent;}#publish-mermaid-0-1789609772955 .label text,#publish-mermaid-0-1789609772955 span{fill:#333;color:#333;}#publish-mermaid-0-1789609772955 .node rect,#publish-mermaid-0-1789609772955 .node circle,#publish-mermaid-0-1789609772955 .node ellipse,#publish-mermaid-0-1789609772955 .node polygon,#publish-mermaid-0-1789609772955 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#publish-mermaid-0-1789609772955 .rough-node .label text,#publish-mermaid-0-1789609772955 .node .label text,#publish-mermaid-0-1789609772955 .image-shape .label,#publish-mermaid-0-1789609772955 .icon-shape .label{text-anchor:middle;}#publish-mermaid-0-1789609772955 .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#publish-mermaid-0-1789609772955 .rough-node .label,#publish-mermaid-0-1789609772955 .node .label,#publish-mermaid-0-1789609772955 .image-shape .label,#publish-mermaid-0-1789609772955 .icon-shape .label{text-align:center;}#publish-mermaid-0-1789609772955 .node.clickable{cursor:pointer;}#publish-mermaid-0-1789609772955 .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#publish-mermaid-0-1789609772955 .arrowheadPath{fill:#333333;}#publish-mermaid-0-1789609772955 .edgePath .path{stroke:#333333;stroke-width:1px;}#publish-mermaid-0-1789609772955 .flowchart-link{stroke:#333333;fill:none;}#publish-mermaid-0-1789609772955 .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#publish-mermaid-0-1789609772955 .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#publish-mermaid-0-1789609772955 .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#publish-mermaid-0-1789609772955 .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#publish-mermaid-0-1789609772955 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#publish-mermaid-0-1789609772955 .cluster text{fill:#333;}#publish-mermaid-0-1789609772955 .cluster span{color:#333;}#publish-mermaid-0-1789609772955 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#publish-mermaid-0-1789609772955 .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#publish-mermaid-0-1789609772955 rect.text{fill:none;stroke-width:0;}#publish-mermaid-0-1789609772955 .icon-shape,#publish-mermaid-0-1789609772955 .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#publish-mermaid-0-1789609772955 .icon-shape p,#publish-mermaid-0-1789609772955 .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#publish-mermaid-0-1789609772955 .icon-shape .label rect,#publish-mermaid-0-1789609772955 .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#publish-mermaid-0-1789609772955 .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#publish-mermaid-0-1789609772955 .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#publish-mermaid-0-1789609772955 .node .neo-node{stroke:#9370DB;}#publish-mermaid-0-1789609772955 [data-look=\”neo\”].node rect,#publish-mermaid-0-1789609772955 [data-look=\”neo\”].cluster rect,#publish-mermaid-0-1789609772955 [data-look=\”neo\”].node polygon{stroke:#9370DB;filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789609772955 [data-look=\”neo\”].swimlane.cluster rect{filter:none;}#publish-mermaid-0-1789609772955 [data-look=\”neo\”].node path{stroke:#9370DB;stroke-width:1px;}#publish-mermaid-0-1789609772955 [data-look=\”neo\”].node .outer-path{filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789609772955 [data-look=\”neo\”].node .neo-line path{stroke:#9370DB;filter:none;}#publish-mermaid-0-1789609772955 [data-look=\”neo\”].node circle{stroke:#9370DB;filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789609772955 [data-look=\”neo\”].node circle .state-start{fill:#000000;}#publish-mermaid-0-1789609772955 [data-look=\”neo\”].icon-shape .icon{fill:#9370DB;filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789609772955 [data-look=\”neo\”].icon-shape .icon-neo path{stroke:#9370DB;filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789609772955 :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

Flink从Kafka消费数据

Flink处理数据并生成Checkpoint

将写入操作发送到ClickHouse临时表

ClickHouse写入临时表成功

Flink记录Checkpoint完成状态

提交ClickHouse事务，移动数据到目标表

生成下一个Checkpoint

故障发生

从最近的Checkpoint恢复

从ClickHouse回滚未完成的事务

重新消费从Checkpoint开始的偏移量

ClickHouse表引擎选择对比：

最小示例代码：

public class ClickHouseExactlyOnceExample {
public static void main(String[] args) {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 启用Checkpoint
env.enableCheckpointing(60000); // 1分钟
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 创建ClickHouse表
String createTableSql = "CREATE TABLE clickhouse_table (" +
"id BIGINT, " +
"name STRING, " +
"timestamp TIMESTAMP(3)" +
") WITH (" +
" 'connector' = 'clickhouse', " +
" 'url' = 'jdbc:clickhouse://localhost:9000/default', " +
" 'table' = 'exactly_once_test', " +
" 'username' = 'default', " +
" 'password' = '', " +
" 'write.flush.max.size' = '80m', " +
" 'write.max.retries' = '3'" +
")";
tableEnv.executeSql(createTableSql);

// 模拟数据源
DataStream<Tuple3<Long, String, Long>> dataStream = env.addSource(new ClickHouseSource());

// 转换为Table
Table table = tableEnv.fromDataStream(dataStream, $("id"), $("name"), $("timestamp"));

// 写入ClickHouse
table.executeInsert("clickhouse_table");

// 执行作业
try {
env.execute("ClickHouse Exactly-Once Example");
} catch (Exception e) {
e.printStackTrace();
}
}

// 模拟数据源
public static class ClickHouseSource extends RichSourceFunction<Tuple3<Long, String, Long>> {
private volatile boolean isRunning = true;

@Override
public void run(SourceContext<Tuple3<Long, String, Long>> ctx) throws Exception {
long count = 0;
while (isRunning && count < 1000) {
Tuple3<Long, String, Long> record = new Tuple3<>(count, "test_" + count, System.currentTimeMillis());
ctx.collect(record);
count++;
Thread.sleep(1000);
}
}

@Override
public void cancel() {
isRunning = false;
}
}
}

注意事项：