Apache Flink双数据流窗口Join错位阻塞问题求助

使用Flink对每日数据执行1天窗口的Join操作，涉及两个Kafka数据源：

• 数据源A：数据无每日一致性保障，部分日期数据量大，部分日期无数据
• 数据源B：由定时任务收集，每日数据稳定

预处理逻辑：

• A流：过滤指定表、过滤SUCCESS状态、分配事件时间水印
• B流：过滤指定表、分配事件时间水印、按asset键分组后用1天滚动窗口去重

采用TumblingEventTimeWindows实现窗口Join，但测试发现Join函数处出现阻塞。

解决方法 #

阻塞核心原因是水印推进停滞、窗口无法触发清理、状态积压，结合代码细节，按以下步骤优化：

1. 修正数据源初始化笔误 #

代码中A、B流的env.fromSource都误用了source变量，导致两个流读取同一数据源，直接引发数据混乱和阻塞，需修正为各自的KafkaSource：

// 修正A流数据源
DataStream<ObjectNode> a = env.fromSource(A, WatermarkStrategy.noWatermarks());
// 修正B流数据源
DataStream<ObjectNode> b = env.fromSource(B, WatermarkStrategy.noWatermarks());

2. 优化水印策略，确保窗口及时触发 #

当前使用forMonotonousTimestamps()，若A流断流或数据延迟，水印无法推进，窗口会一直等待。给A流添加延迟容忍+空闲源处理：

a = a.filter(...)
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<ObjectNode>forBoundedOutOfOrderness(Duration.ofHours(2)) // 允许2小时数据延迟
            .withTimestampAssigner((objectNode, l) -> {
                String updateTime = objectNode.get("value").get("data").get("db_update_time").asText();
                return DateTime.parse(updateTime, DateTimeFormat.dateFormatSecond).getMillis();
            })
            .withIdleness(Duration.ofMinutes(10)) // 10分钟无数据则标记为空闲，自动推进水印
    )
    .filter(...);

B流因数据规整，可保留forMonotonousTimestamps()，但建议同样添加延迟容忍，避免个别数据迟到阻塞窗口。

3. 优化B流去重逻辑，减少状态积压 #

原用ProcessWindowFunction去重效率低且易积压状态，改用AggregateFunction实现更高效的去重：

b = b.filter(...)
    .assignTimestampsAndWatermarks(...)
    .keyBy(o -> o.get("value").get("data").get("base").asText())
    .window(TumblingEventTimeWindows.of(Time.days(1)))
    .aggregate(new AggregateFunction<ObjectNode, ObjectNode, ObjectNode>() {
        @Override
        public ObjectNode createAccumulator() {
            return null;
        }

        @Override
        public ObjectNode add(ObjectNode value, ObjectNode accumulator) {
            // 保留窗口内insert_time最新的一条数据
            if (accumulator == null) return value;
            long currentTime = DateTime.parse(value.get("value").get("data").get("insert_time").asText(), DateTimeFormat.dateFormatSecond).getMillis();
            long accTime = DateTime.parse(accumulator.get("value").get("data").get("insert_time").asText(), DateTimeFormat.dateFormatSecond).getMillis();
            return currentTime > accTime ? value : accumulator;
        }

        @Override
        public ObjectNode getResult(ObjectNode accumulator) {
            return accumulator;
        }

        @Override
        public ObjectNode merge(ObjectNode a, ObjectNode b) {
            return a;
        }
    });

4. 改用Interval Join替代滚动窗口Join #

滚动窗口Join要求两个流事件时间落在同一窗口，当A流某一天无数据时，窗口会因水印停滞挂起。改用Interval Join更适配A流不稳定的场景：

// 关联B流insert_time前后1天内的A流数据（可根据业务调整时间范围）
DataStream<Tuple3<String, BigDecimal, BigDecimal>> joinedStream = a
    .keyBy(o -> o.get("value").get("data").get("currency").asText())
    .intervalJoin(b.keyBy(o -> o.get("value").get("data").get("quote").asText()))
    .between(Time.days(-1), Time.days(0))
    .process(new ProcessJoinFunction<ObjectNode, ObjectNode, Tuple3<String, BigDecimal, BigDecimal>>() {
        @Override
        public void processElement(ObjectNode a, ObjectNode b, Context ctx, Collector<Tuple3<String, BigDecimal, BigDecimal>> out) throws Exception {
            // 实现业务Join逻辑
            String key = a.get("value").get("data").get("currency").asText();
            BigDecimal valA = new BigDecimal(a.get("value").get("data").get("amount").asText());
            BigDecimal valB = new BigDecimal(b.get("value").get("data").get("price").asText());
            out.collect(Tuple3.of(key, valA, valB));
        }
    });

Interval Join会基于水印自动清理过期状态，避免因某一方无数据导致的阻塞。

5. 配置状态TTL，防止无限积压 #

在代码中添加全局状态TTL，限制状态保留时长：

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofDays(3)) // 保留3天状态
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();

// 在使用状态的算子（如去重窗口、Join算子）的open方法中启用
@Override
public void open(Configuration parameters) throws Exception {
    ValueState<ObjectNode> state = getRuntimeContext().getState(
        new ValueStateDescriptor<>("state", ObjectNode.class)
            .enableTimeToLive(ttlConfig)
    );
}

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