1. 这条链路到底在解决什么问题第一次接到Kafka 里的埋点数据要进 StarRocks 做实时看板这个需求的时候我脑子里第一反应是这不就是标准的流式 ETL 吗Source 接 Kafka中间做点清洗和关联Sink 打到 StarRocks一天能搞定。真动手之后才发现代码写完只花了半天剩下的时间全耗在为什么数据有重复为什么看板延迟十分钟为什么导入报 label 已存在为什么半夜任务挂了没人知道这些问题上。所以这篇东西我不打算写成一份 API 手册而是把这条 Flink 读取 Kafka 数据、处理之后写进 StarRocks 的链路从选型、建表、写代码、调参到排障完整地讲一遍我在实际项目里踩出来的路径。先说清楚它是什么。这条链路本质上是把消息队列里的原始事件流经过 Flink 的清洗、过滤、补维、聚合落到 StarRocks 这种能支撑高并发点查和实时分析的 OLAP 库里让下游的报表、看板、指标接口能在一个可接受的延迟内读到数据。它解决的问题很具体业务方不满足于 T1 的离线结果希望分钟级甚至秒级看到分渠道、分地区、分版本的实时指标同时原始数据又不能直接查格式乱、字段缺、口径不统一必须有一层流处理做加工。适合读这篇的人大致是三类正在做实时数仓链路的同学手上刚好要用 Flink 和 StarRocks 但不确定参数怎么配已经跑起来但被数据重复、延迟、反压搞得头疼想优化的人还有准备做技术选型、想知道这条链路在哪些地方容易翻车的人。这里有个认知要先摆正这条链路不是三个组件连起来这么简单它是一条数据一致性的契约链。Kafka 保证消息不丢Flink 保证处理状态可恢复StarRocks 保证写入可幂等任何一环的语义没对齐最终结果就是脏数据。所以后面的内容里Checkpoint 配置、Label 机制、攒批参数这些看起来琐碎的东西才是真正决定这条链路能不能上生产的关键代码本身反而是最短的那块板。2. 整体架构与选型背后的取舍2.1 为什么是 Flink 而不是别的流处理引擎流处理引擎能选的其实不止一个轻量点的有基于微批的思路重一点的有 Spark Structured Streaming。之所以最后落在 Flink 上核心原因有三个。第一是真正的事件驱动不是靠微批攒出来的伪流窗口触发和定时器精度能到毫秒级对于需要按事件时间做迟到数据处理、乱序纠正的场景这个能力是刚需。第二是状态管理成熟RocksDB 状态后端加上 Checkpoint 和 Savepoint可以让一个跑了几个月的任务在扩缩容之后从断点恢复中间做维表缓存、去重、会话切分这些有状态的操作不用自己造轮子。第三是连接器生态完整Kafka Source 和 StarRocks Sink 都有官方或社区维护的成熟实现不用自己写一堆 HTTP 请求去拼。当然也不是没有代价。Flink 的运维复杂度明显高于微批方案checkpoint 大了之后做一次全量恢复可能要几十分钟TaskManager 内存管理也要调。所以如果你的场景是每分钟跑一次小批量同步其实没必要上 Flink用调度工具跑批更省心。但只要是持续不断的流、要做窗口聚合和有状态计算Flink 的投入产出比就很划算了。2.2 三段式链路的职责边界怎么划这条链路我习惯分成三段来看每段有明确的边界越界就会出问题。第一段是 Kafka 侧它的职责只有两个把数据按合理的分区键投递保证同一业务主键的消息落在同一分区从而有序以及配置足够长的保留时间让 Flink 在故障恢复时还有数据可以重放。至于数据长什么样、字段合不合法Kafka 不负责。第二段是 Flink 侧职责是清洗、标准化、补维、聚合和去重把原始事件加工成可以直接查询的宽表或者指标表。这里最常见的设计错误是把大量查询型逻辑放到 Flink 里做比如每次来一条数据就查一次外部接口补维结果外部接口成了瓶颈反压一路顶回 Kafka。第三段是 StarRocks 侧职责是存好、建好索引、支撑查询写入性能依赖表模型和分区分桶设计查询性能依赖物化视图和排序列这些都是在建表时定死的事后改成本很高。我见过一个典型的越界案例某业务的埋点数据在 Flink 里做了七层嵌套的 JSON 解析然后把几十个字段全打成一张明细表写进 StarRocks结果单表一天几百亿行查询全靠现场聚合看板加载要十几秒。后来改成 Flink 里直接按分钟粒度预聚合写入量降到原来的百分之一查询变成点查。这个改动说明一件事聚合放在流里做明细放在仓里做位置放反了性能就是灾难。2.3 数据一致性语义怎么对齐这是很多人容易忽略但必须提前想清楚的部分。Kafka 到 Flink 这一段靠isolation.levelread_committed加上关闭自动提交 offset、把 offset 存在 checkpoint 里能做到精确一次消费。Flink 到 StarRocks 这一段靠 Stream Load 的 Label 机制实现幂等每个批次带一个全局唯一的 labelStarRocks 侧相同 label 的导入只会成功一次重复提交会返回已存在。两段合起来配合 Flink 的 Checkpoint整条链路才能做到端到端的精确一次。这里面任何一个环节掉链子比如 checkpoint 没开、label 没配语义就退化成至少一次数据重复是必然的。所以我在项目里有个硬规矩只要下游是给人看的指标checkpoint 和 label 必须同时开宁可牺牲一点延迟也不接受对不上账。3. 动手前必须搞定的环境与版本对齐3.1 版本矩阵怎么选才不踩坑这条链路上最容易出问题的地方不是代码是版本。Flink 的大版本之间 API 有不兼容改动StarRocks 的 Flink 连接器又是跟着 Flink 版本走的选错了要么编译不过要么运行时抛NoSuchMethodError排查起来非常费劲。我的建议是先把三个组件的版本钉死再去找匹配的连接器而不是反过来。组件建议版本区间选择理由Flink1.17 或 1.18这两版 API 相对稳定社区连接器覆盖完整1.19 之后部分接口有调整StarRocks3.0 及以上主键模型持久化索引默认开启写入性能比 2.x 明显好Kafka3.x客户端协议兼容性好配合 Flink 的 Kafka 连接器没坑flink-connector-starrocks与 Flink 大版本匹配的最新小版本连接器版本必须严格对应 Flink 版本不能跨大版本使用flink-connector-kafka与 Flink 大版本匹配Flink 1.15 之后 Kafka 连接器从主仓库剥离需要单独引入表里最后两行要特别强调一下这是新手最容易翻车的地方。Flink 1.15 之后把 Kafka 连接器拆出去单独发版了如果你用的是 1.17却引了一个旧版本的flink-connector-kafka编译期可能不报错运行起来就是类找不到或者方法签名不匹配。我的做法是在pom.xml里用flink.version和connector.version两个 property 统一管理改版本只改一处避免同一个项目里依赖打架。3.2 依赖引入的两种方式和各自的坑第一种是 Maven 引入本地开发打包提交集群。这种方式下要注意依赖的作用域Flink 自身提供的包flink-streaming-java、flink-clients这些必须标成provided否则打进 fat jar 之后会跟集群上的包冲突典型症状是提交任务时报序列化异常或者ClassCastException。而 Kafka 连接器、StarRocks 连接器这些集群里没有的包必须打进包里不能标provided否则运行时报ClassNotFoundException。这个边界划错了本地能跑集群跑不了或者反之非常折磨人。第二种是 Flink SQL 的CREATE TABLE ... WITH (connector starrocks)方式把 jar 放到集群的lib目录或者用-C参数指定。这种方式的好处是不用写 Java 代码纯 SQL 就能跑适合逻辑简单的同步任务坏处是复杂的数据清洗和自定义函数不好写而且参数拼错的时候报错信息比较隐晦常常只告诉你连接失败不告诉你是哪个参数写错了。我的经验是简单同步用 SQL有复杂加工逻辑用 DataStream API不要为了省事硬用 SQL最后写成一坨两千行的嵌套查询维护成本比 Java 还高。注意连接器的 jar 包版本和 Flink 版本不匹配时最常见的报错不是编译错误而是运行时抛java.lang.NoSuchMethodError或IncompatibleClassChangeError。看到这两类异常先查版本别急着看业务代码。4. StarRocks 侧的表设计与写入前提4.1 数据模型怎么选StarRocks 有三种数据模型选错了后面全是麻烦。明细模型适合保留所有原始行、需要查明细的场景相同主键不去重聚合模型适合预先定义好聚合方式的指标表写入时就聚合查询极快但不能查明细主键模型适合需要按主键更新的场景比如维表、状态表、需要 upsert 的订单表写入新行会覆盖旧行。流式写入场景下我大部分时候选主键模型或者明细模型。选主键模型的判断标准是数据是否存在同主键多次写入、且只需要保留最新状态。比如用户画像宽表、订单状态表典型的主键模型。而埋点流水、日志这种每一条都要留痕的用明细模型配合分区做冷热分离。聚合模型的坑在于一旦后续口径变了要新增维度就得重新建表重刷数据灵活度差所以我一般只在指标口径极其稳定的场景才用。主键模型建表时还有两个参数要留意enable_persistent_index建议开启把主键索引落到磁盘避免内存撑爆replication_num生产环境至少 3测试环境可以调成 1但千万别把测试配置带到生产。这里我踩过一次测试环境建表时图省事写了replication_num 1上线时脚本直接复用结果某台 BE 节点重启数据不可用看板直接白屏。4.2 分区分桶参数怎么算这是建表环节唯一需要动脑子算的地方也是最容易被忽略的地方。分区决定数据怎么按时间切分分桶决定数据怎么在 BE 之间打散。分区太细会导致小文件多、元数据压力大分区太粗会导致单分区数据量过大查询扫描不友好。我的经验值是单分区数据量控制在 10GB 到 100GB 之间按天分区是最常见的粒度如果日增量超过 100GB 就考虑按小时分区。分桶数的计算有个简单公式单分区数据量除以单 tablet 的目标大小建议 1GB 到 3GB向上取整到 2 的幂并且保证是 BE 节点数的整数倍。举个实际例子假设日增 20 亿行、单行平均 200 字节那一天大概 400GB按天分区单 tablet 目标 2GB那需要 200 个 tablet 左右取 256 个桶。如果集群有 8 台 BE256 除以 8 等于 32正好整除数据分布均匀。如果算出个 200取 256 比取 128 更合适因为 128 意味着单 tablet 3GB 出头略超目标值但 256 会让单 tablet 降到 1.5GB 左右查询并行度更好。分桶键的选择原则是高基数、查询中经常作为过滤条件或者 join 键的字段。分桶键选错会导致数据倾斜比如选了个 90% 都是同一个值的字段那所有数据都挤到一两个 tablet 上写入直接成瓶颈。我一般用业务主键或者用户 ID 做分桶键不用时间字段时间已经做分区了。CREATE TABLE dws_realtime_event ( dt DATE NOT NULL COMMENT 分区日期, user_id BIGINT NOT NULL COMMENT 用户ID, channel VARCHAR(64) NOT NULL COMMENT 渠道, event_cnt BIGINT NOT NULL COMMENT 事件数, amount DECIMAL(20,2) NOT NULL COMMENT 金额, update_time DATETIME NOT NULL COMMENT 更新时间 ) PRIMARY KEY (dt, user_id, channel) PARTITION BY RANGE (dt) ( START (2024-01-01) END (2025-01-01) EVERY (INTERVAL 1 DAY) ) DISTRIBUTED BY HASH (user_id) BUCKETS 256 PROPERTIES ( replication_num 3, enable_persistent_index true, storage_medium SSD );这段建表语句里有几个细节值得说。主键必须包含分区列所以dt放在了主键第一位分区用START ... END ... EVERY批量创建避免手写一堆PARTITION子句storage_medium明确指定 SSD因为实时写入频繁机械盘会成为拖累。如果 BE 集群是混合存储还可以加storage_cooldown_time让冷数据自动下沉到便宜介质。4.3 Stream Load 的 Label 机制与事务边界Flink 写 StarRocks 走的是 Stream Load 协议本质是一个 HTTP 请求把一批数据推过去。这个协议有几个特性必须理解。第一它是同步的请求发出去要等返回返回成功才算写入完成这跟某些异步落盘的方式不一样好处是失败能立刻感知。第二它是按 label 幂等的同一个 label 重复提交第二次会返回Label Already Exists数据不会重复导入。第三单次导入会形成一个事务要么整批成功要么整批失败不会出现半批数据的情况。理解这三点之后很多设计就顺理成章了。比如为什么必须要配sink.label-prefix因为连接器需要根据前缀加上 checkpoint ID 和子任务编号来生成唯一 labelFlink 从 checkpoint 恢复时用同样的 label 重放StarRocks 侧自动去重。再比如为什么攒批不能攒太大因为单次 Stream Load 的默认超时是 600 秒攒得太多一旦超时整批白写还要重试。提示sink.label-prefix在整个集群内必须全局唯一。多个任务共用同一个前缀会导致 label 冲突表现为大量导入失败并报 label 已存在。建议用业务名_表名_任务版本这种格式改逻辑时顺手改版本号。5. Flink 任务骨架的搭建与核心配置5.1 Source 端Kafka 消费怎么配对Source 端的配置看着简单实际上有好几个参数决定了下游的稳定性和数据准确性。首先是启动位点scan.startup.mode有几个取值earliest-offset从最早开始、latest-offset从最新开始、group-offsets从消费组提交的位点开始、timestamp从指定时间开始。生产环境我一般用group-offsets配合properties.auto.offset.resetearliest兜底这样第一次启动能从头消费后续重启从断点接上。绝对不要用latest-offset做生产任务一旦任务停了几天再启动中间的数据就永久丢了。然后是properties.isolation.level如果要保证精确一次必须设成read_committed只读已提交的消息避免读到上游事务中途的数据。同时要把enable.auto.commit设成 falseoffset 交给 Flink 的 checkpoint 管理而不是让消费者自己提交否则会出现checkpoint 失败但 offset 已经提交的尴尬局面恢复之后数据就丢了。Properties props new Properties(); props.setProperty(bootstrap.servers, kafka-broker-1:9092,kafka-broker-2:9092); props.setProperty(group.id, realtime-dws-event); props.setProperty(enable.auto.commit, false); props.setProperty(isolation.level, read_committed); props.setProperty(auto.offset.reset, earliest); KafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(props.getProperty(bootstrap.servers)) .setTopics(ods_user_event) .setGroupId(props.getProperty(group.id)) .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST)) .setProperty(isolation.level, read_committed) .setValueOnlyDeserializer(new SimpleStringSchema()) .build();还一个容易被忽略的点是分区数与并行度的关系。Kafka 单个分区同一时刻只能被一个消费者消费所以 Source 的并行度上限就是分区数。如果你设了 16 个并行度但 topic 只有 4 个分区那有 12 个 Source 子任务是空转的白占资源。反过来说如果分区数是 64 而并行度只有 8那么每个子任务要消费 8 个分区吞吐上不去。合理的配置是并行度等于分区数或者分区数能被并行度整除。5.2 Checkpoint整条链路的生命线Checkpoint 的配置直接决定了任务的容错能力和数据一致性这部分我建议单独拉出来认真调。核心参数有四个间隔、超时、最小间隔、模式。间隔设太短每次 checkpoint 都要对齐屏障、上传状态开销大设太长故障恢复时要重放的数据多而且 StarRocks 侧的攒批数据在 checkpoint 之前是不可靠的。我一般设 30 秒到 1 分钟配合超时 10 分钟最小间隔给间隔的一半避免前一次还没完成就触发下一次。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(60_000, CheckpointingMode.EXACTLY_ONCE); CheckpointConfig cpConfig env.getCheckpointConfig(); cpConfig.setCheckpointTimeout(600_000); cpConfig.setMinPauseBetweenCheckpoints(30_000); cpConfig.setMaxConcurrentCheckpoints(1); cpConfig.setExternalizedCheckpointCleanup( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); cpConfig.setTolerableCheckpointFailureNumber(3); env.setStateBackend(new EmbeddedRocksDBStateBackend(true));tolerableCheckpointFailureNumber这个参数值得单独说默认是 0意思是一次 checkpoint 失败任务就挂。生产环境我一般设成 3给网络抖动、StarRocks 短暂不可用留点缓冲。但也不能设太大否则连续失败十几次任务还在苟活状态早就落后了问题被掩盖。RocksDB 状态后端建议开启增量 checkpoint构造参数传 true状态大的时候能把每次 checkpoint 的数据量降一个数量级。配合本地恢复常规故障恢复能在一两分钟内完成而不是几十分钟。这里有个前提是集群要配好 checkpoint 存储目录本地恢复依赖每个 TaskManager 上有历史状态副本节点挂了就没法本地恢复走远程恢复。5.3 Sink 端三种写法怎么选往 StarRocks 写数据实践中有三种写法适用场景差别很大。第一种是官方连接器的 DataStream API用StarRocksSink.sink(...)构造一个 SinkFunction配置通过StarRocksSinkOptions传入。这是最主流的做法控制力最强能精确控制攒批参数、序列化格式、重试策略也能拿到写入失败的异常做自定义处理。缺点是要自己管序列化字段顺序必须和表结构严格对应改了表结构忘了改代码就是线上事故。第二种是Flink SQL 的 connectorCREATE TABLE里指定connector starrocks参数通过WITH传。好处是纯声明式配合 SQL 做数据加工很顺手表结构变更时改 SQL 就行。缺点是参数校验比较松写错了不报错直到运行时而且不支持一些高级特性比如自定义序列化。第三种是自己封装 Stream Load 的 HTTP 请求用RichSinkFunction加上批量缓冲自己做。这种写法除非有非常特殊的诉求比如需要自己控制事务边界、或者连接器版本有 bug 绕不过去否则不建议因为 label 管理、重试、超时这些细节全要自己处理坑很多。StarRocksSinkOptions options StarRocksSinkOptions.builder() .withProperty(jdbc-url, jdbc:mysql://fe-host:9030) .withProperty(load-url, fe-host:8030) .withProperty(database-name, realtime_dw) .withProperty(table-name, dws_realtime_event) .withProperty(username, etl_user) .withProperty(password, ******) .withProperty(sink.properties.format, json) .withProperty(sink.properties.strip_outer_array, true) .withProperty(sink.buffer-flush.max-rows, 50000) .withProperty(sink.buffer-flush.max-bytes, 33554432) .withProperty(sink.buffer-flush.interval-ms, 5000) .withProperty(sink.max-retries, 3) .withProperty(sink.semantic, exactly-once) .withProperty(sink.label-prefix, dws_event_v1) .build(); DataStreamString jsonStream ...; jsonStream.sinkTo(StarRocksSink.sink(options));load-url这里填的是 FE 的 HTTP 端口请求会被 FE 重定向到具体的 BE所以填 FE 地址就行不要直接填 BE因为 BE 可能扩缩容。如果 FE 前面有负载均衡可以把多个 FE 地址用分号隔开连接器会轮询选择提高可用性。6. 数据处理层的写法与那些绕不开的坑6.1 反序列化与脏数据处理从 Kafka 读出来的原始数据第一道关卡就是反序列化。看起来简单实际上一半的线上事故都出在这里。常见情况是上游字段类型变了、字段缺了、某条消息体是半截的 JSONJSON.parseObject直接抛异常整个任务挂掉。我一开始的做法是 try-catch 之后返回 null让这条数据被过滤掉结果跑了一周发现数据量比上游统计少了 3%排查半天才定位到是某类消息带了非法字符全被静默丢弃了。正确的做法是把脏数据显式收集起来而不是静默丢弃。可以用侧输出流把解析失败的消息连同失败原因一起输出到另一个 Kafka topic 或者对象存储定期检查。这样既能保证主链路不被单条坏数据拖垮又能保证丢失的数据可追溯。public class EventDeserializer implements DeserializationSchemaEvent { Override public Event deserialize(byte[] message) { try { String raw new String(message, StandardCharsets.UTF_8); return JSON.parseObject(raw, Event.class); } catch (Exception e) { // 记录原始报文交给侧输出流处理 FailedRecordHolder.set(new String(message, StandardCharsets.UTF_8), e.getMessage()); return null; } } Override public boolean isEndOfStream(Event nextElement) { return false; } Override public TypeInformationEvent getProducedType() { return TypeInformation.of(Event.class); } }还一个坑是时间字段的格式。上游给的可能是毫秒时间戳也可能是2024-01-01 10:00:00这种字符串还可能是 ISO8601 带时区的。StarRocks 的DATETIME类型只认yyyy-MM-dd HH:mm:ss这个格式如果直接把时间戳塞进去写入会失败或者变成 1970 年。所以中间必须有一个明确的格式转换步骤把各类时间格式统一转成目标字符串。这里要特别注意时区如果上游是 UTC 时间戳转换时要显式加上时区偏移否则整条链路的统计口径都会偏 8 小时。我们线上就出过一次指标在凌晨 0 点到 8 点之间总是对不上查了一下午才发现是时区没处理。6.2 维表关联什么时候该用 Lookup Join实时链路做补维也就是把 ID 换成名称有两种做法查外部存储Lookup Join和广播维表Broadcast Join。选错的话性能差距可能是十倍。Lookup Join 是每条流数据来了就去外部存储查一次适合维表大、更新频繁、查询量可控的场景典型实现是 Flink SQL 的LOOKUP JOIN加上 HBase 或 StarRocks 作为维表。广播维表是把整张维表加载到每个 TaskManager 的内存里适合维表小、更新不频繁的场景比如渠道字典、地区映射这种几千行的表。我踩过的一个坑是在 Lookup Join 里没加缓存。每条数据都去查一次维表QPS 一上来维表库直接被打满反压一路传到 Kafka消费延迟飙到几分钟。后来给 Lookup Join 加了 LRU 缓存缓存 1 万条、过期时间 10 分钟命中率到 95% 以上数据库压力瞬间降下来。这个参数在 Flink SQL 里可以用LOOKUPhint 指定也可以在连接器配置里设置。但缓存会导致维表更新有延迟业务方要能接受这个延迟一般十分钟内的延迟是能接受的。SELECT e.user_id, e.channel, d.channel_name, e.amount FROM ods_event AS e LEFT JOIN dim_channel FOR SYSTEM_TIME AS OF e.proc_time AS d ON e.channel d.channel_code还有个细节是维表关联不上时的处理。如果用 INNER JOIN关联不上的数据直接丢了业务方会发现总量比上游少用 LEFT JOIN维表字段是 null下游展示成未知渠道。我的习惯是 LEFT JOIN 加上默认值填充同时把关联不上的比例做成监控指标超过阈值就告警因为关联率突然下降往往意味着维表更新出了问题。6.3 水位线与迟到数据只要涉及事件时间窗口水位线就是绕不过去的话题。水位线的本质是对数据还会不会更早的一个估计设得激进延迟小会丢迟到数据设得保守延迟大窗口触发慢。我的经验是先看上游数据的乱序程度如果 Kafka 里同一分区的数据基本有序水位线延迟可以设 3 到 5 秒如果乱序严重比如多端上报、网络抖动就设到 30 秒甚至 1 分钟。CREATE TABLE ods_event ( user_id BIGINT, channel VARCHAR(64), amount DECIMAL(20,2), event_ts TIMESTAMP(3), WATERMARK FOR event_ts AS event_ts - INTERVAL 10 SECOND ) WITH ( connector kafka, topic ods_user_event, properties.bootstrap.servers kafka-broker-1:9092, properties.group.id realtime-dws-event, scan.startup.mode group-offsets, format json );光设水位线还不够迟到数据得有地方去。Flink 的窗口算子支持allowedLateness允许数据迟到一段时间内还能更新窗口结果超过这个时间的可以用侧输出流接住。我在实际项目里会把超迟数据单独写到一张明细表定期回刷保证最终结果和上游对得上。这件事不做的话每天早上对账都是一场灾难。提示水位线延迟设大了窗口触发慢看板数据会延迟设小了迟到数据被丢指标偏低。这个权衡没有标准答案只能根据业务对延迟和准确性的取舍来定建议前期先设保守值观察两周的迟到分布之后再收紧。7. 写入性能调优与参数计算7.1 攒批参数怎么算才不浪费StarRocks 的写入性能高度依赖攒批一批数据越大摊到每行的开销越小但延迟也越高而且单批过大会撞上超时或者内存上限。三个核心参数是sink.buffer-flush.max-rows、sink.buffer-flush.max-bytes、sink.buffer-flush.interval-ms任意一个条件满足就触发一次 flush所以实际批次大小取决于最先触发的那个条件。举个实际的算法。假设业务要求端到端延迟不超过 10 秒上游 QPS 稳定在 2 万单条记录序列化后大约 400 字节。如果按时间触发配 5 秒一次的话每批大概 10 万行、40MB这个量级对 StarRocks 来说很舒服。但如果 QPS 波动大高峰期可能 10 秒就攒到 200 万行、800MB那就太大了。所以要把max-bytes设成一个上限比如 64MB那么高峰期实际是按字节触发每批 16 万行左右延迟反而更低。参数就按这个逻辑反推sink.buffer-flush.interval-ms 5000不超过业务延迟容忍度的一半sink.buffer-flush.max-bytes 6710886464MB防止突发流量攒太大sink.buffer-flush.max-rows 200000作为行数兜底防止单行特别小的时候字节数触发不了这三者的关系是区间控制不是精确控制。实际批次会在某个区间浮动这是正常的不用追求精确。还有一点是并行度与分桶的对应关系。每个 Sink 子任务独立攒批、独立发起 Stream Load如果并行度远小于分桶数那么每个子任务要写很多个 tablet单次导入涉及的分片多协调开销大如果并行度远大于分桶数多个子任务写同一批 tablet容易产生导入冲突和锁竞争。我的经验是Sink 并行度等于或者略小于分桶数除以 BE 数量具体来说256 个桶、8 台 BESink 并行度设在 8 到 32 之间比较合适。7.2 常见性能瓶颈怎么定位写入慢的表现有很多种定位思路是从后往前推。第一步看 StarRocks 侧从 FE 的审计日志里看每个 Stream Load 的耗时、数据量、错误码。如果单次导入耗时超过 5 秒而且数据量不大那可能是 BE 负载高或者 tablet 有版本堆积需要 compaction 了。第二步看 Flink 侧从 Web UI 看 Sink 算子的 busy 和 backPressured 指标如果 Sink 一直处于 100% busy说明写入是瓶颈如果 Sink 是空闲但上游反压说明上游处理慢了。第三步看网络和连接如果导入频繁超时可能是 FE 到 BE 的带宽不够或者连接数打满了。实际项目中我遇到最多的是两类。一类是小批频繁写入每隔一两秒就发一次 Stream Load数据量才几百行导入开销远大于数据处理开销BE 的导入线程被占满。解决办法就是调大攒批参数宁可牺牲几秒延迟。另一类是主键模型写入热点所有数据都往少数几个 tablet 上写其他 tablet 闲着整体吞吐上不去。这种要么换分桶键要么在写入前给分桶键加个随机后缀打散。现象可能原因排查手段处理方式导入耗时忽高忽低BE 在做 compaction看 BE 的 compaction 指标错峰写入调大攒批持续报超时单批数据量过大看日志里的 batch size调小 max-bytes写入吞吐上不去分桶键数据倾斜查各 tablet 的数据量分布更换分桶键CPU 打满但吞吐低序列化开销大火焰图看热点换二进制格式或精简字段内存持续增长状态没配 TTL看 checkpoint 大小变化给状态加 TTL8. 线上问题排查实录与避坑清单8.1 问题速查表把过去一年踩过的坑整理成一张表遇到问题时按图索骥能省下不少时间。报错或现象根因解决方式Label Already Existslabel 前缀重复或未配每个任务用唯一前缀Tablet ... version冲突并发导入同一批 tablet降低 Sink 并行度或增大分桶数数据写入成功但查不到攒批未 flush 或事务未提交检查 flush 间隔等待 checkpoint 完成字段全部为 nullJSON 字段名与表列名不匹配配jsonpaths显式映射时间字段变成 1970时间戳未格式化统一转成yyyy-MM-dd HH:mm:ss任务启动即失败连接器版本与 Flink 不匹配核对版本矩阵消费延迟持续上涨反压写入或处理慢从 Sink 往上游逐级排查恢复后数据重复未配 exactly-once 或 label开启 checkpoint 加 label8.2 几个印象最深的具体案例第一个是字段顺序错位。有次改了 StarRocks 表结构加了个字段在中间Flink 侧用的是 CSV 格式按位置写没同步改代码结果数据全都串位了用户 ID 写进了金额字段。这个问题最难发现因为导入全部成功没有任何报错。后来我们定了规矩一律用 JSON 格式加显式字段名不用 CSV 按位置写。虽然 JSON 的序列化开销大一点但换来的是结构变更时的安全这个代价值得。第二个是checkpoint 一直失败导致任务反复重启。日志里报的是Checkpoint expired before completing但看 checkpoint 大小才几十 MB按理说很快。查了半天发现是某个 TaskManager 的磁盘 IO 打满了写本地 state 卡住。根因是那台机器上还跑着别的服务磁盘争抢。这个问题提醒我Flink 集群最好独立部署不要跟其他服务混部尤其是对磁盘 IO 敏感的状态后端。第三个是数据延迟规律性地出现尖峰。每天下午两点到四点看板延迟从几秒涨到十几分钟其他时间正常。最后定位到是那个时段有个大的离线任务在跑把 StarRocks 的 BE CPU 占满了导致 Stream Load 排队。解决办法是把离线任务的资源组和实时写入隔离开用 StarRocks 的资源组功能做 CPU 和内存的硬隔离之后延迟就稳定了。8.3 上线前建议做的几项检查任务写完能跑通只是第一步上线前我一般会过一遍这几件事。压测用生产流量的 1.5 倍打一遍看反压和延迟确认资源够用。故障演练手动 kill 一个 TaskManager看恢复时间是否在可接受范围内数据是否有重复或丢失。监控埋点至少要监控消费延迟、checkpoint 时长和大小、写入成功率、端到端延迟这四个指标任何一个异常都要能告警。数据对账跑一个批任务统计上游总量跟 StarRocks 里的总量对比差异率控制在万分之一以内才敢上线。-- 对账查询示例按小时统计上游与目标表行数 SELECT e.hour_str, e.src_cnt, s.dst_cnt, ROUND((e.src_cnt - s.dst_cnt) * 100.0 / e.src_cnt, 4) AS diff_pct FROM ( SELECT DATE_FORMAT(event_time, %Y-%m-%d %H) AS hour_str, COUNT(1) AS src_cnt FROM ods_event_source GROUP BY 1 ) e LEFT JOIN ( SELECT DATE_FORMAT(update_time, %Y-%m-%d %H) AS hour_str, COUNT(1) AS dst_cnt FROM dws_realtime_event GROUP BY 1 ) s ON e.hour_str s.hour_str WHERE e.hour_str DATE_FORMAT(DATE_SUB(NOW(), INTERVAL 1 DAY), %Y-%m-%d %H) ORDER BY e.hour_str DESC;这段对账 SQL 我一般做成定时任务每天跑一次差异率超过阈值就发告警。一开始觉得这是额外负担后来发现它是发现隐性数据问题最有效的手段有好几次都是靠它发现了上游字段变更导致的部分数据丢失。注意对账口径一定要和业务方确认清楚是算去重后的用户数还是算原始事件数是算写入时间还是算事件时间。口径没对齐对账永远是数据有问题最后大家都不看了。最后分享一个我在实践中最深的体会这条链路的复杂度不在技术组件本身而在于每一个环节的语义假设是否被显式地写下来并被验证。Kafka 的消息什么时候算消费过了Flink 的状态什么时候算持久化了StarRocks 的写入什么时候算生效了这三个问题的答案构成了整条链路的可靠性基础。我现在的习惯是每接一个新任务先把这三个问题在文档里写清楚再动手写代码看似慢了半天实际省下的排查时间要以周计。另外链路监控的优先级要提到和功能开发一样高一个没有监控的实时任务等于在生产环境埋了一颗不知道什么时候爆的雷。
