38、Flink 的 WindowAssigner 之 GlobalWindows 示例

1、注意
使用 GlobalWindows 需要自定义 Trigger，否则窗口中的数据不会被计算。
2、代码示例

1. import org.apache.flink.streaming.api.datastream.DataStreamSource;
2. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
3. import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
4. import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
5. import org.apache.flink.streaming.api.windowing.triggers.Trigger;
6. import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
7. import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
8. import org.apache.flink.util.Collector;
10. public class _05_WindowAssignerGlobal {
11.     public static void main(String[] args) throws Exception {
12.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
14.         DataStreamSource<String> input = env.socketTextStream("localhost", 8888);
16.         // 此窗口模式仅在指定了自定义的 trigger 时有用，否则计算不会发生，因为全局窗口没有天然的终点去触发其中积累的数据
17.         input
18.                 .keyBy(e -> e)
19.                 // 多并行 Task
20.                 .window(GlobalWindows.create())
21.                 .trigger(new Trigger<String, GlobalWindow>() {
22.                     @Override
23.                     public TriggerResult onElement(String s, long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
24.                         return null;
25.                     }
27.                     @Override
28.                     public TriggerResult onProcessingTime(long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
29.                         return null;
30.                     }
32.                     @Override
33.                     public TriggerResult onEventTime(long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
34.                         return null;
35.                     }
37.                     @Override
38.                     public void clear(GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {
40.                     }
41.                 })
42.                 .apply(new WindowFunction<String, String, String, GlobalWindow>() {
43.                     @Override
44.                     public void apply(String s, GlobalWindow globalWindow, Iterable<String> iterable, Collector<String> collector) throws Exception {
45.                         for (String res : iterable) {
46.                             collector.collect(res);
47.                         }
48.                     }
49.                 })
50.                 .print();
52.         env.execute();
53.     }
54. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。