Flink SQL空闲状态保留时间（idle state retention time）实现原理

前言

如果要列举Flink SQL新手有可能犯的错误，笔者认为其中之一就是忘记设置空闲状态保留时间导致状态爆炸。2021年的第一篇技术文，时间很紧张，聊聊这个简单的话题吧。

为什么要设置

如果我们在数据流上进行分组查询，分组处理产生的结果(不仅仅是聚合结果)会作为中间状态存储下来。随着分组key的不断增加，状态自然也会不断膨胀。但是这些状态数据基本都有时效性，不必永久保留。例如，使用Top-N语法进行去重，重复数据的出现一般都位于特定区间内(例如一小时或一天内)，过了这段时间之后，对应的状态就不再需要了。Flink SQL提供的idle state retention time特性可以保证当状态中某个key对应的数据未更新的时间达到阈值时，该条状态被自动清理。设置方法是：

stenv.getConfig().setIdleStateRetentionTime(Time.hours(24), Time.hours(36))

注意setIdleStateRetentionTime()方法需要传入两个参数：状态的最小保留时间minRetentionTime和最大保留时间maxRetentionTime(根据实际业务决定)，且两者至少相差5分钟。为什么会有这种限制呢？看一下源码就知道了。

如何实现的

idle state retention time特性在底层以o.a.f.table.runtime.functions.CleanupState接口来表示，代码如下。

public interface CleanupState {default void registerProcessingCleanupTimer(ValueState<Long> cleanupTimeState,long currentTime,long minRetentionTime,long maxRetentionTime,TimerService timerService)throws Exception {// last registered timerLong curCleanupTime = cleanupTimeState.value();// check if a cleanup timer is registered and// that the current cleanup timer won't delete state we need to keepif (curCleanupTime == null || (currentTime + minRetentionTime) > curCleanupTime) {// we need to register a new (later) timerlong cleanupTime = currentTime + maxRetentionTime;// register timer and remember clean-up timetimerService.registerProcessingTimeTimer(cleanupTime);// delete expired timerif (curCleanupTime != null) {timerService.deleteProcessingTimeTimer(curCleanupTime);}cleanupTimeState.update(cleanupTime);}}
}

由上可知，每个key对应的最近状态清理时间会单独维护在ValueState中。如果满足以下两条件之一：

• ValueState为空(即这个key是第一次出现)
• 或者当前时间加上minRetentionTime已经超过了最近清理的时间

就用当前时间加上maxRetentionTime注册新的Timer，并将其时间戳存入ValueState，用于触发下一次清理。如果有已经过期了的Timer，则一并删除之。可见，如果minRetentionTime和maxRetentionTime的间隔设置太小，就会比较频繁地产生Timer与更新ValueState，维护Timer的成本会变大(参见之前笔者写的Timer原理文章)，所以一般建议设置间隔比较长的清理区间。

CleanupState接口的继承关系如下图所示。

可见支持空闲状态清理的Function很多，但基类都是KeyedProcessFunctionWithCleanupState抽象类。它的源码如下。

public abstract class KeyedProcessFunctionWithCleanupState<K, IN, OUT>extends KeyedProcessFunction<K, IN, OUT> implements CleanupState {private static final long serialVersionUID = 2084560869233898457L;private final long minRetentionTime;private final long maxRetentionTime;protected final boolean stateCleaningEnabled;// holds the latest registered cleanup timerprivate ValueState<Long> cleanupTimeState;public KeyedProcessFunctionWithCleanupState(long minRetentionTime, long maxRetentionTime) {this.minRetentionTime = minRetentionTime;this.maxRetentionTime = maxRetentionTime;this.stateCleaningEnabled = minRetentionTime > 1;}protected void initCleanupTimeState(String stateName) {if (stateCleaningEnabled) {ValueStateDescriptor<Long> inputCntDescriptor =new ValueStateDescriptor<>(stateName, Types.LONG);cleanupTimeState = getRuntimeContext().getState(inputCntDescriptor);}}protected void registerProcessingCleanupTimer(Context ctx, long currentTime) throws Exception {if (stateCleaningEnabled) {registerProcessingCleanupTimer(cleanupTimeState,currentTime,minRetentionTime,maxRetentionTime,ctx.timerService());}}protected boolean isProcessingTimeTimer(OnTimerContext ctx) {return ctx.timeDomain() == TimeDomain.PROCESSING_TIME;}protected void cleanupState(State... states) {for (State state : states) {state.clear();}this.cleanupTimeState.clear();}protected Boolean needToCleanupState(Long timestamp) throws IOException {if (stateCleaningEnabled) {Long cleanupTime = cleanupTimeState.value();// check that the triggered timer is the last registered processing time timer.return timestamp.equals(cleanupTime);} else {return false;}}
}

可以发现，空闲状态保留时间目前(1.12版本)仍然只支持processing time语义，并且minRetentionTime只有设为大于0的值才会生效。

KeyedProcessFunctionWithCleanupState只是提供了一些helper方法，具体发挥作用需要到实现类中去找。以计算Top-N的AppendOnlyTopNFunction为例，它的processElement()方法中会对到来的每个元素注册清理Timer：

@Override
public void processElement(RowData input, Context context, Collector<RowData> out) throws Exception {long currentTime = context.timerService().currentProcessingTime();// register state-cleanup timerregisterProcessingCleanupTimer(context, currentTime);// ......
}

而一旦Timer触发，在onTimer()方法中调用基类的cleanupState()方法来实际清理：

@Override
public void onTimer(long timestamp,OnTimerContext ctx,Collector<RowData> out) throws Exception {if (stateCleaningEnabled) {// cleanup cachekvSortedMap.remove(keyContext.getCurrentKey());cleanupState(dataState);}
}

空闲状态保留的逻辑并不仅应用在上述Function中。在Table/SQL模块中还有一个内置的触发器StateCleaningCountTrigger，它可以对窗口中的元素进行计数，并按照计数阈值或者空闲状态保留的时间阈值来清理(即FIRE_AND_PURGE)。看官可自行参考对应的源码，不再废话了。

The End

今天号称是帝都21世纪以来最冷的一天，趁早洗洗睡吧。

民那晚安。