当前位置：首页 > news >正文

Flink 广播状态（Broadcast State）实战从原理到落地

news 2025/10/12 4:15:13

一、为什么需要广播状态？

典型需求：

规则/配置会持续变更，需实时生效；
规则需要在所有并行子任务上保持一致；
业务流按 key 分区（如按用户、设备、颜色），但规则应应用到所有 key。

直接用 keyed state 不现实，因为 keyed state 的生命周期与 key 绑定；而广播状态属于 Operator State，面向算子实例本地存储，天然适合“全量分发、一致存储”。

二、核心设计与数据流

整体思路：

规则流（低吞吐） → 广播 → 在每个并行 task 内存映射为 Broadcast State；
业务流（高吞吐，通常 keyed） → 与广播流 connect → 在自定义函数中同时读取业务数据与规则，产出结果。

规则更新只允许在广播侧写入；业务侧仅只读访问，保证一致性与可重放。

三、关键 API 与最小可用示例

1）按业务键分区（示例：按颜色 Color）

KeyedStream<Item, Color> colorPartitionedStream =itemStream.keyBy((KeySelector<Item, Color>) Item::getColor);

2）定义并广播规则状态

MapStateDescriptor<String, Rule> ruleStateDescriptor =new MapStateDescriptor<>("RulesBroadcastState",BasicTypeInfo.STRING_TYPE_INFO,TypeInformation.of(new TypeHint<Rule>() {}));BroadcastStream<Rule> ruleBroadcastStream = ruleStream.broadcast(ruleStateDescriptor);

3）连接两条流 + 编写匹配逻辑

对 keyed 业务流 使用 KeyedBroadcastProcessFunction；
若业务流非 keyed，使用 BroadcastProcessFunction。

DataStream<String> output = colorPartitionedStream.connect(ruleBroadcastStream).process(new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {// 用于暂存“部分匹配”的 Item（例如已到达的第一元素，等待第二元素）private final MapStateDescriptor<String, List<Item>> partialDesc =new MapStateDescriptor<>("items",BasicTypeInfo.STRING_TYPE_INFO,new ListTypeInfo<>(Item.class));private final MapStateDescriptor<String, Rule> ruleDesc =new MapStateDescriptor<>("RulesBroadcastState",BasicTypeInfo.STRING_TYPE_INFO,TypeInformation.of(new TypeHint<Rule>() {}));@Overridepublic void processBroadcastElement(Rule rule, Context ctx, Collector<String> out) throws Exception {// 仅在广播侧可写ctx.getBroadcastState(ruleDesc).put(rule.name, rule);}@Overridepublic void processElement(Item value, ReadOnlyContext ctx, Collector<String> out) throws Exception {final MapState<String, List<Item>> partial = getRuntimeContext().getMapState(partialDesc);// 只读访问规则for (Map.Entry<String, Rule> e :ctx.getBroadcastState(ruleDesc).immutableEntries()) {String ruleName = e.getKey();Rule r = e.getValue();List<Item> buf = partial.get(ruleName);if (buf == null) buf = new ArrayList<>();// 命中第二元素 → 输出全部配对if (value.getShape() == r.second && !buf.isEmpty()) {for (Item first : buf) {out.collect("MATCH: " + first + " - " + value + " by " + ruleName);}buf.clear();}// 是第一元素 → 暂存if (value.getShape().equals(r.first)) {buf.add(value);}// 清理或回写if (buf.isEmpty()) partial.remove(ruleName);else partial.put(ruleName, buf);}}});

四、两类函数的能力边界

1）`BroadcastProcessFunction`（业务流非 keyed）

广播侧：读写 Broadcast State；
非广播侧：只读 Broadcast State；
无 Timer 能力。

2）`KeyedBroadcastProcessFunction`（业务流 keyed）

具备上述全部能力；
业务侧 ReadOnlyContext 提供 TimerService（事件/处理时间定时器），配合 onTimer() 做超时、会话窗口等；
广播侧 Context 额外提供 applyToKeyedState(...)（Java API），将函数应用于所有 key 的 keyed state（PyFlink 暂不支持）。