★ FEATURED ARTICLE 做实时数据开发这几年有个绕不开的组合越来越常见Flink SQL 负责实时计算ClickHouse 负责快速分析查询。用 Flink SQL 把数据写进 ClickHouse几乎成了实时数仓、实时报表和指标大屏场景里的标配链路。很多人第一次接触时容易想简单了以为建两张表、写个 INSERT INTO 就能跑通但真正上线后性能瓶颈、写入失败、数据重复、类型不匹配这些问题全冒出来了。这篇文章我就结合自己实际做过的一个某实时数据平台项目把 Flink SQL 写 ClickHouse 的整套流程、参数设计、调优思路和踩坑记录完整梳理一遍。无论你是刚入门的开发还是已经写了半年 SQL 任务但被写入问题折磨过的同行应该都能从中找到可用的参考。这套链路能解决的问题很直接用一套 SQL 语法同时搞定 Kafka 消费、数据加工和 ClickHouse 写入不再需要写一堆 Java 代码去封装 Source/Sink也不用手动维护分布式写入的逻辑。开发效率高了好几倍而且 Flink 官方连接器已经比较成熟配合好参数配置吞吐量和稳定性完全可以满足生产要求。2.1 什么是一对多与多对一写入场景先说清楚为什么要用 Flink SQL 而不是其他方式。在实时数仓里数据通常是分级处理的ODS 层消费 Kafka 原始数据DWD 层做清洗、补齐、维度关联DWS 层做轻量汇总最后落到 ClickHouse 供查询分析。每一层之间的数据流转如果都写 Java 代码开发成本和维护成本都很高。Flink SQL 最直观的价值在于把读 Kafka、处理数据、写 ClickHouse全部声明成一张张逻辑表Everything is a TableSQL 既定义了数据流也定义了数据形态上下游对开发者来说就是一个数据库视角底层实时流的复杂性被框架屏蔽了。那为什么选择 ClickHouse 作为目标存储因为 ClickHouse 是真正的列式存储压缩比高、聚合查询极快特别适合写入多、查询快、按时间范围扫描这类分析场景。实时指标大屏、业务看板、用户行为分析这类应用查询基本都是对某几列做大范围聚合列式存储在 IO 上就有天然优势。相比之下传统行式数据库即使同样能存数据在这种分析负载下性能和成本都差一个量级。而 Flink SQL 的写入方式和 ClickHouse 的特性其实非常匹配Flink 是流式的会持续不断产生小批次数据ClickHouse 的 MergeTree 家族引擎适合高频批量插入每次写入都会在后台做分区合并整体吞吐可以很高。这个组合本质上是一个流计算引擎 OLAP 存储引擎的分工协作。Flink 负责状态管理、窗口计算、精确一次语义ClickHouse 负责把算好的结果可靠地存下来、供上层快速查询。两者各干各的强项。2.2 Flink SQL 工程中为什么用声明式表结构建表语句里最容易被低估的部分其实是 Schema 声明。很多人随便写几个字段就完事结果运行时报类型不匹配。Flink SQL 连接器在启动时会按照声明的 Schema 去读取上游消息、推断下游字段如果类型对不上轻则数据全部落在默认值里重则任务启动即报错。以 Kafka 消息为例即使上游 JSON 里某个字段是字符串形式的数字比如12.5但你希望以DOUBLE类型落地必须在声明字段时给对格式。Flink SQL 有明确的类型体系Jackson 解析出的字符串和 Double 类型不会自动隐式转换。如果你没有在 DDL 中声明相关格式参数数据会直接脏掉或丢弃。所以建表不是随便写几行字段而是把数据格式、类型映射、物理属性全部定死写清楚任务运行才稳。另外声明式表结构还带来一个好处Flink 优化器可以根据这些元信息做下推优化。比如把过滤条件下推到 Kafka 读取端或者在写入 ClickHouse 前裁剪掉不需要的列减少网络 IO 和存储开销。这些优化虽然不在表面功能里但直接影响任务成本。2. 环境选型与参数设计被忽略的坑最深2.1 准备就绪的环境要素在碰 SQL 之前先把环境里的几个关键版本和依赖确认清楚否则后面会非常难受。我以常见的搭建方案为例说明前提是已经有一个完整的 Flink 集群Flink 1.16 或更高版本目标端 ClickHouse 集群也是单分片或多分片均可单副本就够测试生产建议双副本。需要提前准备的依赖包括 Flink 官方发布包中自带的 Kafka connector 和 JDBC connector。ClickHouse 没有官方独立的 Flink connector大家通常走 JDBC 模式。使用 JDBC 意味着需要在部署环境里确保相关的驱动和依赖被正确打包尤其要注意驱动版本和 ClickHouse 服务端版本兼容。如果你用的是 Flink SQL Client 或者 SQL 任务提交平台得确认 classpath 里已经有这些 jar否则任务会直接报找不到类。我自己习惯的做法是先在一个独立的测试目录里搭一个最小可复用的环境用 SQL Client 做语法验证再把调好的 SQL 搬到生产调度平台。这样做的好处是语法问题、依赖问题能在本地花十分钟暴露而不是到生产环境被周期性的运行报告逼着去排查。2.2 核心参数逐个拆解连接器、格式与写入频率控制Flink SQL 写入 ClickHouse 的核心 DDL 可以拆成几段来看。先说连接器参数典型写法如下CREATE TABLE clickhouse_sink_table ( user_id STRING, event_name STRING, event_time TIMESTAMP(3), metric_value DECIMAL(10, 2), PRIMARY KEY (user_id, event_time) NOT ENFORCED ) WITH ( connector jdbc, url jdbc:clickhouse://clickhouse-host:8123/default, table-name event_metrics, username default, password ******, sink.buffer-flush.max-rows 1000, sink.buffer-flush.interval 5s, sink.max-retries 3 );参数少的时候感觉没什么但每个参数背后都是一次取舍。sink.buffer-flush.max-rows控制攒批的行数阈值sink.buffer-flush.interval控制攒批的时间窗口。ClickHouse 本质上不擅长单条高频插入更擅长大批次顺序写。Flink 通过攒批把流式小写入合并成较大的批次可以显著降低写入频率、减少 ClickHouse 的合并压力。但攒批参数也不能盲目调大如果行数设太大在低流量场景下内存里会一直等不到积满的行数数据延迟会变大如果时间间隔设太短每次刷入的数据块又太小ClickHouse 可能会不断产生碎片数据块后台 merge 线程就会繁忙查询时 partition 过多导致性能下降。我实际中的一个经验值峰值吞吐在每秒几千条的场景下把 max-rows 设置为 1000–2000interval 设置为 3–5 秒比较稳。如果任务数据量很小比如每分钟只有百来条那直接把 interval 调低到 1–2 秒也行保证延迟可控别让积压太久。sink.max-retries决定写入失败时重试几次。这个参数要区分失败类型来看如果是网络抖动重试一般都有用但如果是 SQL 执行层面报错比如字段类型不匹配、目标表不存在重试多少次都没用只会阻塞任务。更关键的是重试有可能导致数据重复写入因为 Flink JDBC Sink 默认不具备精确一次的语义。具体到 ClickHouse如果要强一致得配合它的INSERT ... SELECT原子性或者通过去重表引擎来做业务侧去重这些细节后面会专门展开。PRIMARY KEY在 Flink DDL 里只提示不强制执行这就是NOT ENFORCED的含义。它主要用于 Upsert 类场景的语义声明比如你在 JDBC 连接器里配置了 upsert 写入这个主键就是参与判断该更新还是该插入的依据。如果只是 append 写入可以不用设置主键。3. 从 Kafka 到 ClickHouse 的完整实操流程2.1 环境准备3.1 目标表设计ClickHouse 侧先想清楚接上继续展开写 Flink SQL 之前ClickHouse 端的表结构必须提前设计好。很多人在 Flink 里反复调参最终发现性能上不去问题其实出在 ClickHouse 建表语句不合理。我的建议是先根据查询模式设计 CK 表再反向推 Flink 表结构。假设业务场景是一个简单的用户行为分析看板需要按事件时间统计事件数量、GMV 等指标那么 CK 表可以这样建CREATE TABLE default.event_metrics ( user_id String, event_name String, event_time DateTime, metric_value Decimal(18, 2), day Date DEFAULT toDate(event_time) ) ENGINE MergeTree() PARTITION BY toYYYYMM(event_time) ORDER BY (event_name, event_time, user_id)ORDER BY决定了 ClickHouse 内部的稀疏索引排序这直接关系查询性能。如果日常查询总是先按 event_name 过滤、再按时间范围扫描就把 event_name 放最前面。分区键用月份粒度适合这个数据量级和查询模式。如果业务上只需要最近一天的实时数据甚至可以考虑按天分区但要小心分区过多导致查询时打开的文件数膨胀。点击查询表结构的类型映射时注意 ClickHouse 的DateTime和 Flink 的TIMESTAMP(3)之间的关系。Flink 的 TIMESTAMP 会带毫秒精度而 ClickHouse 的 DateTime 只精确到秒写入时会做截断如果业务上对毫秒敏感提前在 SQL 里做处理把不需要的精度裁剪掉。3.2 Flink SQL 建表与数据处理逻辑接下来定义 Flink 端的源表、维表和结果表。源表从 Kafka 读取用户行为日志注意声明 Kafka 的 format 为 JSON并带上json.ignore-parse-errors true这类容错项防止某些脏数据把整个任务拖垮。CREATE TABLE kafka_source ( user_id STRING, event_name STRING, event_time STRING, metric_value DECIMAL(10, 2), proc_time AS PROCTIME() ) WITH ( connector kafka, topic ods_user_behavior, properties.bootstrap.servers kafka-host:9092, properties.group.id flink_ck_demo, scan.startup.mode latest-offset, format json ); CREATE TABLE clickhouse_sink ( user_id STRING, event_name STRING, event_time TIMESTAMP(3), metric_value DECIMAL(10, 2), day STRING ) WITH ( connector jdbc, url jdbc:clickhouse://clickhouse-host:8123/default, table-name event_metrics, username default, password ******, sink.buffer-flush.max-rows 1000, sink.buffer-flush.interval 5s, sink.max-retries 3 );然后一个简单的转换插入就完事了。INSERT INTO clickhouse_sink SELECT user_id, event_name, TO_TIMESTAMP(event_time) AS event_time, metric_value, DATE_FORMAT(TO_TIMESTAMP(event_time), yyyy-MM-dd) AS day FROM kafka_source;这里有个容易出错的地方如果event_time上游是2025-02-14 12:30:45这种字符串格式直接用TO_TIMESTAMP指定格式最稳妥。Flink 的TO_TIMESTAMP默认格式是yyyy-MM-dd HH:mm:ss如果你的字符串不带毫秒可以不加格式参数直接转换。如果带时区务必检查 Flink 任务时区和 ClickHouse 服务端时区是否一致否则写入的时间会整体偏移几个小时查出来的数据和业务时间对不上。3.3 部署与验证在 SQL Client 里依次执行先创建源表、结果表再执行 INSERT。启动后可以观察到 Flink UI 的算子指标Sink 的numRecordsOutPerSecond和currentSendTime可以直观看到吞吐情况。再到 ClickHouse 里执行SELECT count() FROM default.event_metrics; SELECT event_name, sum(metric_value) FROM default.event_metrics GROUP BY event_name;确认数据有条不紊地落入目标表并且聚合结果符合预期。这个链路就算基本跑通了。4. SQL 调优思路与典型坑位规避4.1 并行度设计源、算子、Sink 三板斧并行度是 Flink 性能最直接的杠杆。很多场景下一个 SQL 任务卡在写入环节不是 ClickHouse 的问题而是 Sink 的并发不够。JDBC Sink 的每个并发子任务维护自己的数据库连接和攒批缓冲。如果并行度太低单缓冲攒批慢、单连接吞吐有限如果并行度太高ClickHouse 要同时接收来自几十个并发连接的大量 insert 请求反而可能把 CK 的 merge 线程打满出现 Too many parts 报错。所以我通常建议把 Sink 并行度单独设置不要和 Source 的并行度保持一致。SQL 里可以通过sink.parallelism显式控制INSERT INTO clickhouse_sink SELECT ... FROM kafka_source /* OPTIONS(sink.parallelism 8) */这里补充说明具体能不能用这个 hint 语法要看 Flink 版本一般 1.15 以上支持动态 hint 的写法在 1.14 及之前更常见的做法是在SET语句中预设默认并行度或者将整个 SQL 拆分出来设置执行参数。如果环境不支持 hint就直接设置全局并行度再配置 Sink 算子独立覆盖。并行度的调整不能一次到位最靠谱的做法是先看 Source 端的峰值速率再逐步增大 Sink 并行度同时监控 ClickHouse 端的 Insert 查询耗时、parts 数量、merge 队列积压情况。如果 CK 侧system.parts里的 active parts 数量持续增长说明写入太快、merge 跟不上需要适当降低写入并行度或者把缓冲刷新的批次调更大。4.2 攒批参数与内存平衡攒批参数背后是内存占用和延迟的权衡。Flink JDBC Sink 在攒批时每个并行子任务会维护一个缓冲列表如果 max-rows 设得太大、实际数据量又很小这个缓冲列表会一直空置内存浪费不说数据延迟也莫名其妙变大。我遇到过一个案例某项目把 max-rows 直接设成 200000结果真实日活只有几千条每次要攒很久才触发一次 flush报表延迟达到分钟级。反过来的问题也很典型interval 设成 1s批次只有百来行ClickHouse 根本不买账碎片数据块大量产生。ClickHouse 的 merge 虽然是自动的但如果碎片多到一定程度查询时打开的文件数就会变多性能下降非常明显。合理的方式是根据业务可接受延迟反推假设业务允许 5 秒延迟那就把 interval 设成 5smax-rows 设成该时间窗口内峰值数据量的 1/2 到 2/3。这样既保证批次足够大又不会让内存长期悬空。4.3 数据一致性问题如何缓解重复与丢失Flink 的 JDBC Sink 是一个典型的 at-least-once 写入语义。也就是说当任务发生故障恢复或者网络抖动触发重试时同一个数据版本有可能写入多次。这对基数统计类指标还可以接受但如果下游做的是金额累加或库存扣减类运算就必须严肃对待。可行的缓解方案有三种按成本和实施难度排序第一在 ClickHouse 的 ReplacingMergeTree 引擎配合版本字段做去重。Flink 侧在写入时额外生成一个_version字段一般写法是用NOW()或CURRENT_TIMESTAMP这样 CK 在后台合并时同一个排序键下会保留版本最大的一条记录。注意这种去重是异步合并的查询时可能临时看到重复需要配合FINAL关键字或者在查询语句里做聚合去重实时性要求高的场景要谨慎使用。第二在 Flink SQL 里通过INSERT OVERWRITE或主键更新的思路来覆盖目标分区。这适合定时批处理而不是完全实时的场景相当于用离线的方式周期性重刷牺牲部分实时性换取准确性。第三尽量在源端或中间层做好数据唯一性保障比如用 Kafka 消息自带的主键字段作为 CK 主键从源头避免重复。没有万金油的方案每个业务对一致性的容忍度不同。如果业务绝对不能容忍重复建议在 Flink 侧使用自定义 Sink 配合幂等写入不过这已经超出普通 SQL 能解决的问题本文不展开。4.4 类型映射与序列化细节类型映射是另一个高频翻车区。Flink 的DECIMAL(10, 2)和 ClickHouse 的Decimal(18, 2)在精度和范围上的声明如果不一致写入时可能直接报 Numeric overflow。常见的错误是 Flink 里定义为DECIMAL(10, 2)但实际业务数据最大能到 20 位整数溢出的数据要么写入报错要么被截断成不可用的值。我的建议是在 Flink DDL 里尽量把 Decimal 精度定义得比预估最大值再宽一档比如预估最多 10 位整数就按DECIMAL(20, 4)定义存储成本不算高但能少很多不必要的麻烦。字符串类型也要注意空串和 NULL 的区别。JSON 序列化时字段缺失、字段值为 null、字段值为空字符串三者转换到 ClickHouse 的结果可能完全不同。如果不做处理下游查询结果会出现莫名的空洞。一个稳妥的办法是在源表 DDL 里声明json.fail-on-missing-field false并在查询里统一用COALESCE(field, )或IFNULL处理可能为空的字段。时间类型方面Flink 的TIMESTAMP_LTZ与 ClickHouse 的DateTime/DateTime64之间也要注意时区。强烈建议在 Flink 配置里统一设置table.local-time-zone并在 ClickHouse 服务端确认timezone参数让两边的墙上时间保持一致。否则你的凌晨数据可能落到前一天排查起来非常崩溃。5. 常见问题与排查技巧实录5.1 报错“Unexpected char”或解析失败现象任务运行后Sink 侧持续报错或者 ClickHouse 表中出现大量空字符串、默认值。原因上游 JSON 数据里有脏数据比如字段值不是合法 JSON 字符串或者字段类型和声明的类型不匹配。排查方法先看 Flink UI 的 TaskManager 日志找到第一次解析失败的样例数据结合样例反推是数据 format 问题还是 Schema 映射问题。在 DDL 里加json.ignore-parse-errors true可以先让任务跑起来但不要长期依赖它因为忽略解析错误意味着这些数据会静默丢失对数据完整性有影响。更该做的是把上游数据质量治理好否则报表缺数了都不知道。5.2 报错“ClickHouseException: Timeout exceeded”现象写入 ClickHouse 时偶发超时任务重启后恢复但过一会儿又开始超时。原因常见是 ClickHouse 服务端 merge 线程繁忙、连接数过多或者单个 insert 批次过大导致执行耗时超过 Flink 端超时配置。排查方向降低 Sink 并行度、减小单批次行数、增大sink.buffer-flush.interval拉长刷入周期同时检查 ClickHouse 的system.merges视图看是否有大量长期未结束的 merge 任务。如果 ClickHouse 端长期处于 merge 繁忙可能还要调整它的background_pool_size和max_threads参数但这些是 DBA 层面的优化对 Flink 使用者来说更好的策略是控制写入节奏别让系统持续过载。5.3 数据重复如何定位现象同一 user_id 在同一事件时间出现多条记录或者金额总和偏高。原因任务发生过重启JDBC Sink 在 at-least-once 语义下把同一批数据重复写入了也可能是上游 Kafka 本身存在重复消息。定位方法在 Flink UI 看任务重启次数和失败时间点对应用例数据查询 ClickHouse 的时间范围确认重复数据是否集中在重启前后。如果重复确实来自 Flink 侧按前面提到的方式引入版本字段和 ReplacingMergeTree 去重如果来自 Kafka 上游则要在源端解决。6. 针对不同数据量级的参数建议参考不同规模的场景参数设计差别很大。这里给出几个方向的建议供参考具体还要以实测为准场景数据峰值Sink 并行度max-rowsintervalClickHouse 分区粒度轻量报表每秒几十条12005s按月常规业务看板每秒几千条4~810005s按月或按周高吞吐大屏每秒十万级以上16~325000~100003s按天高吞吐场景要注意并行度上去之后ClickHouse 的写入并发也会增加这时最好协调 ClickHouse 服务端提前调大max_insert_threads并确保网络带宽充足。我自己碰到过一个有意思的案例某次吞吐一直上不去最后排查发现瓶颈不在 Flink 也不在 ClickHouse而是 Kafka 到 Flink 的消费端反压把 Source 并行度从 4 提到 8 立刻就好了。所以定位性能问题时要有一个全局的视角不要只盯着目标端看。再补一个实际经验ClickHouse 的写入速度对批次大小非常敏感。相同数据量下1000 行一批的写入耗时可能是 100 行一批的 1/5 甚至更低。这不是 ClickHouse 单次插入本身慢而是插入次数太多会产生大量日志记录和系统调用开销。所以参数调整的核心思想永远是只要延迟能接受尽量攒大块、少插入。7. 写在最后一点实战体会做数据开发的人大概都有一个体会花一小时把 SQL 写通很容易但花三天把性能和稳定性调到能上生产才是真正值钱的工作。用 Flink SQL 写 ClickHouse 这套链路表面上是一段建表语句加一段 INSERT但真实生产环境的挑战都在表层语法之下。我个人经验里最深刻的几条第一建表 DDL 里的每个字段都值得认真对待类型、格式、时区、空值策略这些细节决定了数据质量的上限。第二攒批参数不是照抄一个模板就完事一定要结合自己的峰值流量和业务延迟要求去推导并且上线后持续观察。第三遇到写入性能问题先别急着怪 ClickHouse从 Source 并行度、Sink 并行度、批次大小、CK 分区粒度四个维度逐一排查绝大部分问题都能在这一层解决。另外还有一个容易被忽略的小技巧在 ClickHouse 的查询端建一个和 Sink 表结构几乎相同的消费视图用ORDER BY字段相同但查询字段裁剪的方式去读数据能明显减少查询时的 IO 开销。这是行式数据库思维带到列式存储里的一个直观误区很多人刚开始没意识到白白浪费了 ClickHouse 的列裁剪能力。如果后续要做更复杂的实时数仓链路比如多流 join 后写入宽表、支持更新删除的 KV 场景这套 Flink SQL 加 ClickHouse 的思路依然适用只需要在表引擎和写入语义上做更多适配。希望这篇文章能帮你少踩几个坑把时间花在更有价值的数据分析上。
阅读完成 · 觉得有帮助？
