Flink移除器Evictor

前言

在 Flink 窗口计算模型中，数据被 WindowAssigner 划分到对应的窗口后，再经过触发器 Trigger 判断窗口是否要 fire 计算，如果窗口要计算，会把数据丢给移除器 Evictor，Evictor 可以先移除部分元素再交给 ProcessFunction 处理，也可以等 ProcessFunction 处理完成后再移除数据。

认识Evictor

Flink中所有的移除器都是org.apache.flink.streaming.api.windowing.evictors.Evictor的子类

public interface Evictor<T, W extends Window> extends Serializable {void evictBefore(Iterable<TimestampedValue<T>> var1, int var2, W var3, EvictorContext var4);void evictAfter(Iterable<TimestampedValue<T>> var1, int var2, W var3, EvictorContext var4);public interface EvictorContext {long getCurrentProcessingTime();MetricGroup getMetricGroup();long getCurrentWatermark();}
}

Evictor定义了两个方法：

evictBefore ProcessFunction处理前调用，用于移除无须计算的元素
evictAfter ProcessFunction处理后调用

内置的Evictor

Flink内置了三个Evictor，当这些Evictor不满足业务场景时，也可以自定义Evictor。

1、TimeEvictor

给定一个时间窗口大小，仅保留该时间窗口范围内的元素，对于超过了窗口时间范围的元素，会一律移除。

public class TimeEvictor<W extends Window> implements Evictor<Object, W> {private static final long serialVersionUID = 1L;private final long windowSize;private final boolean doEvictAfter;public TimeEvictor(long windowSize) {this.windowSize = windowSize;this.doEvictAfter = false;}public TimeEvictor(long windowSize, boolean doEvictAfter) {this.windowSize = windowSize;this.doEvictAfter = doEvictAfter;}public void evictBefore(Iterable<TimestampedValue<Object>> elements, int size, W window, Evictor.EvictorContext ctx) {if (!this.doEvictAfter) {this.evict(elements, size, ctx);}}public void evictAfter(Iterable<TimestampedValue<Object>> elements, int size, W window, Evictor.EvictorContext ctx) {if (this.doEvictAfter) {this.evict(elements, size, ctx);}}private void evict(Iterable<TimestampedValue<Object>> elements, int size, Evictor.EvictorContext ctx) {if (this.hasTimestamp(elements)) {long currentTime = this.getMaxTimestamp(elements);long evictCutoff = currentTime - this.windowSize;Iterator<TimestampedValue<Object>> iterator = elements.iterator();while(iterator.hasNext()) {TimestampedValue<Object> record = (TimestampedValue)iterator.next();if (record.getTimestamp() <= evictCutoff) {iterator.remove();}}}}
}

2、DeltaEvictor

给定一个 double 阈值和一个差值计算函数 DeltaFunction，依次计算窗口内元素和最后一个元素的差值 delta，所有 delta 超过阈值的元素都会被移除。

public class DeltaEvictor<T, W extends Window> implements Evictor<T, W> {private static final long serialVersionUID = 1L;DeltaFunction<T> deltaFunction;private double threshold;private final boolean doEvictAfter;private DeltaEvictor(double threshold, DeltaFunction<T> deltaFunction) {this.deltaFunction = deltaFunction;this.threshold = threshold;this.doEvictAfter = false;}private DeltaEvictor(double threshold, DeltaFunction<T> deltaFunction, boolean doEvictAfter) {this.deltaFunction = deltaFunction;this.threshold = threshold;this.doEvictAfter = doEvictAfter;}public void evictBefore(Iterable<TimestampedValue<T>> elements, int size, W window, Evictor.EvictorContext ctx) {if (!this.doEvictAfter) {this.evict(elements, size, ctx);}}public void evictAfter(Iterable<TimestampedValue<T>> elements, int size, W window, Evictor.EvictorContext ctx) {if (this.doEvictAfter) {this.evict(elements, size, ctx);}}private void evict(Iterable<TimestampedValue<T>> elements, int size, Evictor.EvictorContext ctx) {TimestampedValue<T> lastElement = (TimestampedValue)Iterables.getLast(elements);Iterator<TimestampedValue<T>> iterator = elements.iterator();while(iterator.hasNext()) {TimestampedValue<T> element = (TimestampedValue)iterator.next();if (this.deltaFunction.getDelta(element.getValue(), lastElement.getValue()) >= this.threshold) {iterator.remove();}}}
}

3、CountEvictor

给定一个 maxCount，依次遍历窗口内的元素，数量超过 maxCount 后的所有元素全部移除。

public class CountEvictor<W extends Window> implements Evictor<Object, W> {private static final long serialVersionUID = 1L;private final long maxCount;private final boolean doEvictAfter;private CountEvictor(long count, boolean doEvictAfter) {this.maxCount = count;this.doEvictAfter = doEvictAfter;}private CountEvictor(long count) {this.maxCount = count;this.doEvictAfter = false;}public void evictBefore(Iterable<TimestampedValue<Object>> elements, int size, W window, Evictor.EvictorContext ctx) {if (!this.doEvictAfter) {this.evict(elements, size, ctx);}}public void evictAfter(Iterable<TimestampedValue<Object>> elements, int size, W window, Evictor.EvictorContext ctx) {if (this.doEvictAfter) {this.evict(elements, size, ctx);}}private void evict(Iterable<TimestampedValue<Object>> elements, int size, Evictor.EvictorContext ctx) {if ((long)size > this.maxCount) {int evictedCount = 0;Iterator<TimestampedValue<Object>> iterator = elements.iterator();while(iterator.hasNext()) {iterator.next();++evictedCount;if ((long)evictedCount > (long)size - this.maxCount) {break;}iterator.remove();}}}
}

自定义Evictor

实现org.apache.flink.streaming.api.windowing.evictors.Evictor接口即可自定义 Evictor，泛型要注意，第一个是元素类型，第二个是窗口类型。

举个例子，我们定义一个 Evictor，它在 ProcessFunction 计算前把窗口内所有的奇数全部移除掉，只保留偶数。

public static class MyEvictor implements Evictor<Integer, GlobalWindow> {@Overridepublic void evictBefore(Iterable<TimestampedValue<Integer>> iterable, int i, GlobalWindow globalWindow, EvictorContext evictorContext) {Iterator<TimestampedValue<Integer>> iterator = iterable.iterator();while (iterator.hasNext()) {TimestampedValue<Integer> value = iterator.next();if (value.getValue() % 2 != 0) {iterator.remove();}}}@Overridepublic void evictAfter(Iterable<TimestampedValue<Integer>> iterable, int i, GlobalWindow globalWindow, EvictorContext evictorContext) {}
}

编写一个简单的 Flink 作业验证一下我们自定义的 Evictor，数据源手动指定为数字1到6，统一分配到 GlobalWindow 窗口，Trigger 元素等于6个就出发计算，最终输出窗口内的元素

public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();environment.fromElements(1, 2, 3, 4, 5, 6).windowAll(GlobalWindows.create()).trigger(CountTrigger.of(6)).evictor(new MyEvictor()).process(new ProcessAllWindowFunction<Integer, Object, GlobalWindow>() {@Overridepublic void process(ProcessAllWindowFunction<Integer, Object, GlobalWindow>.Context context, Iterable<Integer> iterable, Collector<Object> collector) throws Exception {String elements = StringUtils.joinWith(",", iterable);System.err.println(elements);}});environment.execute();
}

运行Flink作业，控制台输出[2, 4, 6]，奇数在计算前就被移除掉了。

尾巴

Flink 的 Evictor 主要用于在窗口计算过程中，对窗口中的元素进行筛选和剔除。通过定义特定的 Evictor 策略，可以有效地控制窗口内数据的留存和输出。 Evictor 有助于提高数据处理的准确性和效率。它能够根据业务需求，如时间、数据特征等，去除不符合条件的数据，从而使窗口的计算结果更具针对性和可靠性。