Flink 1.4.1堆后端Savepoint状态持续增长问题求助

Flink 1.4.1尚未引入State TTL特性，默认窗口算子也不会自动清理已关闭窗口的状态，可通过以下方式强制删除窗口触发后的状态：

1. 自定义触发器，触发后清理状态 #

扩展默认窗口触发器（如EventTimeTrigger），在窗口触发计算完成后手动清理关联状态：

public class CleanupAfterTrigger extends EventTimeTrigger {
    @Override
    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        TriggerResult result = super.onElement(element, timestamp, window, ctx);
        // 窗口触发后执行清理
        if (result == TriggerResult.FIRE_AND_PURGE) {
            ValueState<YourAggType> aggState = ctx.getPartitionedState(new ValueStateDescriptor<>("aggState", YourAggType.class));
            aggState.clear();
        }
        return result;
    }
}

在窗口算子中指定该触发器：

stream.keyBy(...)
      .timeWindow(Time.seconds(10))
      .trigger(new CleanupAfterTrigger())
      .reduce(...);

2. 使用ProcessWindowFunction手动清理状态 #

如果使用ProcessWindowFunction，可在处理完窗口结果后，通过上下文直接清理当前窗口的所有状态：

public class CleanupProcessWindowFunction extends ProcessWindowFunction<InputType, OutputType, KeyType, TimeWindow> {
    private transient ValueState<YourAggType> aggState;

    @Override
    public void open(Configuration parameters) throws Exception {
        aggState = getRuntimeContext().getState(new ValueStateDescriptor<>("aggState", YourAggType.class));
    }

    @Override
    public void process(KeyType key, Context context, Iterable<InputType> elements, Collector<OutputType> out) throws Exception {
        // 执行窗口聚合逻辑
        YourAggType result = ...;
        out.collect(result);
        
        // 清理当前窗口的状态
        aggState.clear();
        // 若存在其他类型状态（如ListState），同样调用clear()方法
    }
}

3. 调整作业配置缓解状态累积 #

• 将窗口的allowedLateness设为0（若无需处理延迟数据），缩短窗口的保留时长
• 合理设置state.backend.heap.memory.size参数，避免内存溢出，但此方式仅能缓解，无法彻底解决状态累积问题

4. 手动遍历清理已关闭窗口状态 #

在定期触发Savepoint前，手动遍历状态存储，识别并清理已关闭的窗口状态。此方式需要对Flink状态API有一定了解，操作时需严格校验窗口活跃状态，避免误删有效数据。

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