Flink源码分析系列文档目录

Evictor

为何使用Evictor

Evictor的作用是在Flink进行计算之前移除元素。

以如下使用场景为例：stream每次进入一个元素(CountTrigger， maxCount设置为1)的时候获取最近2小时内的数据。这种情况下可以使用Flink提供的EventTimeSessionWindows。EventTimeSessionWindows使用EventTimeTrigger（当前时间为window的maxTimestamp的时候该trigger会被触发）。该window初始化时接受一个gap参数。多个元素依次到来，如果这些元素之间的时间间隔均不大于gap，他们会被合并至同一个window中。如果两个元素的时间间隔大于gap，则之前的window会被截断，后面的元素会进入一个新的window中。

val stream = ...

stream.keyBy(0)
      .window(EventTimeSessionWindows.withGap(Time.hours(2)))
      .trigger(CountTrigger.of(1))
      .evictor(TimeEvictor.of(Time.hours(2)))

Flink中每个window中的数据可以简单理解为以key value的形式在HeapListState中的CopyOnWriteStateTable中存储。key为window对象本身，value为该window中的数据。window的合并运算是将window进行并集运算，同时合并value集合的内容。这样一来合并之后的window所含数据很可能存在2小时之前的数据。在计算前排除他们我们需要evictor来帮忙。在这个例子中，我们用TimeEvictor将2小时之前的老数据清理出去。


public interface Evictor<T, W extends Window> extends Serializable {

    /**
     * Optionally evicts elements. Called before windowing function.
     *
     * @param elements The elements currently in the pane.
     * @param size The current number of elements in the pane.
     * @param window The {@link Window}
     * @param evictorContext The context for the Evictor
     */
    // 在计算操作执行前执行evict操作
    void evictBefore(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);

    /**
     * Optionally evicts elements. Called after windowing function.
     *
     * @param elements The elements currently in the pane.
     * @param size The current number of elements in the pane.
     * @param window The {@link Window}
     * @param evictorContext The context for the Evictor
     */
     
    // 在计算操作执行完后执行evict操作
    void evictAfter(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);


    /**
     * A context object that is given to {@link Evictor} methods.
     */
    interface EvictorContext {

        /**
         * Returns the current processing time.
         */
        long getCurrentProcessingTime();

        /**
         * Returns the metric group for this {@link Evictor}. This is the same metric
         * group that would be returned from {@link RuntimeContext#getMetricGroup()} in a user
         * function.
         *
         * <p>You must not call methods that create metric objects
         * (such as {@link MetricGroup#counter(int)} multiple times but instead call once
         * and store the metric object in a field.
         */
        MetricGroup getMetricGroup();

        /**
         * Returns the current watermark time.
         */
        long getCurrentWatermark();
    }
    
}

通常来说操作window的类为WindowOperator，它并没有使用evictor。一旦把window指定Evictor，该window会由EvictWindowOperator类来负责操作。

evictBefore和eviceAfter的调用时间点如下：

// EvictWindowOperator.java
private void emitWindowContents(W window, Iterable<StreamRecord<IN>> contents, ListState<StreamRecord<IN>> windowState) throws Exception {
        timestampedCollector.setAbsoluteTimestamp(window.maxTimestamp());

        // Work around type system restrictions...
        FluentIterable<TimestampedValue<IN>> recordsWithTimestamp = FluentIterable
            .from(contents)
            .transform(new Function<StreamRecord<IN>, TimestampedValue<IN>>() {
                @Override
                public TimestampedValue<IN> apply(StreamRecord<IN> input) {
                    return TimestampedValue.from(input);
                }
            });
        
        // 后续处理逻辑之前调用evictorBefore
        evictorContext.evictBefore(recordsWithTimestamp, Iterables.size(recordsWithTimestamp));

        FluentIterable<IN> projectedContents = recordsWithTimestamp
            .transform(new Function<TimestampedValue<IN>, IN>() {
                @Override
                public IN apply(TimestampedValue<IN> input) {
                    return input.getValue();
                }
            });

        processContext.window = triggerContext.window;
        
        // 后续处理逻辑
        userFunction.process(triggerContext.key, triggerContext.window, processContext, projectedContents, timestampedCollector);
        
        // 后续处理逻辑之后调用evictorAfter
        evictorContext.evictAfter(recordsWithTimestamp, Iterables.size(recordsWithTimestamp));

        //work around to fix FLINK-4369, remove the evicted elements from the windowState.
        //this is inefficient, but there is no other way to remove elements from ListState, which is an AppendingState.
        windowState.clear();
        for (TimestampedValue<IN> record : recordsWithTimestamp) {
            windowState.add(record.getStreamRecord());
        }
    }

Evictor在Flink中有如下3个实现

TimeEvictor
CountEvictor
DeltaEvictor

TimeEvictor

以时间为判断标准，决定元素是否会被移除。

private void evict(Iterable<TimestampedValue<Object>> elements, int size, EvictorContext ctx) {

    // 如果element没有timestamp，直接返回
    if (!hasTimestamp(elements)) {
        return;
    }

    // 获取elements中最大的时间戳（到来最晚的元素的时间）
    long currentTime = getMaxTimestamp(elements);
    // 截止时间为： 到来最晚的元素的时间 - 窗口大小（可以理解为保留最近的多久的元素）
    long evictCutoff = currentTime - windowSize;

    for (Iterator<TimestampedValue<Object>> iterator = elements.iterator(); iterator.hasNext(); ) {
        TimestampedValue<Object> record = iterator.next();
        
        // 清除所有时间戳小于截止时间的元素
        if (record.getTimestamp() <= evictCutoff) {
            iterator.remove();
        }
    }
}

CountEvictor

以元素计数为标准，决定元素是否会被移除。

private void evict(Iterable<TimestampedValue<Object>> elements, int size, EvictorContext ctx) {
    if (size <= maxCount) {
        // 小于最大数量，不做处理
        return;
    } else {
        int evictedCount = 0;
        for (Iterator<TimestampedValue<Object>> iterator = elements.iterator(); iterator.hasNext();){
            iterator.next();
            evictedCount++;
            if (evictedCount > size - maxCount) {
                break;
            } else {
                // 移除前size - maxCount个元素，只剩下最后maxCount个元素
                iterator.remove();
            }
        }
    }
}

DeltaEvictor

DeltaEvictor通过计算DeltaFunction的值（依次传入每个元素和最后一个元素），并将其与threshold进行对比，如果函数计算结果大于等于threshold，则该元素会被移除。
evict方法如下：

private void evict(Iterable<TimestampedValue<T>> elements, int size, EvictorContext ctx) {

    // 获取最后一个元素
    TimestampedValue<T> lastElement = Iterables.getLast(elements);
    
    for (Iterator<TimestampedValue<T>> iterator = elements.iterator(); iterator.hasNext();){
        TimestampedValue<T> element = iterator.next();
        // 依次计算每个元素和最后一个元素的delta值，同时和threshold的值进行比较
        // 若计算结果大于threshold值或者是相等，则该元素会被移除
        if (deltaFunction.getDelta(element.getValue(), lastElement.getValue()) >= this.threshold) {
            iterator.remove();
        }
    }
}

Flink 源码之Evictor