Flink中带水印流与无水印流的连接方案咨询

场景回顾 #

流A具备事件时间戳和水印，流B是定期API拉取的无时间戳/水印的维度数据，需要将流B与流A连接以丰富流A，且最终流的水印由流A主导（避免流B拖低整体水印）。

现有方案分析 #

方案1：给流B发送最大水印 #

• 实现思路：给流B的每条数据分配一个极大值时间戳（如Long.MAX_VALUE），并发送对应最大水印。
• 问题：强行赋予不符合语义的时间戳，违背事件时间的设计初衷；若流B后续有数据更新，这种方式显得生硬且无必要。

方案2：每次发射后标记流B为临时空闲 #

• 实现思路：在流B的数据源/水印发射器中，每次完成数据发射后调用markAsTemporarilyIdle()，标记流B暂时无数据。
• 可行性：Flink在计算水印时会忽略处于空闲状态的流，因此整体水印将由流A决定；当流B下次拉取并发射数据时，水印发射器会自动恢复工作。
• 注意点：需确保流B的水印发射器正确实现空闲标记逻辑，若流B无事件时间，需先为其分配处理时间作为时间戳（或直接使用处理时间语义）。

更优方案：将流B作为广播流 + 处理时间语义 #

由于流B是定期更新的维度数据，更适合采用广播流的方式处理：

1. 将流B转换为广播流，把维度数据广播到下游所有算子实例，确保每个流A的分区都能获取到最新的维度信息。
2. 为流B配置处理时间语义，无需为其设置事件时间戳和水印。此时流B不会生成事件时间水印，Flink在连接流A（事件时间）和流B（处理时间）时，仅会以流A的水印作为整体水印。

代码示例（核心逻辑） #

// 流A：具备事件时间戳和水印
DataStream<StreamAData> streamA = ...;

// 流B：定期API拉取，配置处理时间语义
DataStream<StreamBData> streamB = env.addSource(new PeriodicAPISource())
    .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()
        .withTimestampAssigner((event, timestamp) -> System.currentTimeMillis())); // 用处理时间作为时间戳

// 将流B转换为广播流
MapStateDescriptor<String, StreamBData> broadcastStateDesc = new MapStateDescriptor<>(
    "broadcast-dim",
    BasicTypeInfo.STRING_TYPE_INFO,
    TypeInformation.of(StreamBData.class)
);
BroadcastStream<StreamBData> broadcastStreamB = streamB.broadcast(broadcastStateDesc);

// 连接流A与广播流B，用流B数据丰富流A
DataStream<EnrichedData> enrichedStream = streamA
    .connect(broadcastStreamB)
    .process(new BroadcastProcessFunction<StreamAData, StreamBData, EnrichedData>() {
        @Override
        public void processElement(StreamAData value, ReadOnlyContext ctx, Collector<EnrichedData> out) throws Exception {
            // 从广播状态中获取最新的流B维度数据
            StreamBData dimData = ctx.getBroadcastState(broadcastStateDesc).get("dim-key");
            // 丰富流A数据并输出
            out.collect(new EnrichedData(value, dimData));
        }

        @Override
        public void processBroadcastElement(StreamBData value, Context ctx, Collector<EnrichedData> out) throws Exception {
            // 更新广播状态，保存最新的维度数据
            ctx.getBroadcastState(broadcastStateDesc).put("dim-key", value);
        }
    });

方案选择建议 #

• 若流B数据量小、更新频率低，**方案2（标记临时空闲）**可快速实现；
• 若流B是需要全局共享的维度数据，广播流+处理时间语义是更优雅、更符合Flink设计理念的方案，既能保证维度数据的全局一致性，又能完全避免流B对整体水印的影响。

内容的提问来源于stack exchange，提问作者Asaad27