Flink Window窗口开始结束时间分析-源码探索

Posted 2022-11-26 总要冲动一次

首先 flink 的窗口分配是发生在 StreamTask 初始化的过程中。
核心方法是 TumblingProcessingTimeWindows.assignWindows(…)

// TODO : 在初始化StreamTask的时候需要分配好窗口
    @Override
    public Collection<TimeWindow> assignWindows(
            Object element, long timestamp, WindowAssignerContext context) 
        final long now = context.getCurrentProcessingTime();
        // TODO : 默认情况下 staggerOffset = 0
        if (staggerOffset == null) 
            staggerOffset =
                    windowStagger.getStaggerOffset(context.getCurrentProcessingTime(), size);
        
        // TODO : 获取窗口起始时间
        long start =
                TimeWindow.getWindowStartWithOffset(
                        now, (globalOffset + staggerOffset) % size, size);
        return Collections.singletonList(new TimeWindow(start, start + size));
    

可以通过这个方法往上点，会发现他是在StreamTask初始化的时候触发的。
这个方法调用了一个很重要的方法来计算窗口开始时间：TimeWindow.getWindowStartWithOffset(…)

/**
     * Method to get the window start for a timestamp.
     *
     * @param timestamp epoch millisecond to get the window start.   当前时间
     * @param offset The offset which window start would be shifted by.  窗口偏移
     * @param windowSize The size of the generated windows.  窗口大小
     *
     * @return window start
     */
    // TODO : 默认 offset = 0
    /**
     * 1、如果 windowSize 是整秒，根据计算公式可知 当前时间-(当前时间+窗口大小)%windowSize 得到的一定是一个整秒时间，以此类推可知整分钟、整小时都是这样
     * 2、举例：整秒，如果windowSize是60的因数，那么窗口的排布是按照每分钟的00秒开始，其余的就不好说了
     * 3、根据公式可知，一旦windowSize确定了，从以前到现在再到未来，所有的窗口都已经分布好了！ 当然如果配置了 offset 就另说了
     */
    public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) 
        return timestamp - (timestamp - offset + windowSize) % windowSize;
        // TODO : 如果 offset = 0，当前时间 - 当前时间除去windowSize的余数
        // TODO : 如果 offset != 0且为正数, 由于 offset 不会大于 windowSize，所以会导致余数变小了，最终得到的窗口 startTime 变大了
    

注释里有我的简单总结，供参考。