Flink车联网远控预警系统实战 —— 第 3 章 Kafka 双流接入L3 文本 L2 JSON章节定位第二篇开篇。第 2 章搭好了技术骨架从本章开始写代码。远控预警的数据源是 Kafka 上的两条流——L3 平台的文本日志和 L2 平台的 JSON 日志格式完全不同。本章解决三个问题怎么消费这两条流、怎么配 Kafka Source 的关键参数消费组、启动模式、Watermark、怎么把两条流 union 成统一的CtrlNode模型。这是整个计算链路的起点解析对了后面才不会错。版本基线Apache Flink 2.x Kafka 3.x章节导读3.1 两条流的格式差异3.2 Kafka Source 构建3.3 关键参数配置详解3.4 统一节点模型 CtrlNode3.5 双流 union 与脏数据侧输出3.6 本章小结与下章预告3.1 两条流的格式差异远控链路的日志分散在两个 Kafka Topic格式完全不同。理解它们的差异是正确解析的前提。3.1.1 L3 Topic单行文本Topiccornex_remote_log_l32026-10-12 10:15:00.100 [vis-rc-async-LCNTBOX0000000005-745188f0481c4ec1] recv from app RCSXX APP_TO_L3 {symbol:VA24931H001186,cmdType:LOCK}格式规律┌──────────────────────────────┬───────────────────────────────────────────┐ │ 字段 │ 提取规则 │ ├──────────────────────────────┼───────────────────────────────────────────┤ │ timestamp时间戳 │ 前 23 个字符 │ │ │ 格式 yyyy-MM-dd HH:mm:ss.SSS │ ├──────────────────────────────┼───────────────────────────────────────────┤ │ requestId │ 第一对 [ ] 之间的内容 │ │ │ 为空或 - → 直接丢弃 │ ├──────────────────────────────┼───────────────────────────────────────────┤ │ forward节点名 │ RCSXX 之后第一个空格前的 token │ │ │ APP_TO_L3 / L3_TO_L2 / L2_TO_L3 / L3_TO_APP│ ├──────────────────────────────┼───────────────────────────────────────────┤ │ data业务数据 │ forward 之后的整段 JSON │ │ │ 含 symbol(VIN) / result 等 │ └──────────────────────────────┴───────────────────────────────────────────┘关键特征纯文本不是 JSON时间戳是格式化字符串不是 epoch 毫秒requestId 藏在中括号里RCSXX是固定分隔标记3.1.2 L2 TopicJSONTopiccornex_remote_log_l2{dcm_no:VA24931H001186,tag:T_L0,sessionId:vis-rc-async-LCNTBOX0000000005-745188f0481c4ec1,time:1791771300320}字段映射JSON 字段含义对应 CtrlNode 字段sessionId会话 IDrequestId等于 L3 的 requestIdtime时间epoch 毫秒字符串timestamptag节点名forwardF_L3/T_SMS/F_SMS/T_L0/F_L0/T_L3dcm_no车架号vinresult/resultMessageF_L0 的车机结果resultCode/resultMsgcode/descT_L3 的 L2 结果resultCode/resultMsg关键特征标准 JSON时间是 epoch 毫秒字符串形式与 L3 侧按本地时区解析出的毫秒同基准tag 字段对应 L3 的 forward3.1.3 两条流的关联桥梁L3 的 requestId L2 的 sessionId 这是把两条流关联到同一次远控的唯一桥梁。 后续 keyBy(requestId) 后L3 和 L2 的节点会落到同一个 key 在同一个 KeyedProcessFunction 里聚合成一条远控记录。3.2 Kafka Source 构建3.2.1 Source 构建方法privatestaticKafkaSourceStringbuildKafkaSource(Propertiesconfig,Stringtopic,StringgroupId){returnKafkaSource.Stringbuilder().setBootstrapServers(config.getProperty(kafka.bootstrap.servers)).setTopics(topic).setGroupId(groupId).setStartingOffsets(OffsetsInitializer.earliest())// dev 环境.setValueOnlyDeserializer(newSimpleStringSchema()).build();}3.2.2 双流接入主流程// L3 流SingleOutputStreamOperatorCtrlNodel3Nodesenv.fromSource(buildKafkaSource(config,config.getProperty(remote.kafka.l3.topic),config.getProperty(remote.kafka.group.id)_l3),WatermarkStrategy.noWatermarks(),Remote L3 Kafka Source).uid(remote-l3-source).process(newL3TextParseProcessFunction()).uid(remote-l3-parse).name(L3 Text Parse);// L2 流SingleOutputStreamOperatorCtrlNodel2Nodesenv.fromSource(buildKafkaSource(config,config.getProperty(remote.kafka.l2.topic),config.getProperty(remote.kafka.group.id)_l2),WatermarkStrategy.noWatermarks(),Remote L2 Kafka Source).uid(remote-l2-source).process(newL2JsonParseProcessFunction()).uid(remote-l2-parse).name(L2 JSON Parse);要点L3 和 L2 用不同的 group.id主 group _l3/_l2后缀避免同 group 下两个消费者互相 rebalance 抢分区两个 Source 都设了.uid()保证作业重启/扩缩容时状态恢复的一致性解析函数L3TextParse / L2JsonParse紧跟 Source解析后输出统一的CtrlNode3.3 关键参数配置详解3.3.1 消费组 IDgroup.id硬约束: group.id 必须从配置文件读取,禁硬编码 配置文件(config-prod.properties): remote.kafka.group.idhealing-remote-warning-prod 代码中: L3 用: remote.kafka.group.id _l3 L2 用: remote.kafka.group.id _l2 为什么 L3/L2 要分后缀: Kafka 同一个 group 下的消费者会分配不同分区。 如果 L3 和 L2 用同一个 group.id,而它们订阅的是不同 topic, Kafka 会把两个 topic 的分区混在一起分配, 可能导致某个消费者被分到另一个 topic 的分区而无数据可消费。 用后缀隔离后,两个 group 独立消费各自的 topic,互不干扰。3.3.2 启动模式scan.startup.mode这是最容易踩坑的参数。当前代码 dev 环境用earliest但生产环境必须用group-offsets。启动模式行为适用场景风险earliest从最早的消息开始消费开发自测、消费历史数据重启后重消费全量数据group-offsets从消费组已提交的位移继续生产环境需先有消费记录latest只消费启动后的新消息不关心历史数据启动前的数据丢失timestamp从指定时间戳开始回溯特定时间段需指定时间生产环境的正确写法// 生产环境: group-offsets.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))// group 无位移时回退到最早硬约束: 生产环境 scan.startup.mode group-offsets 违反后果: 用 earliest 提交到生产后,作业重启会从最早消息重新消费, 把几个月的历史数据全跑一遍,造成: ① Doris 写入压力暴增 ② 大量重复数据(虽然 Unique Key 会去重,但浪费算力) ③ 告警统计被历史数据污染 当前代码 dev 用 earliest 是为了消费 topic 里已有的真实数据做验证, 上线前必须改成 group-offsets。3.3.3 Watermark 策略当前代码用的是WatermarkStrategy.noWatermarks().fromSource(kafkaSource,WatermarkStrategy.noWatermarks(),...)为什么不用事件时间 Watermark本作业后续的计算全部基于处理时间(Processing Time): · 聚合层双定时器: pt90s 兜底 / pt3s 延迟(第 5 章) · 统计层整点触发: 每分钟墙钟整点(第 10 章) 不涉及事件时间窗口,所以不需要 Watermark。 如果硬加 Watermark,反而需要处理乱序、迟到等问题,增加复杂度。 如果未来要加事件时间窗口(如按远控发生时间统计), 再引入 Watermark,乱序容忍 5~30s。3.3.4 反序列化器.setValueOnlyDeserializer(newSimpleStringSchema())只用SimpleStringSchema把 Kafka 的 value 反序列化成 String不在 Source 层做 JSON 解析。解析逻辑放在后续的ProcessFunction里原因Source 层只管把字节流变成字符串职责单一解析失败的处理侧输出、计数放在 ProcessFunction 里更灵活如果在 Source 层解析失败整条流会报错中断放在 ProcessFunction 里可以侧输出脏数据主流程不中断3.3.5 并行度Kafka Source 的并行度由 topic 的分区数决定, 不受 table.exec.default-parallelism 控制。 如果 L3 topic 有 16 个分区,L3 Source 最多 16 并行度。 要提高吞吐,先扩 Kafka 分区数,再调 Flink 并行度。 硬约束: Source 并行度 ≤ Kafka 分区数 超过分区数的 Source 子任务会空转,浪费 Slot。3.4 统一节点模型 CtrlNodeL3 文本和 L2 JSON 格式完全不同但解析后要输出统一的模型才能union到一起做后续聚合。这个统一模型就是CtrlNode。3.4.1 CtrlNode 字段定义DataBuilderNoArgsConstructorAllArgsConstructorpublicclassCtrlNodeimplementsSerializable{privatestaticfinallongserialVersionUID1L;/** 一次远控的唯一标识(L3 的 [] 内容 L2 的 sessionId) */privateStringrequestId;/** 来源侧: 3L3 平台文本日志, 2L2 平台 JSON 日志 */privateintside;/** 节点方向: L3 侧 forward, L2 侧 tag */privateStringforward;/** 节点事件时间(本地时间毫秒,两侧同基准直接相减) */privatelongtimestamp;/** VIN: L3 取 data.symbol, L2 取 dcm_no */privateStringvin;/** 该节点携带的结果码(无结果节点为 null) */privateStringresultCode;/** 该节点携带的结果描述(无结果节点为 null) */privateStringresultMsg;}3.4.2 字段设计思路为什么是这 7 个字段? requestId —— 关联 L3 和 L2 的桥梁,后续 keyBy 的 key side —— 区分来源,聚合层需要知道节点来自哪一侧 forward —— 节点名,聚合层据此写入对应锚点(APP_TO_L3→start 等) timestamp —— 时间锚点,耗时计算的基础 vin —— 车辆标识,统计层按 VIN 去重 resultCode —— 结果码,失败判定的依据 resultMsg —— 结果描述,用于展示和兜底 这 7 个字段覆盖了后续所有计算需要的信息, 多一个冗余,少一个不够。3.4.3 时间基准统一L3 侧: 前 23 字符 2026-10-12 10:15:00.100 → LocalDateTime.parse(...).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() → 本地时间毫秒 L2 侧: 1791771300320(epoch 毫秒字符串) → Long.parseLong(time) → 本地时间毫秒 两侧都是本地时间毫秒,相减时时区偏移自动抵消, 保证耗时计算准确。 硬约束: TaskManager JVM 时区必须是 Asia/Shanghai L3 侧的时间解析用 ZoneId.systemDefault(), 如果 JVM 时区不是东八区,解析出的毫秒会偏 8 小时, 与 L2 侧的 epoch 毫秒不同基准,耗时计算全错。 且这是静默错误——不报错,只是数据不对。3.5 双流 union 与脏数据侧输出3.5.1 双流 unionSingleOutputStreamOperatorRemoteControlRecordrecordsl3Nodes.union(l2Nodes).keyBy(CtrlNode::getRequestId).process(newRemoteControlAggregateKeyedProcessFunction(...));union 的语义: L3 节点流 ─┐ ├─→ 混合流 ── keyBy(requestId) ── 聚合 L2 节点流 ─┘ 两条流的类型必须相同(都是 CtrlNode),才能 union。 union 不做去重、不做排序,只是简单地把两条流拼在一起。 后续的 keyBy 会把同一个 requestId 的节点路由到同一个 subtask, 在 KeyedProcessFunction 里按 forward 语义写入对应锚点。3.5.2 脏数据侧输出L3 文本和 L2 JSON 解析都可能失败格式不对、缺字段等。脏数据不能让作业挂掉而是侧输出单独处理。// 侧输出标签定义publicfinalclassRemoteParseTags{publicstaticfinalOutputTagStringL3_DIRTY_TAGnewOutputTagString(l3-dirty){};publicstaticfinalOutputTagStringL2_DIRTY_TAGnewOutputTagString(l2-dirty){};}// 脏数据收集l3Nodes.getSideOutput(RemoteParseTags.L3_DIRTY_TAG).print([remote-l3-dirty]);l2Nodes.getSideOutput(RemoteParseTags.L2_DIRTY_TAG).print([remote-l2-dirty]);脏数据处理纪律: □ 脏数据走侧输出,绝不抛异常中断主流程 □ 当前阶段 print 观察,生产可改为计数指标或写入 Doris 脏数据表 □ 脏数据不参与后续计算,也不影响正常数据 □ 典型脏数据: 无 [] 的 L3 行、无 RCSXX 的行、JSON 解析失败、 无 sessionId/time/tag 的 L2 消息3.5.3 完整数据流Kafka L3 (文本) ──▶ L3TextParse ──┐ ├─ union ── keyBy(requestId) ──▶ 聚合 Kafka L2 (JSON) ──▶ L2JsonParse ──┘ │ 脏数据侧输出(print)3.6 本章小结与下章预告本章小结┌──────────────────────────────────────────────────────────────┐ │ 第 3 章 要点回顾 │ └──────────────────────────────────────────────────────────────┘ ✓ 双流格式: L3 单行文本(前23字符时间 []requestId RCSXX forward) vs L2 JSON(sessionIdrequestId, timeepoch毫秒, tagforward) ✓ Kafka Source: KafkaSource.builder() 构建, L3/L2 用不同 group.id(主group后缀)避免 rebalance 冲突 ✓ 关键参数: · group.id 配置化,禁硬编码 · 生产用 group-offsets,dev 可用 earliest · Watermark 用 noWatermarks(后续全是处理时间) · Source 并行度 ≤ Kafka 分区数 ✓ CtrlNode 统一模型: requestId/side/forward/timestamp/vin/ resultCode/resultMsg 7 字段,覆盖后续所有计算需求 ✓ 时间基准: L3 文本时间用 systemDefault 解析为本地毫秒, L2 直接用 epoch 毫秒,两侧同基准可直接相减; JVM 时区必须 Asia/Shanghai ✓ 双流 union: 类型一致才能 union,keyBy(requestId) 路由; 脏数据走侧输出 print,不中断主流程下章预告第 4 章 Flink 解析层纯字符串截取与结果归一化数据接入了该解析了。本章写两个 ProcessFunction——L3TextParseProcessFunction纯字符串截取不用正则和L2JsonParseProcessFunctionJackson 反序列化。重点讲三个坑L2 的statusCode0要归一化为1、L3_TO_APP 的中间态指令已下发怎么处理、脏数据怎么侧输出。解析错了后面的耗时计算和结果判定全错。官方参考资料Apache Flink Kafka ConnectorApache Kafka Documentation — Consumer Configs本章实战代码清单序号内容说明1L3 文本 vs L2 JSON 格式对比3.12KafkaSource 构建方法与双流接入主流程3.23group.id / 启动模式 / Watermark / 并行度参数详解3.34CtrlNode 统一节点模型 7 字段定义3.45双流 union 脏数据侧输出3.5
