window算子在flink中是非常重要的，要理解window算子首先要明白window的相关机制和原理。本文将从实战的角度讲解api的使用，详细的原理机制建议先阅读官方文档Windows。下面以Tumbling Windows为例讲解一些常见用法。下面基于ProcessingTime的样例都适用于EventTime。

基于ProcessingTime的基本用法

Window Join

定义

Transformation Description
DataStream,DataStream → DataStream Join two data streams on a given key and a common window.

说明

将两个window的数据进行join

样例

代码

  1. public class WindowJoinDemo {
  2.     public static void main(String[] args) throws Exception {
  3.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  4.         env.setParallelism(1);
  5.         DataStream<Tuple2<String, Integer>> orangeStream = env.addSource(new DataSource());
  6.         DataStream<Tuple2<String, Integer>> greenStream = env.addSource(new DataSource());
  7.         DataStream<Tuple3<String, Integer, Integer>> joinedStream = runWindowJoin(orangeStream, greenStream, 5);
  8.         joinedStream.print("join");
  9.         env.execute("Windowed Join Demo");
  10.     }
  12.     public static DataStream<Tuple3<String, Integer, Integer>> runWindowJoin(
  13.             DataStream<Tuple2<String, Integer>> grades,
  14.             DataStream<Tuple2<String, Integer>> salaries,
  15.             long windowSize) {
  17.         return grades.join(salaries)
  18.                 .where(new NameKeySelector())
  19.                 .equalTo(new NameKeySelector())
  20.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(windowSize)))
  21.                 .apply(new JoinFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple3<String, Integer, Integer>>() {
  23.                     @Override
  24.                     public Tuple3<String, Integer, Integer> join(
  25.                             Tuple2<String, Integer> first,
  26.                             Tuple2<String, Integer> second) {
  27.                         return new Tuple3<String, Integer, Integer>(first.f0, first.f1, second.f1);
  28.                     }
  29.                 });
  30.     }
  32.     private static class NameKeySelector implements KeySelector<Tuple2<String, Integer>, String> {
  33.         @Override
  34.         public String getKey(Tuple2<String, Integer> value) {
  35.             return value.f0;
  36.         }
  37.     }
  39.     private static class DataSource extends RichParallelSourceFunction<Tuple2<String, Integer>> {
  40.         private volatile boolean running = true;
  42.         @Override
  43.         public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
  44.             int bound = 50;
  45.             String[] keys = new String[]{"foo", "bar", "baz"};
  47.             final long numElements = RandomUtils.nextLong(10, 20);
  48.             int i = 0;
  49.             while (running && i < numElements) {
  50.                 Thread.sleep(RandomUtils.nextLong(1, 5) * 1000L);
  51.                 Tuple2 data = new Tuple2<>(keys[RandomUtils.nextInt(0, 3)], RandomUtils.nextInt(0, bound));
  52.                 ctx.collect(data);
  53.                 System.out.println(Thread.currentThread().getId() + "-sand data:" + data);
  54.                 i++;
  55.             }
  56.         }
  58.         @Override
  59.         public void cancel() {
  60.             running = false;
  61.         }
  62.     }
  63. }

输出结果

  1. 59-sand data:(bar,49)
  2. 58-sand data:(bar,44)
  3. 58-sand data:(foo,2)
  4. 59-sand data:(baz,34)
  5. 58-sand data:(baz,2)
  6. 59-sand data:(baz,29)
  7. join> (baz,34,2)
  8. join> (baz,29,2)

说明

两条流里面的数据类型都是Tuple2，随机生成一些数据，窗口大小设置为5秒，根据两个流数据中的key进行join

Window CoGroup

定义

Transformation Description
DataStream,DataStream → DataStream Cogroups two data streams on a given key and a common window.

说明

coGroup方法的是用与上面join方法类似，不同的地方在于coGroup方法可以拿到两个窗口的所有数据，所以可以实现更多的场景，例如join就相当于coGroup的特例，也就是两个窗口的数据集根据key取交集。

样例

代码

  1. public class WindowCoGroupDemo {
  2.     public static void main(String[] args) throws Exception {
  3.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  4.         env.setParallelism(1);
  6.         DataStream<Tuple2<String, Integer>> orangeStream = env.addSource(new DataSource());
  7.         DataStream<Tuple2<String, Integer>> greenStream = env.addSource(new DataSource());
  9.         DataStream<Tuple3<String, Integer, Integer>> joinedStream = runWindowCoGroup(orangeStream, greenStream, 10);
  11.         joinedStream.print();
  13.         env.execute("Windowed CoGroup Demo");
  14.     }
  16.     public static DataStream<Tuple3<String, Integer, Integer>> runWindowCoGroup(
  17.             DataStream<Tuple2<String, Integer>> orangeStream,
  18.             DataStream<Tuple2<String, Integer>> greenStream,
  19.             long windowSize) {
  21.         return orangeStream.coGroup(greenStream)
  22.                 .where(new NameKeySelector())
  23.                 .equalTo(new NameKeySelector())
  24.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(windowSize)))
  25.                 .apply(new Join());
  26.     }
  28.     private static class Join implements CoGroupFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple3<String, Integer, Integer>>{
  30.         @Override
  31.         public void coGroup(Iterable<Tuple2<String, Integer>> first, Iterable<Tuple2<String, Integer>> second, Collector<Tuple3<String, Integer, Integer>> out) throws Exception {
  32.             first.forEach(x -> {
  33.                 second.forEach(y -> {
  34.                     if (x.f0.equals(y.f0)){
  35.                         out.collect(new Tuple3<>(x.f0, x.f1, y.f1));
  36.                     }
  37.                 });
  38.             });
  39.         }
  40.     }
  42.     private static class NameKeySelector implements KeySelector<Tuple2<String, Integer>, String> {
  43.         @Override
  44.         public String getKey(Tuple2<String, Integer> value) {
  45.             return value.f0;
  46.         }
  47.     }
  49.     private static class DataSource extends RichParallelSourceFunction<Tuple2<String, Integer>> {
  50.         private volatile boolean running = true;
  52.         @Override
  53.         public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
  54.             int bound = 50;
  55.             String[] keys = new String[]{"foo", "bar", "baz"};
  57.             final long numElements = RandomUtils.nextLong(10, 20);
  58.             int i = 0;
  59.             while (running && i < numElements) {
  60.                 Thread.sleep(RandomUtils.nextLong(1, 5) * 1000L);
  61.                 Tuple2 data = new Tuple2<>(keys[RandomUtils.nextInt(0, 3)], RandomUtils.nextInt(0, bound));
  62.                 ctx.collect(data);
  63.                 System.out.println(Thread.currentThread().getId() + "-sand data:" + data);
  64.                 i++;
  65.             }
  66.         }
  68.         @Override
  69.         public void cancel() {
  70.             running = false;
  71.         }
  72.     }
  73. }

输出结果

  1. 59-sand data:(baz,4)
  2. 59-sand data:(foo,48)
  3. 57-sand data:(foo,34)
  4. 57-sand data:(foo,40)
  5. 59-sand data:(baz,24)
  6. 59-sand data:(bar,1)
  7. 57-sand data:(bar,22)
  8. 57-sand data:(bar,41)
  9. (bar,1,22)
  10. (bar,1,41)
  11. (foo,48,34)
  12. (foo,48,40)

说明

样例中对两个窗口的数据进行了类似join的计算

基于EventTime的基本用法

EventTime&Watermark

样例

代码

  1. public class EventTimeWindowDemo {
  2.     public static void main(String[] args) throws Exception {
  3.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  4.         env.setParallelism(1);
  5.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  7.         DataStream<Tuple3<String, Integer, Long>> orangeStream = env.addSource(new DataSource("orangeStream"))
  8.                 .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator());
  10.         orangeStream.keyBy(0)
  11.                 .window(TumblingEventTimeWindows.of(Time.seconds(30)))
  12.                 .apply(new WindowFunction<Tuple3<String, Integer, Long>, Object, Tuple, TimeWindow>() {
  13.                     @Override
  14.                     public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple3<String, Integer, Long>> input, Collector<Object> out) throws Exception {
  15.                         System.out.println(window.toString());
  16.                         out.collect(input);
  17.                     }
  18.                 }).name("EventTimeWindow").print("out");
  20.         env.execute("EventTime Demo");
  21.     }
  23.     private static class DataSource extends RichParallelSourceFunction<Tuple3<String, Integer, Long>> {
  25.         private volatile boolean running = true;
  26.         private volatile String name;
  28.         public DataSource(String name) {
  29.             this.name = name;
  30.         }
  32.         @Override
  33.         public void run(SourceContext<Tuple3<String, Integer, Long>> ctx) throws Exception {
  34.             Random random = new Random();
  35.             int bound = 100;
  36.             final long numElements = 10;
  37.             int i = 0;
  39.             while (running && i < numElements) {
  40.                 Thread.sleep(1500);
  41.                 Tuple3 data = new Tuple3<>("foo", random.nextInt(bound), getRandomInt(i*10, 60+i*10));
  42.                 ctx.collect(data);
  43.                 System.out.println(Thread.currentThread().getId() + "-" + this.name + "-sand data:" + data);
  44.                 i++;
  45.             }
  46.             Thread.sleep(5000);
  47.         }
  49.         @Override
  50.         public void cancel() {
  51.             running = false;
  52.         }
  54.         private long getRandomInt(int min, int max){
  55.             return 1573441860000L + 1000* RandomUtils.nextInt(min, max);
  56.         }
  57.     }
  59.     private static class BoundedOutOfOrdernessGenerator implements AssignerWithPeriodicWatermarks<Tuple3<String, Integer, Long>> {
  60.         private final long maxOutOfOrderness = 10000;
  61.         private long currentMaxTimestamp;
  63.         @Override
  64.         public long extractTimestamp(Tuple3<String, Integer, Long> row, long previousElementTimestamp) {
  65.             long timestamp = row.f2;
  66.             currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
  67.             System.out.println(Thread.currentThread().getId() + "-" + row + ",time="+stampToDate(row.f2.toString()) + ",watermark=" + stampToDate(String.valueOf(currentMaxTimestamp - maxOutOfOrderness)));
  68.             return timestamp;
  69.         }
  71.         @Override
  72.         public Watermark getCurrentWatermark() {
  73.             return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
  74.         }
  76.         private static String stampToDate(String s) {
  77.             String res;
  78.             SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  79.             long lt = new Long(s);
  80.             Date date = new Date(lt);
  81.             res = simpleDateFormat.format(date);
  82.             return res;
  83.         }
  84.     }
  85. }

输出结果

wm.m4v (16.45MB)

说明

  1. 时间窗口设置为30s
  2. watermark的计算公式为当前最大时间戳减去10s，也就是最大可容忍延迟10s的数据
  3. 默认采用的是EventTimeTrigger，下面是触发窗口计算的公式，其中window.maxTimestamp()返回的是窗口结束时间-1毫秒
  1. @Override
  2.     public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
  3.         if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
  4.             // if the watermark is already past the window fire immediately
  5.             return TriggerResult.FIRE;
  6.         } else {
  7.             ctx.registerEventTimeTimer(window.maxTimestamp());
  8.             return TriggerResult.CONTINUE;
  9.         }
  10.     }

Interval Join

定义

Transformation Description
KeyedStream,KeyedStream → DataStream Join two elements e1 and e2 of two keyed streams with a common key over a given time interval, so that e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound

说明

  1. 这个算子只支持EventTime
  2. 下图为这个算子的基本原理，watermark是对单个流中数据允许迟到多久进行控制的一个机制，而两个流进行join，就会涉及到两条流中的窗口是否同步的问题，这样就要考虑流和流之间的窗口存在延迟的情况，也就是between要指定的时间

image.png

  1. 上面介绍的Window Join算子(如下图)是基于两个相同时间窗口内所有数据的inner join；而Interval Join是以每个元素为视角，一条流中的元素去另一条流中查找key相同的元素，并且两个元素的时间戳要满足a.timestamp + lowerBound <= b.timestamp <= a.timestamp + upperBound

image.png

样例

代码

  1. public class IntervalJoinDemo {
  2.     public static void main(String[] args) throws Exception {
  3.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  4.         env.setParallelism(1);
  5.         //Time-bounded stream joins are only supported in event time
  6.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  8.         DataStream<Tuple3<String, Integer, Long>> orangeStream = env.addSource(new DataSource("orangeStream")).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator());
  9.         DataStream<Tuple3<String, Integer, Long>> greenStream = env.addSource(new DataSource("greenStream")).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator());
  11.         orangeStream
  12.                 .keyBy(0)
  13.                 .intervalJoin(greenStream.keyBy(0))
  14.                 .between(Time.seconds(-5), Time.seconds(5))
  15.                 .process(new ProcessJoinFunction<Tuple3<String, Integer, Long>, Tuple3<String, Integer, Long>, Object>() {
  16.                     @Override
  17.                     public void processElement(Tuple3<String, Integer, Long> left, Tuple3<String, Integer, Long> right, Context ctx, Collector<Object> out) throws Exception {
  18.                         out.collect(new Tuple5<>(left.f0, left.f1, left.f2, right.f1, right.f2));
  19.                     }
  20.                 }).name("intervalJoin").print("xxxxxx");
  22.         env.execute("Interval Join Demo");
  23.     }
  25.     private static class DataSource extends RichParallelSourceFunction<Tuple3<String, Integer, Long>> {
  27.         private volatile boolean running = true;
  28.         private volatile String name;
  30.         public DataSource(String name) {
  31.             this.name = name;
  32.         }
  34.         @Override
  35.         public void run(SourceContext<Tuple3<String, Integer, Long>> ctx) throws Exception {
  36.             Random random = new Random();
  37.             int bound = 100;
  38.             Tuple3[] data = new Tuple3[]{
  39.                     new Tuple3<>("foo", random.nextInt(bound), getRandomInt(50, 70)), new Tuple3<>("foo", random.nextInt(bound),  getRandomInt(40, 60))};
  40.             final long numElements = data.length;
  41.             int i = 0;
  42.             while (running && i < numElements) {
  43.                 Thread.sleep(RandomUtils.nextLong(1, 5) * 1000L);
  44.                 ctx.collect(data[i]);
  45.                 System.out.println(Thread.currentThread().getId() + "-" + this.name + "-sand data:" + data[i]);
  46.                 i++;
  47.             }
  48.         }
  50.         @Override
  51.         public void cancel() {
  52.             running = false;
  53.         }
  55.         private long getRandomInt(int min, int max){
  56.             return 1573441870000L + 1000*(new Random().nextInt(max-min+1)+min);
  57.         }
  58.     }
  60.     private static class BoundedOutOfOrdernessGenerator implements AssignerWithPeriodicWatermarks<Tuple3<String, Integer, Long>> {
  62.         private final long maxOutOfOrderness = 10000;
  64.         private long currentMaxTimestamp;
  66.         @Override
  67.         public long extractTimestamp(Tuple3<String, Integer, Long> row, long previousElementTimestamp) {
  68.             System.out.println(Thread.currentThread().getId() + "-" + row + ",time="+stampToDate(row.f2.toString()));
  69.             long timestamp = row.f2;
  70.             currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
  71.             System.out.println(Thread.currentThread().getId() + "-watermark:" + stampToDate(String.valueOf(currentMaxTimestamp - maxOutOfOrderness)));
  72.             return timestamp;
  73.         }
  75.         @Override
  76.         public Watermark getCurrentWatermark() {
  77.             return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
  78.         }
  80.         private static String stampToDate(String s) {
  81.             String res;
  82.             SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  83.             long lt = new Long(s);
  84.             Date date = new Date(lt);
  85.             res = simpleDateFormat.format(date);
  86.             return res;
  87.         }
  88.     }
  89. }