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 统一节点模型 CtrlNode
• 3.5 双流 union 与脏数据侧输出
• 3.6 本章小结与下章预告

3.1 两条流的格式差异

远控链路的日志分散在两个 Kafka Topic，格式完全不同。理解它们的差异是正确解析的前提。

3.1.1 L3 Topic：单行文本

Topic：cornex_remote_log_l3

2026-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 Topic：JSON

Topic：cornex_remote_log_l2

{"dcm_no":"VA24931H001186","tag":"T_L0","sessionId":"vis-rc-async-LCNTBOX0000000005-745188f0481c4ec1","time":"1791771300320"}

字段映射：

关键特征：

• 标准 JSON
• 时间是 epoch 毫秒（字符串形式），与 L3 侧按本地时区解析出的毫秒同基准
• tag 字段对应 L3 的 forward

3.1.3 两条流的关联桥梁

L3 的 requestId == L2 的 sessionId 这是把两条流关联到同一次远控的唯一桥梁。 后续 keyBy(requestId) 后，L3 和 L2 的节点会落到同一个 key， 在同一个 KeyedProcessFunction 里聚合成一条远控记录。

3.2 Kafka Source 构建

3.2.1 Source 构建方法

privatestaticKafkaSource<String>buildKafkaSource(Propertiesconfig,Stringtopic,StringgroupId){returnKafkaSource.<String>builder().setBootstrapServers(config.getProperty("kafka.bootstrap.servers")).setTopics(topic).setGroupId(groupId).setStartingOffsets(OffsetsInitializer.earliest())// dev 环境.setValueOnlyDeserializer(newSimpleStringSchema()).build();}

3.2.2 双流接入主流程

// L3 流SingleOutputStreamOperator<CtrlNode>l3Nodes=env.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 流SingleOutputStreamOperator<CtrlNode>l2Nodes=env.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，解析后输出统一的CtrlNode

3.3 关键参数配置详解

3.3.1 消费组 ID（group.id）

硬约束: group.id 必须从配置文件读取,禁硬编码 配置文件(config-prod.properties): remote.kafka.group.id=healing-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。

生产环境的正确写法：

// 生产环境: 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): · 聚合层双定时器: pt+90s 兜底 / pt+3s 延迟(第 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 统一节点模型 CtrlNode

L3 文本和 L2 JSON 格式完全不同，但解析后要输出统一的模型，才能union到一起做后续聚合。这个统一模型就是CtrlNode。

3.4.1 CtrlNode 字段定义

@Data@Builder@NoArgsConstructor@AllArgsConstructorpublicclassCtrlNodeimplementsSerializable{privatestaticfinallongserialVersionUID=1L;/** 一次远控的唯一标识(L3 的 [] 内容 == L2 的 sessionId) */privateStringrequestId;/** 来源侧: 3=L3 平台文本日志, 2=L2 平台 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 双流 union

SingleOutputStreamOperator<RemoteControlRecord>records=l3Nodes.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{publicstaticfinalOutputTag<String>L3_DIRTY_TAG=newOutputTag<String>("l3-dirty"){};publicstaticfinalOutputTag<String>L2_DIRTY_TAG=newOutputTag<String>("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(sessionId=requestId, time=epoch毫秒, tag=forward) ✓ 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（纯字符串截取，不用正则）和L2JsonParseProcessFunction（Jackson 反序列化）。重点讲三个坑：L2 的statusCode=0要归一化为1、L3_TO_APP 的中间态"指令已下发"怎么处理、脏数据怎么侧输出。解析错了，后面的耗时计算和结果判定全错。

官方参考资料

• Apache Flink Kafka Connector
• Apache Kafka Documentation — Consumer Configs

本章实战代码清单

序号	内容	说明
1	L3 文本 vs L2 JSON 格式对比	3.1
2	KafkaSource 构建方法与双流接入主流程	3.2
3	group.id / 启动模式 / Watermark / 并行度参数详解	3.3
4	CtrlNode 统一节点模型 7 字段定义	3.4
5	双流 union + 脏数据侧输出	3.5