Flink基于ProcessWindowFunction的去重及延迟数据处理方案优化咨询

• 状态非托管且绑定维度错误：你在WindowedFilter中定义的HashSet<Long> previousRecordTimestamps是堆内存中的普通成员变量，不属于Flink托管的Keyed状态，既无法在作业故障重启、扩缩容时保证数据一致性，也没有和key/窗口绑定，不同key、不同窗口的时间戳会存在同一个集合中，会出现跨key、跨窗口的误过滤，完全不符合Keyed流的处理逻辑。
• 去重规则不合理：仅用timestamp作为去重依据，同一个时间戳完全可能存在多条不同的业务记录，会导致正常数据被误过滤；如果是相同记录重复，仅靠时间戳也无法保证唯一匹配，可能出现漏去重。
• 内存泄漏风险：HashSet没有清理逻辑，随着作业运行时间拉长，存储的时间戳会越来越多，最终导致算子内存溢出。
• 窗口触发逻辑导致重复处理：你使用ContinuousEventTimeTrigger每3秒触发一次窗口，每次触发都会全量迭代窗口内的所有记录，不仅性能开销大，还会导致相同记录多次向下游输出，下游第二个聚合窗口如果没有做去重处理，最终统计结果会偏大。
• 延迟数据处理缺失：没有配置窗口的allowedLateness允许延迟，水位线超过窗口结束时间后到达的延迟数据会被直接丢弃，且没有对应的兜底处理链路。
• 冗余的分组与窗口开销：连续两次使用相同的MetricGrouper做keyBy，且多了一层60秒的窗口处理，额外增加了数据shuffle和窗口管理的性能开销。

• 替换为Flink托管状态：删除堆上的HashSet，改用MapState或者ValueState存储去重标记，状态和当前处理的key绑定，同时配置状态TTL，TTL时长设置为业务允许的最大数据延迟时间，到期自动清理状态，既避免内存泄漏，也保证故障恢复时状态一致。
  示例参考：

  // 在WindowedFilter中定义状态
  private MapState<String, Boolean> deduplicateState;
  @Override
  public void open(Configuration parameters) {
      MapStateDescriptor<String, Boolean> descriptor = new MapStateDescriptor<>("deduplicateState", Types.STRING, Types.BOOLEAN);
      // 设置状态TTL，比如允许最大延迟1小时
      StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(1))
              .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
              .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
              .build();
      descriptor.enableTimeToLive(ttlConfig);
      deduplicateState = getRuntimeContext().getMapState(descriptor);
  }
  

• 修正去重主键：使用业务唯一标识作为去重键，比如消息的唯一ID、或者timestamp + 业务维度字段的组合值，避免仅用时间戳导致的误过滤、漏过滤。
• 调整窗口触发逻辑：如果不需要实时输出部分窗口结果，可移除ContinuousEventTimeTrigger，使用默认的窗口触发逻辑，仅在窗口结束时触发一次；如果需要低延迟输出，配合增量聚合函数（ReduceFunction/AggregateFunction）使用，避免每次触发全量迭代所有数据。
• 补充延迟数据处理：给窗口配置合理的allowedLateness，超过允许延迟时间的数据输出到侧流做兜底处理，避免数据丢失。
• 简化链路逻辑：可将去重逻辑从窗口中拆出，放到前置的KeyedProcessFunction中实现，去掉第一层冗余的60秒窗口，减少不必要的性能开销。
• 大流量场景优化：如果数据量非常大、允许极小概率的去重误差，可使用布隆过滤器实现去重，大幅降低状态存储的内存开销。

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