Flink窗口大小怎么确定的,开始时间是时候?

Source

面试官:Flink窗口大小怎么确定的,开始时间是什么时候,结束是什么时候?

这篇文章会很短,但是我认为很有意义,今天晚上微醺了,纯手敲的。

我还记得的在我刚学习flink的时候,B站的老师说过,Flink窗口的开始时间和结束时间和你想的不一样。那个时候我好像记得老师说过,flink的窗口大小会根据你的时间单位来进行修正。

然后在现如今,很多人还是不是很了解窗口机制,以及watermark。更别提什么窗口什么时候,什么时候结束。所以呢,今天从源码角度给大家普及一下窗口什么时候开始,什么时候结束。

我们可以来编写一个简单的代码,来看一下效果,我习惯用java来写flink,所以也就使用java了。

	@Override
	public Collection<TimeWindow> assignWindows(Object element, long timestamp, WindowAssignerContext context) {
    
      
		if (timestamp > Long.MIN_VALUE) {
    
      
			List<TimeWindow> windows = new ArrayList<>((int) (size / slide));
      //获取窗口开始时间
			long lastStart = TimeWindow.getWindowStartWithOffset(timestamp, offset, slide);
			for (long start = lastStart;
				start > timestamp - size;
				start -= slide) {
    
      
				windows.add(new TimeWindow(start, start + size));
			}
			return windows;
		} else {
    
      
			throw new RuntimeException("Record has Long.MIN_VALUE timestamp (= no timestamp marker). " +
					"Is the time characteristic set to 'ProcessingTime', or did you forget to call " +
					"'DataStream.assignTimestampsAndWatermarks(...)'?");
		}
	}
	public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
    
      
    //窗口开始计算的时间
		return timestamp - (timestamp - offset + windowSize) % windowSize;
	}

我们可以看出来窗口开始时间, 是取模过后的时间,我们来简单的分析一番。

假如我们第一条数据的时间戳是1000,offset暂时不需要管,因为他是时间的偏移量例如,东八区什么的。我们假如窗口大小是5s,

那么接下来的公式计算也就是 1000 - (1000 - 0 + 5000)%5000,那么我们可以计算出来的结果就是0,也就是说,开始的时间窗口是0.更大的时间窗口大小,各位大佬可以下面自己算一下。

也就是说开始时间是 0,结束的时间窗口也就是4999,因为到5000的时候就触发计算了。那么我们接下来就进行验证一番和我们分析的是否一致。

接下来我们写一个简单的wordcount,因为在多并行度下,不是很好分析,我们设置为单并行读。如果有对watermark还不是很理解的大佬,可以看我的这篇文章,https://blog.csdn.net/weixin_43704599/article/details/117411252

/**
 * @author :qingzhi.wu
 * @date :2022/1/18 8:36 下午
 */
public class WindowSizeTest {
    
      
    public static void main(String[] args) throws Exception {
    
      
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<String> source1 = source.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.of(0, TimeUnit.MILLISECONDS)) {
    
      
            @Override
            public long extractTimestamp(String s) {
    
      

                return Long.parseLong(s.split(",")[0]);
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> map = source1.map(new MapFunction<String, Tuple2<String, Integer>>() {
    
      
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
    
      
                return Tuple2.of(s.split(",")[1],1);
            }
        });

        WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> window = map.keyBy(0).window(TumblingEventTimeWindows.of(Time.of(5000, TimeUnit.MILLISECONDS)));
        window.sum(1).print();
        env.execute();
    }
}

接下来我们,就看看我们分析的是否正确

在这里插入图片描述

很明显是正确的。那么一天的窗口大小你会计算吗?