Flink 中的检查点随时间增加

Question

聚合到this question 我仍然不清楚为什么我的 Flink 作业的检查点会随着时间的推移而增长，而目前，在运行大约 7 天的时间里，这些检查点从未达到稳定状态。 我目前使用的是 Flink 1.10 版本，FS State Backend 作为我的工作无法承受使用 RocksDB 的延迟成本。

查看检查点在 7 天内的演变： 假设我在我的所有有状态操作符中的状态 TTL 配置为一小时或更长，在一种情况下为一天：

public static final StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(1))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .cleanupFullSnapshot().build();

在我看来，所有进入状态的对象都将在过期时间后被清理，因此应该减少检查点的大小，并且我们预计每天的数据量或多或少相同。

另一方面，我们有一条流量曲线，它在一天中的几个小时内有更多的传入数据，但深夜流量下降，所有进入过期状态的对象都应该被清理，导致检查点大小应该减少而不是保持相同的大小，直到流量再次上升。

让我们看一下这个用例的代码示例：

DataStream<Event> stream = addSource(source);
KeyedStream<Event, String> keyedStream = stream.filter((FilterFunction<Event>) event ->
                    apply filters here;))
                    .name("Events filtered")
                    .keyBy(k -> k.rType.equals("something") ? k.id1 : k.id2);
keyedStream.flatMap(new MyFlatMapFunction())


public class MyFlatMapFunction extends RichFlatMapFunction<Event, Event>{
private final MapStateDescriptor<String, Event> descriptor = new MapStateDescriptor<>("prev_state", String.class, Event.class);
private MapState<String, Event> previousState;

@Override
    public void open(Configuration parameters) {
        /*ttlConfig described above*/
        descriptor.enableTimeToLive(ttlConfig);
        previousState = getRuntimeContext().getMapState(descriptor);
    }

@Override
    public void flatMap(Event event, Collector<Event> collector) throws Exception {
      final String key = event.rType.equals("something") ? event.id1 : event.id2;
      Event previous = previousState.get(key);
      if(previous != null){
        /*something done here*/
      }else /*something done here*/
        previousState.put(key, previous);
        collector.collect(previous);
 }
}

这些或多或少是用例的结构，还有一些使用 Windows（时间窗口或会话窗口）的结构

问题：

• 我在这里做错了什么？
• 状态是否在到期时被清理，这种情况与其余用例相同吗？
• 如果检查点工作不正常，有什么办法可以帮助我修复检查点的大小？
• 这种行为正常吗？

亲切的问候！

Answer 1

在这段代码中，您似乎只是在写回已经存在的状态，这仅用于重置 TTL 计时器。这可以解释为什么状态没有过期。

Event previous = previousState.get(key);
if (previous != null) {
  /*something done here*/
} else
  previousState.put(key, previous);

看来您应该使用ValueState 而不是MapState。 ValueState 有效地提供了一个分片键/值存储，其中键是用于在 keyBy 中对流进行分区的键。 MapState 为您提供每个键的嵌套映射，而不是单个值。但是，由于您在 flatMap 中使用的键与您最初用于对流进行键控的键相同，因此键分区的 ValueState 似乎就是您所需要的。