凌晨两点被电话叫醒，打开监控面板发现实时看板的 GMV 数字比半小时前还低，这种场景只要做过实时链路的团队基本都遇到过。这条从 Kafka 到 Flink 再到 StarRocks 的链路，看起来只是“读—算—写”三步，但真正把它跑稳、跑出秒级延迟、保证数据不重不丢，靠的是对每个环节参数的较真和对边界情况的预判。这篇内容讲的就是这条链路怎么搭、参数怎么调、坑怎么避。

关键词围绕Flink 读取 Kafka 数据、数据处理、写入 StarRocks展开，它解决的核心问题是：把业务库或埋点系统实时投递到消息队列的原始数据，在 Flink 里完成清洗、补维、聚合等处理后，以可控的延迟和明确的语义一致性落到 StarRocks，支撑实时报表、漏斗分析、明细点查、实时风控看板这类对“新鲜度”敏感的场景。适合已经写过批任务、想转实时链路的开发者，也适合手上有实时链路但性能不稳、数据对不上的同学对着排查。全文所有参数都给出取值理由，代码可以直接改成自己的字段跑起来。

1. 链路整体设计与选型思路拆解

在动手写第一行代码之前，把链路的形状画清楚，比上来就调 API 更重要。Flink 在这条链路里扮演的是“流式加工厂”，上游 Kafka 是原料仓库，下游 StarRocks 是成品展厅。任何一个环节的模型设计错了，后面靠调参数是救不回来的。这一章先把选型逻辑讲透，再说拓扑怎么分层。

1.1 为什么是 Kafka + Flink + StarRocks 这个组合

先说 Kafka 为什么出现在上游。实时链路最怕的不是数据量大，而是“消费端写错了还能回头重来”。Kafka 的多副本加 offset 机制给了这条链路最基础的后悔药：Flink 作业重启、逻辑改版、下游表重建，只要能回到某个 offset，数据就能重新流一遍。这点在业务口径频繁调整的早期阶段价值极高，很多团队吃过“数据只能消费一次、清算逻辑写错就得等第二天”的亏。

Flink 的价值在于它是少数把“事件时间语义”和“状态管理”都做成第一公民的流引擎。清洗、去重、窗口聚合、维表关联这些操作，如果换成自研消费者去拼，光乱序数据和迟到数据的处理就够写两个月。Flink 的 watermark 机制配合 allowed lateness，让“数据晚到 30 秒依然算进正确窗口”这类需求变成几行配置。

下游选 StarRocks，关键看重的是它同时吃两种负载：高并发点查和聚合分析。实时看板通常是一边写一边查，且查询模式多为大宽表上的条件过滤加聚合。StarRocks 的主键模型支持按主键 Upsert，天然适配 Flink 的“至少一次”写入——重复写不会产生重复行，只会让行被覆盖成相同值。这比在 Flink 侧维护大状态去重省了太多资源。

三者叠加，形成了一个“可回放、可容错、可幂等”的组合，这是这条链路能长期稳定运行的底层原因，而不是某个单点的性能优势。

1.2 数据流转的整体拓扑与分层设计

我习惯把这条链路拆成四层来看，每一层的职责边界清晰之后，排障会快很多：

举一个常见的分层落地：ODS 层的 Kafka topic 存放原始 JSON 事件，字段多且不规范；Flink 作业先做字段裁剪和类型转换，把脏数据分流到侧输出流；主流程做维表关联补齐商品分类和渠道信息；最后按业务粒度写入 StarRocks 的明细宽表。看板直接查宽表，不再做多表 join，查询延迟能压到几百毫秒。

这里有个容易被忽略的设计点：不要在 Flink 里做过重的聚合再写 StarRocks。很多同学习惯把 Flink 当作唯一的计算引擎，做完全部聚合只写结果。一旦口径改了，历史数据要重算，就得回放很长一段 Kafka 数据，成本高、时间长。更实用的做法是明细宽表与聚合表并存，聚合口径变更时直接用 StarRocks 的物化视图或异步物化视图重刷，比重跑 Flink 作业快一个量级。

1.3 版本与依赖矩阵

版本错配是新人踩的第一个坑，尤其是 Flink 的小版本和 connector 的兼容关系。

注意：flink-connector-starrocks 的 groupId 是 com.starrocks，artifact 名是 flink-connector-starrocks，别和社区早期的 starrocks-flink-connector 搞混，两者参数不完全一致。上线前一定要拿目标 Flink 版本实际编译一次，光看文档版本号是不够的。

2. 环境准备与核心依赖配置

配置环节看着枯燥，但这条链路上 80% 的“玄学问题”都源于依赖冲突或者建表模型选错。这一章把依赖树和建表语句逐项过一遍，给出可以直接抄的配置。

2.1 版本对齐这件事比想象中重要

Flink 的 connector 体系有一个明显特征：大版本之间 API 不兼容，小版本之间有类冲突。比如作业里同时出现 flink-connector-kafka 的两份不同版本，常见现象是作业能提交、能跑起来，但在 checkpoint 恢复到某个点时抛NoSuchMethodError，日志里看不出任何业务原因。

我的做法是在 pom 里显式声明 Flink 核心依赖为provided，把所有 connector 的版本集中放在dependencyManagement里统一管理，禁止子模块自行指定版本。同时用mvn dependency:tree输出一份依赖树存档，出问题时第一时间对比。

<properties> <flink.version>1.17.2</flink.version> <scala.binary.version>2.12</scala.binary.version> <starrocks.connector.version>1.2.8</starrocks.connector.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>com.starrocks</groupId> <artifactId>flink-connector-starrocks</artifactId> <version>${starrocks.connector.version}</version> </dependency> </dependencies>

打包时用 shade 插件把 connector 的依赖一起打进去，但要把 Flink 核心包排除掉，否则集群上会出现类加载冲突。一个细节：shade 之后一定要检查META-INF/services下的 SPI 文件有没有被合并覆盖，Kafka 的序列化器和 Flink 的 Factory 都依赖这个目录，文件丢了会报“找不到 connector factory”这类看起来莫名其妙的错误。

2.2 StarRocks 侧建表：模型选择与分区分桶

建表是整条链路的地基。选错的典型表现是：数据写进去了，查询慢得没法看；或者写入报too many versions，越跑越卡。

先看模型选择。StarRocks 有三种表模型，对应三种写入语义：

• 明细模型（Duplicate）：不合并、不去重，适合只追加的日志类数据，写入吞吐最高。
• 主键模型（Primary Key）：按主键 Upsert，适合需要更新维度和修正数据的场景，是 Flink 写入的首选。
• 聚合模型（Aggregate）：按聚合函数合并，适合预聚合指标表，但只支持有限几种聚合方式。

实时链路里，明细宽表优先主键模型。主键模型虽然在写入时要维护主键索引，但它带来的好处是 Flink 侧不用做任何去重，重复数据写两次结果依然正确，作业的语义从“精确一次”降级到“至少一次”也能接受，容错压力小很多。

CREATE TABLE dws_order_detail ( order_id BIGINT NOT NULL COMMENT '订单ID', user_id BIGINT NOT NULL COMMENT '用户ID', channel_code VARCHAR(32) COMMENT '渠道编码', category_id INT COMMENT '类目ID', pay_amount DECIMAL(18,2) COMMENT '实付金额', order_status VARCHAR(16) COMMENT '订单状态', event_time DATETIME COMMENT '事件时间', dt DATE NOT NULL COMMENT '分区日期' ) PRIMARY KEY (order_id, dt) DISTRIBUTED BY HASH(order_id) BUCKETS 32 PARTITION BY RANGE(dt) ( START ("2024-01-01") END ("2025-01-01") EVERY (INTERVAL 1 DAY) ) PROPERTIES ( "replication_num" = "3", "enable_persistent_index" = "true", "replicated_storage" = "true" );

几个参数值得展开说。主键必须包含分区列，所以这里主键是(order_id, dt)组合，这是 StarRocks 的硬性约束，很多同学把dt漏掉导致建表失败。BUCKETS 32的依据是单分区数据量：一般建议单 bucket 数据量在 1GB 以内，单 tablet 压缩后 500MB 到 1GB 比较舒服，如果日增数据是 20GB 左右，32 个桶刚好。桶数不是越多越好，桶太多会增加元数据压力和写入的小文件数，反而拖慢导入。

enable_persistent_index在主键模型上是必开的，尤其是内存有限的机器。不开的话主键索引全放内存，数据量一上来直接 OOM。开了之后索引落盘，内存占用会降下来，代价是写入时多一次磁盘 IO，实测在 SSD 上影响可以忽略。

注意：分区方式是START ... END ... EVERY的动态创建，只覆盖到 2025-01-01，超过这个范围的数据导入会报“no partition for this value”。要么把 END 设得更远，要么使用动态分区"dynamic_partition.enable" = "true"让 StarRocks 自动建分区，后者更省心。

2.3 连接与权限的最小化配置

创建专用账号，只给目标库表的 INSERT 权限和必要的 SELECT，别直接用 root。Stream Load 走的是 HTTP 接口，账号密码会出现在请求头里，虽然走内网，仍然建议单独建号，方便审计和轮换。

CREATE USER 'flink_writer'@'%' IDENTIFIED BY 'replace_with_strong_pwd'; GRANT INSERT, SELECT ON dws.dws_order_detail TO 'flink_writer'@'%'; GRANT SELECT ON dim.product_info TO 'flink_writer'@'%';

Flink 做维表关联时需要查询维表，所以 SELECT 权限要给。另外 Stream Load 需要访问 FE 的 HTTP 端口（默认 8030），而 JDBC 走 9030，两个地址在配置里是分开的：load-url填 FE 的 8030，jdbc-url填 9030。这两个地址填反了是新手最常见的错误，报错信息是 HTTP 405 或者连接超时，看起来完全不像地址问题。

3. Kafka 数据读取与反序列化实操

Kafka Source 的参数决定了作业的启动行为和长期稳定性。这一章把每个参数为什么这么设讲清楚，重点说反序列化和启动消费位点这两块最容易出问题的地方。

3.1 Source 参数逐项拆解

KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("kafka-broker-1:9092,kafka-broker-2:9092") .setTopics("ods_order_event") .setGroupId("flink-order-dws-v3") .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST)) .setDeserializer(new SimpleStringSchema()) .setProperty("partition.discovery.interval.ms", "60000") .setProperty("fetch.max.bytes", "52428800") .setProperty("max.partition.fetch.bytes", "1048576") .build();

setGroupId这个名字要带上版本号或者业务标识，比如flink-order-dws-v3。原因是一次逻辑大改后，如果沿用旧的 group id，作业会从上次提交的 offset 继续消费，而新的处理逻辑可能要求从头回放。带上版本号，新作业自然从头开始，旧作业的 offset 也不会被污染。

setStartingOffsets有三种常用策略，选哪个取决于场景：

关键点在于开启 checkpoint 之后，Kafka 的消费位点由 Flink 管理，而不是 consumer group 的自动提交。所以enable.auto.commit应该设成 false，否则会出现两套位点互相打架的情况：Flink 从 checkpoint 恢复到了一小时前的位置，但 group 的自动提交已经跑到了当前最新，重启的瞬间作业会跳过这一小时的数据。这个坑非常隐蔽，因为作业看起来是正常运行的，只有对数据量的时候才会发现少了。

partition.discovery.interval.ms设成 60 秒，意味着上游新增分区能在一分钟内被发现并分配到消费者。这个值不要设得太小，每次发现动作都要访问元数据服务，太频繁会给集群带来不必要的压力。

3.2 反序列化与脏数据处理

直接SimpleStringSchema只适合消息是纯字符串、格式完全可控的情况。线上更常见的场景是消息体里掺杂了空串、截断的 JSON、字段类型漂移的老版本数据。这时候要么在反序列化阶段就把脏数据分出去，要么在 map 阶段做过滤，我更推荐前者。

一种实用的做法是自己实现KafkaRecordDeserializationSchema，解析失败时不抛异常，而是返回一个带标记的包装对象：

public class SafeJsonDeserializer implements KafkaRecordDeserializationSchema<RawEvent> { private transient ObjectMapper mapper; @Override public void open(DeserializationSchema.InitializationContext ctx) { mapper = new ObjectMapper(); mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); } @Override public void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<RawEvent> out) { if (record.value() == null) { return; } try { out.collect(RawEvent.ok(mapper.readTree(record.value()))); } catch (Exception e) { out.collect(RawEvent.dirty(record.topic(), record.partition(), record.offset(), e.getMessage())); } } @Override public TypeInformation<RawEvent> getProducedType() { return TypeInformation.of(RawEvent.class); } }

这样一来，脏数据不会让整个作业失败，而是进入侧输出流单独落一张表，运维同学可以定期看脏数据比例。实践中这个比例很有价值：如果某天从 0.1% 跳到了 5%，基本可以确定上游发版改了字段格式。

提示：FAIL_ON_UNKNOWN_PROPERTIES一定要关掉。上游加字段是很常见的操作，开启严格校验会导致作业在对方发版时直接挂掉，而这个问题往往在半夜才暴露。关掉之后新增字段被忽略，老作业继续跑，新字段等作业升级后再消费。

3.3 水印、checkpoint 与 Source 的配合

如果后续要做窗口聚合，水印的生成直接影响结果的准确性和延迟。水印延迟设得太大，窗口迟迟不触发，看板数据迟迟不出来；设得太小，迟到数据被判为垃圾丢弃。

我的经验值是把水印延迟设为上游数据最大乱序时间的两倍左右。比如实测业务系统的乱序通常在 10 秒内，水印延迟就设 20 秒，同时开启allowedLateness(30s)，给迟到数据一个兜底窗口。

WatermarkStrategy<RawEvent> wm = WatermarkStrategy .<RawEvent>forBoundedOutOfOrderness(Duration.ofSeconds(20)) .withTimestampAssigner((e, ts) -> e.getEventTime()) .withIdleness(Duration.ofMinutes(1));

withIdleness这个配置在分区多的场景下非常关键。如果某个分区长时间没数据（比如某个渠道夜里停投），该分区的水印会卡住不动，进而拖住整个算子的水印推进，导致所有窗口都不触发。加上空闲检测之后，一分钟没数据的分区会被标记为空闲，水印能继续推进。这个坑我见过至少三次，表现都是“作业看着正常，但窗口结果不出来”。

checkpoint 的间隔建议 30 秒到 60 秒之间。太短会增加 JobManager 和状态后端的压力，太长会导致故障恢复时要重放大量数据。同时把minPauseBetweenCheckpoints设为间隔的一半，避免 checkpoint 排队把算子反压住。

4. 数据处理与写入 StarRocks 的核心实现

这一章是整条链路的核心。写入环节的参数调优直接决定了吞吐、延迟和导入频率，也是 StarRocks 侧报错最集中的地方。

4.1 Stream Load 与 Connector 两条路线对比

写入 StarRocks 有两条主流路线，选哪条取决于团队的技术栈和维护能力。

我的选择是先上官方 Connector，99% 的场景够用。只有在需要做非常规的导入控制（比如按特定规则动态路由到不同表、需要在导入前后做额外校验）时，才考虑自研。自研的成本主要不在写代码，而在处理各种异常：网络超时、导入任务排队、label 重复、内存溢出，每一样都需要单独设计补偿逻辑。

Connector 的核心优势是攒批 + 两阶段提交。攒批把高频小批次合并成大批次提交，减少 StarRocks 的事务数和版本数；两阶段提交让作业在 checkpoint 完成时才真正提交事务，实现端到端的精确一次。

4.2 核心参数配置与取值依据

StarRocksSinkOptions options = StarRocksSinkOptions.builder() .withProperty("jdbc-url", "jdbc:mysql://fe-host:9030") .withProperty("load-url", "fe-host:8030") .withProperty("username", "flink_writer") .withProperty("password", "******") .withProperty("table-name", "dws_order_detail") .withProperty("database-name", "dws") .withProperty("sink.buffer-flush.max-bytes", "94371840") .withProperty("sink.buffer-flush.max-rows", "200000") .withProperty("sink.buffer-flush.interval-ms", "10000") .withProperty("sink.max-retries", "3") .withProperty("sink.properties.format", "json") .withProperty("sink.properties.strip_outer_array", "true") .withProperty("sink.properties.timeout", "300") .withProperty("sink.semantic", "exactly-once") .withProperty("sink.transaction.timeout", "600") .build();

逐个说取值理由。

sink.buffer-flush.max-bytes默认是 90MB 左右，我没有改。这个值决定了单个批次的内存占用上限，乘以并行度就是 sink 端的总内存。假设并行度是 8，最坏情况下 8 × 90MB ≈ 720MB，在 TaskManager 内存规划里要留出这个量，否则会出现java.lang.OutOfMemoryError: Java heap space，而且堆栈指向的是序列化代码，很容易误导排查方向。

sink.buffer-flush.max-rows从默认的 50 万调到 20 万，是我根据单行数据大小做的调整。如果单行平均 400 字节，20 万行约 80MB，刚好和 bytes 上限接近，两个条件同时满足时哪个先到就触发 flush，行为可预期。如果保持 50 万行，行数先到但只积累了 200MB 的数据，反而更容易打爆内存。

sink.buffer-flush.interval-ms从默认的 5 分钟调到 10 秒，这是延迟和导入频率的权衡。默认 5 分钟意味着看板数据最多延迟 5 分钟才看得到，对大多数实时场景都太长了。但调得太小，比如 1 秒，会导致频繁触发小批次导入，StarRocks 侧版本数飙升，反而拖慢查询。10 秒是一个比较平衡的取值，配合max-rows和max-bytes，在数据量大的时候由容量触发，数据量小的时候由时间触发。

注意：sink.properties.timeout是单次 Stream Load 的超时时间，单位秒，默认 600。设得太小，在导入大批次时会报超时，然后重试；重试又超时，最终失败。300 秒对百万行级别的批次足够，如果单批更大就往上调。这个参数和sink.transaction.timeout是两回事，后者是事务整体超时，要大于等于前者。

sink.semantic设成exactly-once的前提是开启 checkpoint 并且 StarRocks 版本支持事务接口。如果没有开 checkpoint，这个配置会失效，Connector 会退化成 at-least-once 并且打警告。生产环境我强烈建议开，因为主键模型的 Upsert 让 at-least-once 也不会产生重复行，但两阶段提交能避免“写入成功但作业失败”导致的数据不一致窗口。

4.3 攒批、并发与内存的联动计算

这一段是我认为整个链路最需要“算一遍”的地方。参数不是拍脑袋定的，是可以算出来的。

先确定目标延迟，比如 P99 要在 15 秒内，那么 flush 间隔必须小于 15 秒，取 10 秒，留出处理时间。再看吞吐，假设峰值 QPS 是 5000 行/秒，平均行大小 400 字节，那么 10 秒会积累 5 万行、约 20MB 数据。这个量级离 max-rows 和 max-bytes 都很远，说明时间触发是主要路径，行为稳定。

如果峰值是 5 万行/秒，10 秒就是 50 万行、200MB，超过 bytes 上限，那么 90MB 会先触发，每约 4.5 秒导入一次。这时要检查 4.5 秒的导入频率会不会给 StarRocks 造成压力：单表每天 2 万次导入，在 3 个副本下会产生约 6 万个版本，超过了 StarRocks 单分区默认的版本数限制（默认 1000 个，可通过partition_version_limit调整）。这种情况下，要么加大 max-bytes 让批次更大，要么改写入业务分区表的聚合层，减少导入频次。

并行度怎么定？基本原则是sink 并行度 ≤ 上游 topic 分区数，否则会有算子空闲。但如果写入是热点集中的（比如按订单 ID 哈希，某些大商户订单特别多），并行度拉满反而会有写冲突，导致 StarRocks 侧频繁报版本冲突。这时可以把 sink 并行度设为分区数的 1/2，在数据倾斜和并发之间找平衡。

再算一下 StarRocks 表的分桶数。如果日增 5000 万行，单行 400 字节，日增约 20GB。按单 bucket 压缩后 1GB 的目标，约 20 到 32 个桶。桶数定了之后不要随意改，因为改桶数需要重建表并重刷数据，成本很高。宁可一开始多给一点余量，但也不要翻倍给，桶太多会导致查询时每个桶的数据量太小，调度开销占比上升。

4.4 两阶段提交的落地细节

两阶段提交的流程是：每次 checkpoint 时，sink 先把当前批次的数据用 Stream Load 预提交（此时数据不可见），并把事务 ID 记录到状态里；当 checkpoint 完成通知到达时，再提交事务让数据可见。这个机制解决的是“checkpoint 回滚了但数据已经写进去”的不一致问题。

有两个落地细节容易被忽略。

第一是事务超时。如果 checkpoint 之间隔了 10 分钟，而sink.transaction.timeout只设了 5 分钟，事务会在提交前就超时失效，导致整个 checkpoint 失败。这个参数必须大于 checkpoint 间隔的两倍以上，留出足够的余量。

第二是 label 的幂等性。Connector 会用label-prefix加上子任务号和批次序号生成 label，同样的 label 重复提交时 StarRocks 会直接返回已存在的状态。这意味着在网络抖动导致重试的情况下，数据不会重复写入。但如果作业从 checkpoint 恢复后重新生成了一批 label，语义就变成了“新的一批数据”，这时依赖主键模型去重是最后一道保险。

// 事务超时要覆盖最坏情况：checkpoint 超时 + 两阶段提交耗时 env.enableCheckpointing(60_000); env.getCheckpointConfig().setCheckpointTimeout(10 * 60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink/checkpoints/order-dws");

setTolerableCheckpointFailureNumber(3)给了一些容错空间。网络抖动导致的单次 checkpoint 失败不该让作业直接重启，连续失败三次再重启是比较稳妥的策略。但也不能设太大，否则真出问题时作业会一直卡着不恢复。

5. 常见问题排查实录

这一章按“现象 → 定位 → 解决”的顺序整理，都是实际运行中反复出现的。

5.1 典型报错与处理方式

too many versions这个报错的值不值得细说？值得。它的本质是 StarRocks 每次导入都会在分区上生成一个新版本，查询时要合并多个版本。版本数太多，查询会变慢甚至直接报错。有些团队为了“实时性”把 flush 间隔调到 1 秒，结果写入看起来很顺畅，查询却越来越卡，最后就是被这个报错拦下来。实时性和导入频率不是线性关系，找到平衡点比一味追低延迟更重要。

5.2 延迟上升与反压排查路径

看板上数据延迟变高，排查顺序我固定为四步：先看 source 的消费延迟（Kafka lag），再看 Flink 的反压指标，再看 checkpoint 耗时，最后看 StarRocks 的导入耗时。

Kafka lag 持续上涨通常说明 Flink 处理能力不足或下游写入被阻塞。这时打开 Flink 的反压面板，如果 sink 算子显示 HIGH，说明写入 StarRocks 慢；如果 sink 正常但中间算子 HIGH，说明业务逻辑有阻塞（比如同步维表查询）。

维表同步查询是延迟的隐形杀手。用RichMapFunction里直接查 JDBC，每条数据都要等一次网络往返，QPS 上千时延迟会飙升。正确做法是用LookupJoin配合缓存，设置lookup.cache.max-rows和lookup.cache.ttl，把热点维表数据缓存在算子内。缓存 TTL 设 5 到 10 分钟，兼顾一致性和性能。

checkpoint 耗时过长也会拖慢整体。如果 checkpoint 持续时间接近间隔，会导致频繁的同步开销。查一下状态大小，如果状态异常大（比如做全局去重），考虑改成 TTL 状态或者把去重逻辑下推到 StarRocks 主键模型。

StarRocks 导入耗时可以在 FE 的审计日志里查，重点看LoadTime和StreamLoadPutTime。前者长说明写入磁盘慢，检查磁盘 IO 和 compaction 情况；后者长说明 FE 处理元数据慢，检查是否元数据操作过于频繁。

5.3 数据重复与丢失的定位方法

数据对不上是实时链路最伤的问题，因为很难复现。我的定位思路是分层比对。

先比对 Kafka 单分区某个 offset 区间的消息条数，和 StarRocks 里对应时间范围的明细条数。如果不一致，说明从 Kafka 到 StarRocks 之间有丢失或重复。再缩小到 Flink 的算子链，在关键节点用侧输出流打点统计，看是哪个环节开始产生差异。

一个高频原因是反序列化时静默丢数据。比如自定义 deserializer 里 catch 了异常但没输出，脏数据就悄无声息地消失了。所以前面强调脏数据要进侧输出流，就是为了让“丢失”变成“可见”。

另一个高频原因是水位线和窗口的配合。如果用了事件时间窗口，而数据的事件时间戳来自业务字段，某些数据的时间戳是 0 或者未来时间，就会被判为迟到或超范围丢弃。排查时统计一下事件时间的分布，看有没有异常值。

至于重复，主键模型基本能兜住，但如果是明细模型（Duplicate）且作业用了 at-least-once，重复几乎必然出现。这种场景要么补一个基于唯一键的去重，要么老老实实开两阶段提交。我在实际项目里见过因为图省事关掉两阶段提交，结果重启几次之后报表数字翻倍的案例，修正数据花了整整一天。

6. 上线前的压测与运维心得

链路跑通和跑稳是两件事。上线前做一轮有目的的压测，上线后盯住几个关键指标，能把大部分事故挡在发生之前。

6.1 压测怎么做才有意义

压测不是简单地把数据量放大，而是要覆盖三种场景：稳态峰值、突发流量、故障恢复。

稳态峰值用生产流量的 1.5 倍持续跑 30 分钟，观察延迟是否稳定、checkpoint 是否按时完成、StarRocks 的版本数是否失控。突发流量用短时间高并发灌入，比如 5 分钟内灌入平时一小时的数据量，观察攒批机制是否按预期触发、有没有 OOM。故障恢复是手动杀掉 TaskManager，观察作业从 checkpoint 恢复的时间和恢复后的数据一致性。

第三项最容易被跳过，但恰恰最重要。我见过很多作业平稳跑了一个月，第一次真故障恢复时才发现 checkpoint 根本没成功过，因为存储路径权限有问题，日志里只有一行不起眼的 warning。

6.2 值得长期盯住的指标

导入频率那条我给的参考值是每小时 360 次，也就是平均 10 秒一次，和前面 flush 间隔的取值对应。如果实际观测到每小时上千次，说明批次太小，要么数据量本身小（可以适当放宽间隔），要么攒批参数没生效（检查 max-rows 是否设得太小）。

6.3 几条实打实的避坑经验

第一条，别在 Flink 里做跨天的大窗口聚合。曾有团队为了算日活，在 Flink 里开 24 小时滚动窗口，状态大到 checkpoint 每次要几分钟，作业稍微抖动就恢复不了。正确做法是明细落 StarRocks，用物化视图或者定时 SQL 做聚合。

第二条，维表关联尽量用 LookupJoin 而不是广播状态。广播状态适合小维表（几千行以内），超过这个量级每次更新都要广播全量，网络和内存压力都大。LookupJoin 走缓存，维表更新只影响缓存失效的那部分。

第三条，StarRocks 的 compaction 要关注。主键模型频繁 Upsert 会产生大量小文件，后台 compaction 跟不上时查询会明显变慢。监控SHOW PROC '/compactions'里的进度，如果 pending 数量持续增长，说明写入压力超过 compaction 能力，需要降低导入频率或者调整 compaction 策略。

第四条，作业的重启策略别用默认值。默认的固定延迟重启在遇到 StarRocks 短暂不可用时会疯狂重试，反而加重下游负担。用指数退避，初始延迟 10 秒，最大 5 分钟，给下游留出恢复时间。

env.setRestartStrategy(RestartStrategies.exponentialDelayRestart( Time.seconds(10), Time.minutes(5), 2.0, Time.hours(1), 10 ));

第五条，也是我个人体会最深的一条：这条链路的可观测性投入，回报率远高于参数调优。把 Kafka lag、checkpoint 时长、导入成功率、脏数据比例这几个指标接到统一面板上，配好告警阈值，比反复调参数省心得多。参数调优解决的是“能不能跑到 10 秒延迟”，可观测性解决的是“出问题时能不能 5 分钟定位”。后者的价值在长期运行中会不断放大。

最后分享一个我在实际搭建时形成的小习惯：每次调整参数之前，先在配置文件里写一行注释说明这次调整的原因和预期效果，比如“max-rows 从 50 万降到 20 万，因为单行变大导致内存超限”。半年后回头看，这一行注释能省掉大量“当初为什么这么设”的困惑。