Flink无限DataStream使用时间窗口过滤重复数据异常问题求解

问题根因 #

1. 窗口参数配置错误：你当前使用的滑动窗口配置为窗口大小10秒、滑动步长5分钟，两个相邻窗口之间存在4分50秒的时间间隙，大部分消息根本不会被同一个窗口覆盖，同key的相邻消息很容易被拆分到不同窗口甚至落入间隙无法处理，直接导致去重逻辑失效。
2. 如果你希望用滑动窗口覆盖全部10秒内的重复数据，滑动步长必须小于等于窗口大小，否则必然出现间隙或数据拆分问题。同时你使用的处理时间语义受系统时钟、任务调度延迟影响，也可能导致同业务时间的消息跨窗口拆分。

适配方案 #

方案1：修正窗口配置（适合允许端到端10秒延迟的场景） #

直接替换为10秒大小的滚动处理时间窗口即可，滚动窗口无重叠无间隙，10秒内的同key数据都会被分配到同一个窗口，仅输出第一条即可实现去重：

DataStream<Row> outputStream = inputStream
                .keyBy(new MyKeySelector())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .process(new DuplicateFilter());

注：该方案需要等待窗口触发才会输出数据，最大延迟为10秒。

方案2：KeyedProcessFunction + 状态TTL（推荐，低延迟无冗余） #

该方案来一条处理一条，仅保留10秒内首次出现的key对应数据，延迟为毫秒级，状态会自动过期清理不会无限膨胀：

public class DedupFunction extends KeyedProcessFunction<Tuple2<String, String>, Row, Row> {
    private final ValueStateDescriptor<Boolean> dedupStateDesc = 
        new ValueStateDescriptor<>("dedup_state", Boolean.class);

    @Override
    public void open(Configuration parameters) {
        // 配置状态10秒过期
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.seconds(10))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build();
        dedupStateDesc.enableTimeToLive(ttlConfig);
    }

    @Override
    public void processElement(Row value, Context ctx, Collector<Row> out) throws Exception {
        ValueState<Boolean> dedupState = getRuntimeContext().getState(dedupStateDesc);
        if (dedupState.value() == null) {
            dedupState.update(true);
            out.collect(value);
        }
    }
}

调用逻辑：

DataStream<Row> outputStream = inputStream
    .keyBy(new MyKeySelector())
    .process(new DedupFunction());

如果你的场景需要更精准的业务时间去重，只需引入事件时间水位线，将状态TTL调整为事件时间语义即可。

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