45、Flink 的自定义窗口剔除器 evictor 代码示例_flank evictor

1、代码示例

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.Evictor;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.runtime.operators.windowing.TimestampedValue;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class _13_WindowCustomEvictors {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> input = env.socketTextStream("localhost", 8888);

        // 测试时限制了分区数，生产中需要设置空闲数据源
        env.setParallelism(2);

        ArrayList<String> keyWords = new ArrayList<>();
        keyWords.add("a");

        // 事件时间需要设置水位线策略和时间戳
        SingleOutputStreamOperator<Tuple2<String, Long>> map = input.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String input) throws Exception {
                String[] fields = input.split(",");
                return new Tuple2<>(fields[0], Long.parseLong(fields[1]));
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = map.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                    @Override
                    public long extractTimestamp(Tuple2<String, Long> input, long l) {
                        return input.f1;
                    }
                }));

        watermarks.keyBy(e -> e.f0)
                .window(TumblingEventTimeWindows.of(Duration.ofSeconds(5)))
                .trigger(EventTimeTrigger.create())
                .evictor(new MyCustomEvictorDoBefore<>(false, keyWords))
                .apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<Tuple2<String, Long>> iterable, Collector<String> collector) throws Exception {
                        System.out.println("Window 的开始和结束时间=>" + timeWindow.getStart() + "-" + timeWindow.getEnd());

                        for (Tuple2<String, Long> tuple2 : iterable) {
                            collector.collect(tuple2.f0);
                        }
                    }
                })
                .print();

        env.execute();
    }
}

/**
 * doEvictAfter = false
 * <p>
 * a,1718157600000
 * b,1718157600000
 * c,1718157600000
 * <p>
 * a,1718157602000
 * b,1718157602000
 * c,1718157602000
 * <p>
 * a,1718157604000
 * b,1718157604000
 * c,1718157604000
 * <p>
 * a,1718157605001
 * b,1718157605001
 * <p>
 * Window 的开始和结束时间=>1718157600000-1718157605000
 * 2> a
 * 2> a
 * 2> a
 * Window 的开始和结束时间=>1718157600000-1718157605000
 * Window 的开始和结束时间=>1718157600000-1718157605000
 * <p>
 * c,1718157605001
 */
class MyCustomEvictorDoBefore<W extends Window> implements Evictor<Object, W> {
    private boolean doEvictAfter;

    private List<String> keyWords;

    public MyCustomEvictorDoBefore(boolean doEvictAfter, List<String> keyWords) {
        this.doEvictAfter = doEvictAfter;
        this.keyWords = keyWords;
    }

    @Override
    public void evictBefore(Iterable<TimestampedValue<Object>> elements, int i, W w, EvictorContext evictorContext) {
        if (!this.doEvictAfter) {
            this.evict(elements);
        }
    }

    @Override
    public void evictAfter(Iterable<TimestampedValue<Object>> elements, int i, W w, EvictorContext evictorContext) {
        if (this.doEvictAfter) {
            this.evict(elements);
        }
    }

    private void evict(Iterable<TimestampedValue<Object>> elements) {
        Iterator<TimestampedValue<Object>> iterator = elements.iterator();

        while (iterator.hasNext()) {
            TimestampedValue<Object> record = (TimestampedValue) iterator.next();
            for (String keyWord : keyWords) {
                if (!record.getValue().toString().contains(keyWord)) {
                    iterator.remove();
                }
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136