前言

如果要列舉Flink SQL新手有可能犯的錯誤，筆者認(rèn)為其中之一就是忘記設(shè)置空閑狀態(tài)保留時間導(dǎo)致狀態(tài)爆炸。

為什么要設(shè)置

如果我們在數(shù)據(jù)流上進(jìn)行分組查詢，分組處理產(chǎn)生的結(jié)果(不僅僅是聚合結(jié)果)會作為中間狀態(tài)存儲下來。隨著分組key的不斷增加，狀態(tài)自然也會不斷膨脹。但是這些狀態(tài)數(shù)據(jù)基本都有時效性，不必永久保留。例如，使用Top-N語法進(jìn)行去重，重復(fù)數(shù)據(jù)的出現(xiàn)一般都位于特定區(qū)間內(nèi)(例如一小時或一天內(nèi))，過了這段時間之后，對應(yīng)的狀態(tài)就不再需要了。Flink SQL提供的idle state retention time特性可以保證當(dāng)狀態(tài)中某個key對應(yīng)的數(shù)據(jù)未更新的時間達(dá)到閾值時，該條狀態(tài)被自動清理。設(shè)置方法是：

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

注意setIdleStateRetentionTime()方法需要傳入兩個參數(shù)：狀態(tài)的最小保留時間minRetentionTime和最大保留時間maxRetentionTime(根據(jù)實際業(yè)務(wù)決定)，且兩者至少相差5分鐘。為什么會有這種限制呢？看一下源碼就知道了。

如何實現(xiàn)的

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 timer
        Long curCleanupTime = cleanupTimeState.value();

        // check if a cleanup timer is registered and
        // that the current cleanup timer won't delete state we need to keep
        if (curCleanupTime == null || (currentTime + minRetentionTime) > curCleanupTime) {
            // we need to register a new (later) timer
            long cleanupTime = currentTime + maxRetentionTime;
            // register timer and remember clean-up time
            timerService.registerProcessingTimeTimer(cleanupTime);
            // delete expired timer
            if (curCleanupTime != null) {
                timerService.deleteProcessingTimeTimer(curCleanupTime);
            }
            cleanupTimeState.update(cleanupTime);
        }
    }
}

由上可知，每個key對應(yīng)的最近狀態(tài)清理時間會單獨維護(hù)在ValueState中。如果滿足以下兩條件之一：

• ValueState為空(即這個key是第一次出現(xiàn))

• 或者當(dāng)前時間加上minRetentionTime已經(jīng)超過了最近清理的時間

就用當(dāng)前時間加上maxRetentionTime注冊新的Timer，并將其時間戳存入ValueState，用于觸發(fā)下一次清理。如果有已經(jīng)過期了的Timer，則一并刪除之。可見，如果minRetentionTime和maxRetentionTime的間隔設(shè)置太小，就會比較頻繁地產(chǎn)生Timer與更新ValueState，維護(hù)Timer的成本會變大(參見之前筆者寫的Timer原理文章)，所以一般建議設(shè)置間隔比較長的清理區(qū)間。

CleanupState接口的繼承關(guān)系如下圖所示。

可見支持空閑狀態(tài)清理的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 timer
    private 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;
        }
    }
}

可以發(fā)現(xiàn)，空閑狀態(tài)保留時間目前仍然只支持processing time語義，并且minRetentionTime只有設(shè)為大于0的值才會生效。

KeyedProcessFunctionWithCleanupState只是提供了一些helper方法，具體發(fā)揮作用需要到實現(xiàn)類中去找。以計算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 timer
    registerProcessingCleanupTimer(context, currentTime);
    // ......
}

而一旦Timer觸發(fā)，在onTimer()方法中調(diào)用基類的cleanupState()方法來實際清理：

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

空閑狀態(tài)保留的邏輯并不僅應(yīng)用在上述Function中。在Table/SQL模塊中還有一個內(nèi)置的觸發(fā)器StateCleaningCountTrigger，它可以對窗口中的元素進(jìn)行計數(shù)，并按照計數(shù)閾值或者空閑狀態(tài)保留的時間閾值來清理(即FIRE_AND_PURGE)?？垂倏勺孕袇⒖紝?yīng)的源碼。