Flink-时间窗口源码分析

谁借莪１个温暖的怀抱￠ 2023-05-21 11:48 29阅读 0赞

# Flink-时间窗口源码分析 #

# Flink系列文章 #

*  更多Flink系列文章请点击[Flink系列文章][Flink]
 *  更多大数据文章请点击[大数据好文推荐][Link 1]

# 1 时间窗口基本概念 #

见[Flink学习1-基础概念-时间窗口][Flink_1-_-]

窗口生命周期：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center]

# 2 滚动窗口 #

## 2.1 基于Blink的滚动EventTime时间窗口源码分析 ##

### 2.1.1 概述 ###

我们现在开始用Flink 1.10，该版本可使用很多Blink特性，我们重点在研究Flink SQL实现流平台，而尽量不写代码。而基于EventTime的滚动时间窗口又是很常用的，我们在使用时遇到了水印时间自动加8的问题，肯定是和处于+8时区有关。现在想具体探究下水印、窗口之间关系。

### 2.1.2 创建WindowAssigner和Trigger源码分析 ###

下面以blink中使用`GROUP BY TUMBLE(ts_field, INTERVAL '5' SECOND)`定义的5秒大小的滚动时间窗口作为例子。

#### 2.1.2.1 SQL转换 ####

开始运行后，将DDL中窗口相关的语句翻译

#### 2.1.2.2 创建WindowAssigner和Trigger ####

org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecGroupWindowAggregateBase\#createWindowOperator  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70]  
这个方法会根据之前已经转好的window对象，利用scala的模式匹配，真正创建分配数据所属窗口的WindowAssigner和窗口触发Trigger。

这里我们创建的是基于eventTime的、基于时间来确定窗口大小的滚动时间窗口，所以分配器是TumblingWindowAssigner。

而Trigger是`EventTimeTriggers.AfterEndOfWindow`。

#### 2.1.2.3 TumblingWindowAssigner与Pane ####

*  作用  
    窗口分配器，继承自WindowAssigner，用来给每个元素分配0个或多个（TumblingWindowAssigner最多有一个）Window。
 *  Pane  
    在Window算子中，当某个列key可用时就用key和所属的Window来分组。而拥有同样的key和Window的所有元素就被称为一个Pane。
    
    由触发器决定何时让某个Pane触发所拥有窗口来生产输出元素。
    
    一个被WindowAssigner分配到多个Window的元素可以在同个Pane中，这些Pane每个都拥有自己独立的一份Trigger实例。
    
    目前只有`SlidingWindowAssigner`实现了`PanedWindowAssigner`
 *  创建  
    创建Window时，首先需要创建一个为每个元素分配窗口的分配器，这里具体就是`TumblingWindowAssigner`。
    
        public static TumblingWindowAssigner of(Duration size) { 
        	return new TumblingWindowAssigner(size.toMillis(), 0, true);
        }
    
    构建`TumblingWindowAssigner`时，传入了窗口大小、窗口开始时偏移量、是否eventTime计算。
    
        protected TumblingWindowAssigner(long size, long offset, boolean isEventTime) { 
        	if (size <= 0) { 
        		throw new IllegalArgumentException
        			("TumblingWindowAssigner parameters must satisfy size > 0");
        	}
        	this.size = size;
        	this.offset = offset;
        	this.isEventTime = isEventTime;
        }
 *  偏移量offset  
    这个偏移量的意思是，从0开始计算，Window从`N * size + offset`开始。这里我们传入的offset是0，就是说第一个窗口从0开始，下一个窗口从0 + 1*szie = 0 + 1*5 = 5秒开始。
    
    如果说想用中国时间来开一个一整天的窗口即每个窗口从中国时间0点开始计算，就应该创入`Time.days(1),Time.hours(-8)`，因为窗口默认按UTC时间计算，即窗口从UTC时间0点也就是中国时间8点开始计算，所以必须减8才是中国时间0点。
 *  assignWindows  
    本类还有一个继承自`org.apache.flink.table.runtime.operators.window.assigners.WindowAssigner`的方法`assignWindows`:  
    本方法传入流入的一条记录和他的时间戳（如果采用EventTime就是该条记录的时间戳，否则就是系统当前时间）。
    
    然后得到该元素应该放入的Window的集合（TumbleWindow因为窗口间无重叠所以只返回一个Window，而SlidingWindowAssigner会返回元素所在的多个Window）
    
        @Override
        public Collection<TimeWindow> assignWindows(BaseRow element, long timestamp) { 
        	long start = TimeWindow.getWindowStartWithOffset(timestamp, offset, size);
        	return Collections.singletonList(new TimeWindow(start, start + size));
        }
 *  TimeWindow.getWindowStartWithOffset  
    用来计算该元素对应窗口的开始时间：
    
        public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) { 
        	return timestamp - (timestamp - offset + windowSize) % windowSize;
        }
    
    比如窗口大小为5秒，当前元素的时间戳`1576080003000`，offset为0，则结果为`1576080000000`，即为该元素所属窗口的开始时间。

#### 2.1.2.4 EventTimeTriggers ####

该类用来生成不同的继承自Trigger的一些基于EventTime的窗口触发器。

这里会调用`afterEndOfWindow`创建一个`AfterEndOfWindow`触发器。

public static <W extends Window> AfterEndOfWindow<W> afterEndOfWindow() { 
    	return new AfterEndOfWindow<>();
    }

AfterEndOfWindow为`EventTimeTriggers`内部类，他的定义及一些重要方法如下:

// 继承自Trigger，泛型继承自Window
    // Trigger主要用来决定何时来评估一个窗口的Pane以发送那部分结果
    // 需要注意的是Trigger不允许内部维护状态，因为他们可被不同的key重建或重用
    // 应该使用TriggerContext来持久化状态
    public static final class AfterEndOfWindow<W extends Window> extends Trigger<W> { 
    	private static final long serialVersionUID = -6379468077823588591L;
    
    	/** * Creates a new {@code Trigger} like the this, except that it fires repeatedly whenever * the given {@code Trigger} fires before the watermark has passed the end of the window. */
    	public AfterEndOfWindowNoLate<W> withEarlyFirings(Trigger<W> earlyFirings) { 
    		checkNotNull(earlyFirings);
    		return new AfterEndOfWindowNoLate<>(earlyFirings);
    	}
    
    	/** * Creates a new {@code Trigger} like the this, except that it fires repeatedly whenever * the given {@code Trigger} fires after the watermark has passed the end of the window. */
    	public Trigger<W> withLateFirings(Trigger<W> lateFirings) { 
    		checkNotNull(lateFirings);
    		if (lateFirings instanceof ElementTriggers.EveryElement) { 
    			// every-element late firing can be ignored
    			return this;
    		} else { 
    			return new AfterEndOfWindowEarlyAndLate<>(null, lateFirings);
    		}
    	}
    
    	// Trigger内部接口，用来注册时间回调函数以及处理状态
    	private TriggerContext ctx;
    
    	// Tigger的初始化方法，一般被Trigger用来创建状态，但这里仅仅将ctx保存到我们的Trigger
    	@Override
    	public void open(TriggerContext ctx) throws Exception { 
    		this.ctx = ctx;
    	}
    
    	// 每个元素被添加到Pane时都会调用
    	// 结果就是是否触发该Pane输出结果，触发Window计算
    	@Override
    	public boolean onElement(Object element, long timestamp, W window) throws Exception { 
    		if (window.maxTimestamp() <= ctx.getCurrentWatermark()) { 
    			// 如果窗口最大时间戳小于等于当前触发器上下文记录的水位，触发窗口计算
    			// if the watermark is already past the window fire immediately
    			return true;
    		} else { 
    			// 否则向trigger ctx注册一个基于EventTime的回调
    			// 这里注册的是窗口的最大时间戳
    			// 一旦advanceWatermark提升水位时，超过注册的时间
    			// 就会回调Trigger#onEventTime方法
    			ctx.registerEventTimeTimer(window.maxTimestamp());
    			return false;
    		}
    	}
    	
    	// 基于ProcessingTime的定时回调，这里我们基于EventTime所以返回false
    	@Override
    	public boolean onProcessingTime(long time, W window) throws Exception { 
    		return false;
    	}
    
    	// 基于EventTime的定时回调，返回true就代表窗口应该被触发计算
    	// 注意：如果窗口没有任何元素则不会触发
    	// time为触发时间，window为目标触发窗口
    	@Override
    	// AfterEndOfWindow#onEventTime
    	public boolean onEventTime(long time, W window) throws Exception { 
    		// 触发时间等于窗口最大时间戳就触发
    		return time == window.maxTimestamp();
    	}
    
    	// 当一个窗口purge清除的时候调用
    	// 一般用来清理Trigger持续持有的Window的状态（用TriggerContext#getPartitionedState获取的）
    	// 以及向TriggerContext注册的时间回调函数
    	@Override
    	public void clear(W window) throws Exception { 
    		ctx.deleteEventTimeTimer(window.maxTimestamp());
    	}
    
    	// 是否支持trigger状态合并
    	@Override
    	public boolean canMerge() { 
    		return true;
    	}
    
    	// 当若干Window已经被WindowAssigner合并为一个Window时触发
    	@Override
    	public void onMerge(W window, OnMergeContext mergeContext) throws Exception { 
    		// 这里也是向trigger ctx注册一个基于EventTime的回调
    		// 这里注册的是窗口的最大时间戳
    		ctx.registerEventTimeTimer(window.maxTimestamp());
    	}
    
    	@Override
    	public String toString() { 
    		return TO_STRING;
    	}
    }

#### 2.1.2.5 TimeWindow ####

先看看TimeWindow的父类`org.apache.flink.table.runtime.operators.window.Window`:

Window就是一组元素进入一个有限的Bucket。

Window拥有一个最大时间戳，元素应该在此之前进入窗口，否则就迟到了。

public abstract class Window implements Comparable<Window> { 
    	// 获取窗口最大时间戳
    	public abstract long maxTimestamp();
    	
    	// 逻辑上相等的Window就应该相等
    	public abstract int hashCode();
    
    	// 逻辑上相等的Window就应该相等
    	public abstract boolean equals(Object obj);
    
    }

我们用的是基于事件时间触发的窗口，所以这里简单说下TimeWindow。

public class TimeWindow extends Window { 
    	
    	// 窗口起点（包含）
    	private final long start;
    	// 窗口终点（这一点被排除）
    	private final long end;
    	
    	public TimeWindow(long start, long end) { 
    		this.start = start;
    		this.end = end;
    	}
    
    	public long getStart() { 
    		return start;
    	}
    
    	public long getEnd() { 
    		return end;
    	}
    	
    	// 获取属于该Window的最后一个毫秒时间戳
    	@Override
    	public long maxTimestamp() { 
    		return end - 1;
    	}
    	
    	// 判断窗口是否逻辑相同
    	@Override
    	public boolean equals(Object o) { 
    		if (this == o) { 
    			return true;
    		}
    		if (o == null || getClass() != o.getClass()) { 
    			return false;
    		}
    
    		TimeWindow window = (TimeWindow) o;
    
    		return end == window.end && start == window.start;
    	}
    
    	@Override
    	public int hashCode() { 
    		// inspired from Apache BEAM
    		// The end values are themselves likely to be arithmetic sequence, which
    		// is a poor distribution to use for a hashtable, so we
    		// add a highly non-linear transformation.
    		return (int) (start + modInverse((int) (end << 1) + 1));
    	}
    
    	/** * Compute the inverse of (odd) x mod 2^32. */
    	private int modInverse(int x) { 
    		// Cube gives inverse mod 2^4, as x^4 == 1 (mod 2^4) for all odd x.
    		int inverse = x * x * x;
    		// Newton iteration doubles correct bits at each step.
    		inverse *= 2 - x * inverse;
    		inverse *= 2 - x * inverse;
    		inverse *= 2 - x * inverse;
    		return inverse;
    	}
    
    	@Override
    	public String toString() { 
    		return "TimeWindow{" +
    			"start=" + start +
    			", end=" + end +
    			'}';
    	}
    
    	// 判断本Window是否和其他Window存在交集
    	public boolean intersects(TimeWindow other) { 
    		return this.start <= other.end && this.end >= other.start;
    	}
    
    	// 创建包含本窗口和指定窗口的最小并集窗口
    	public TimeWindow cover(TimeWindow other) { 
    		return new TimeWindow(Math.min(start, other.start), Math.max(end, other.end));
    	}
    	
    	// 比较当前窗口和指定窗口哪个更早
    	@Override
    	public int compareTo(Window o) { 
    		TimeWindow that = (TimeWindow) o;
    		if (this.start == that.start) { 
    			return Long.compare(this.end, that.end);
    		} else { 
    			return Long.compare(this.start, that.start);
    		}
    	}
    
    	/** * The serializer used to write the TimeWindow type. * 这里内容省略 */
    	public static class Serializer extends TypeSerializerSingleton<TimeWindow> { 
    		private static final long serialVersionUID = 1L;
    	}
    
    	// ------------------------------------------------------------------------
    	// Utilities
    	// ------------------------------------------------------------------------
    
    	/** * 获取指定时间戳对应窗口的开始时间 * 比如窗口大小为5秒，当前元素的时间戳`1576080003000`，offset为0， * 则结果为`1576080000000`，即为该元素所属窗口的开始时间。 * * @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 */ 
    	public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) { 
    		return timestamp - (timestamp - offset + windowSize) % windowSize;
    	}
    	
    	// 创建TimeWindow的方法
    	public static TimeWindow of(long start, long end) { 
    		return new TimeWindow(start, end);
    	}
    }

#### 2.1.2.6 创建WindowAssigner和Trigger源码小结 ####

1.  根据SQL中的窗口相关语句创建WindowAssigner（TumblingWindowAssigner）和Trigger（EventTimeTriggers.AfterEndOfWindow）
2.  TumblingWindowAssigner用来分配每条流入元素所属TimeWindow窗口  
    注意这里每次都是用`assignWindows`方法新建一个窗口，所以TimeWindow里面有方法判断窗口是否逻辑相等
3.  EventTimeTriggers.AfterEndOfWindow触发器有两件事：
    
     *  onElement方法-根据流入元素判断是应该触发窗口计算还是用Window最大时间戳来注册基于EventTime的回调函数，到时候会触发onEventTime方法
     *  onEventTime方法-根据EventTime定时触发窗口计算  
        用来根据给定时间戳判断是否应该触发窗口计算，触发条件就是该时间戳是否等于窗口的End-1。
4.  水位生成机制见2.1.3，更多详细内容可参考[Flink-水位][Flink-]
5.  触发器向`TriggerContext`注册的EventTimeTimer的触发以及窗口的触发见 2.1.4。

### 2.1.3 WatermarkAssignerOperator源码分析 ###

**注意：本节介绍的是在首个EventTime相关算子之前定义的、Blink FlinkSql 使用的水位生成器。**

WatermarkAssignerOperator属于周期性水位，水位和窗口密切相关，所以这里简单分析下源码。

*  关键源码1：`StreamExecutionEnvironment#setStreamTimeCharacteristic(TimeCharacteristic.EventTime)`  
    这和前面提到的相同，会自动将`autoWatermarkInterval`设为200ms：
 *  关键源码2：`WatermarkAssignerOperatorFactory#createStreamOperator`  
    这里会用现编译、加载、生成一个水印分配器WatermarkGenerator的实现对象，然后用来生成`WatermarkAssignerOperator`
    
        public StreamOperator createStreamOperator(StreamTask containingTask, StreamConfig config, Output output) { 
        	WatermarkGenerator watermarkGenerator = generatedWatermarkGenerator.newInstance(containingTask.getUserCodeClassLoader());
        	WatermarkAssignerOperator operator = new WatermarkAssignerOperator(rowtimeFieldIndex, watermarkGenerator, idleTimeout);
        	operator.setup(containingTask, config, output);
        	return operator;
        }
 *  关键源码3：WatermarkGenerator的实现对象`WatermarkGenerator$2`的代码  
    这个是动态生成的，会根据我们定义的DDL中EventTime字段顺序、`watermark_strategy_expression`不同而略有不同。主要功能是计算event\_time字段的水位时间。
    
    以下是设置`WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND`时生成的代码：
    
        public final class WatermarkGenerator$2
                extends org.apache.flink.table.runtime.generated.WatermarkGenerator { 
            public WatermarkGenerator$2(Object[] references) throws Exception { }
        
            @Override
            public void open(org.apache.flink.configuration.Configuration parameters) throws Exception { }
        
            @Override
            public Long currentWatermark(org.apache.flink.table.dataformat.BaseRow row) throws Exception { 
                org.apache.flink.table.dataformat.SqlTimestamp field$3;
                // 用来判断event_time字段是否为空
                boolean isNull$3;
                boolean isNull$4;
                org.apache.flink.table.dataformat.SqlTimestamp result$5;
                // 因为我们测试例子中event_time字段为第3个字段（字段序号从0开始），所以判断第三个是否空
                isNull$3 = row.isNullAt(3);
                field$3 = null;
                if (!isNull$3) { 
                	// event_time字段不为空，就获取该字段的Timestamp(3)类型数据
                    field$3 = row.getTimestamp(3, 3);
                }
                
                isNull$4 = isNull$3 || false;
                result$5 = null;
                if (!isNull$4) { 
                // event_time字段不为空，就将刚才获取到的Timestamp(3)类型数据转为毫秒数后减1，然后构成SqlTimestamp
                // 精确到纳秒
                    result$5 = org.apache.flink.table.dataformat.SqlTimestamp.fromEpochMillis(field$3.getMillisecond() - ((long) 1L), field$3.getNanoOfMillisecond());
                }
        
                if (isNull$4) { 
                    return null;
                } else { 
                	// 返回转换后的SqlTimestamp的毫秒数，即计算得到的水位时间
                    return result$5.getMillisecond();
                }
            }
        
            @Override
            public void close() throws Exception { }
        }
    
    当设置`WATERMARK FOR ts AS ts`时生成代码略有不同，因为水位直接就用该字段的时间戳转换即可：
    
        public Long currentWatermark(org.apache.flink.table.dataformat.BaseRow row) throws Exception { 
            org.apache.flink.table.dataformat.SqlTimestamp field$3;
            boolean isNull$3;
            isNull$3 = row.isNullAt(3);
            field$3 = null;
            if (!isNull$3) { 
                field$3 = row.getTimestamp(3, 3);
            }
        
            if (isNull$3) { 
                return null;
            } else { 
                return field$3.getMillisecond();
            }
        }
 *  关键源码4：`WatermarkAssignerOperator`构造方法

public WatermarkAssignerOperator(int rowtimeFieldIndex, WatermarkGenerator watermarkGenerator, long idleTimeout) { 
    	this.rowtimeFieldIndex = rowtimeFieldIndex;
    	this.watermarkGenerator = watermarkGenerator;
    	// 0
    	this.idleTimeout = idleTimeout;
    	// 该算子可以被连接到其他算子尾部和头部
    	this.chainingStrategy = ChainingStrategy.ALWAYS;
    }

*  关键源码5：`WatermarkAssignerOperator#open`  
    会在初始化算子时调用该方法，这里初始化了当前水位`currentWatermark`、发送waterMark的时间间隔`watermarkInterval`、向`ProcessingTimeService`注册定时发送水位的任务：
    
        public void open() throws Exception { 
        	super.open();
        
        	// watermark and timestamp should start from 0
        	this.currentWatermark = 0;
        	this.watermarkInterval = getExecutionConfig().getAutoWatermarkInterval();
        	this.lastRecordTime = getProcessingTimeService().getCurrentProcessingTime();
        	this.streamStatusMaintainer = getContainingTask().getStreamStatusMaintainer();
        
        	if (watermarkInterval > 0) { 
        	// 水位检测间隔默认为0，
        	// 但只要我们设置大于0或(TimeCharacteristic.EventTime)则会注册`ProcessingTimeService`
        		long now = getProcessingTimeService().getCurrentProcessingTime();
        		getProcessingTimeService().registerTimer(now + watermarkInterval, this);
        	}
        	// 将StreamingRuntimeContext放入watermarkGenerator
        	FunctionUtils.setFunctionRuntimeContext(watermarkGenerator, getRuntimeContext());
        	FunctionUtils.openFunction(watermarkGenerator, new Configuration());
        }
 *  关键源码6：`WatermarkAssignerOperator#advanceWatermark`  
    判断当前水位是否大于之前的水位，大于就更新lastWatermark并发送当前水位到下游
    
        private void advanceWatermark() { 
        	if (currentWatermark > lastWatermark) { 
        		lastWatermark = currentWatermark;
        		// emit watermark
        		output.emitWatermark(new Watermark(currentWatermark));
        	}
        }
 *  关键源码7：`WatermarkAssignerOperator#processElement`  
    每条记录都会触发调用此方法：
    
    1.  尝试更新`currentWatermark`（如果本记录时间戳更大），
    2.  将记录发送到下游。
    3.  如果当前记录水位减去当前系统最后更新水位的差大于水位监测间隔时间，则还会尝试提升水位。  
        这里为了代码复用调用`advanceWatermark`方法内又判断了一次`if (currentWatermark > lastWatermark)`，但其实是重复判断？！因为我理解`watermarkInterval`始终应该是非负的，也就是说已经判断`currentWatermark > lastWatermark`：
    
        public void processElement(StreamRecord<BaseRow> element) throws Exception { 
        	if (idleTimeout > 0) { 
        		// mark the channel active
        		streamStatusMaintainer.toggleStreamStatus(StreamStatus.ACTIVE);
        		lastRecordTime = getProcessingTimeService().getCurrentProcessingTime();
        	}
        	// 该条记录
        	BaseRow row = element.getValue();
        	if (row.isNullAt(rowtimeFieldIndex)) { 
        		throw new RuntimeException("RowTime field should not be null," +
        				" please convert it to a non-null long value.");
        	}
        	// 根据该条记录的event_time字段计算当前水位
        	Long watermark = watermarkGenerator.currentWatermark(row);
        	if (watermark != null) { 
        	// 当前记录水位不为空且大于当前高水位就更新高水位
        		currentWatermark = Math.max(currentWatermark, watermark);
        	}
        	// 发送该条记录到算子链下游
        	output.collect(element);
        
        	// 如果当前计算出的水位减去之前最后更新的水位大于水位监测间隔时间就尝试提升水位
        	// 在这里也做提升水位的尝试是为了避免在系统负载高（如CPU负载太高）时，不能及时周期性调用onProcessingTime
        	if (currentWatermark - lastWatermark > watermarkInterval) { 
        		advanceWatermark();
        	}
        }
 *  关键源码8：`WatermarkAssignerOperator#onProcessingTime`  
    这是前面提到的向`ProcessingTimeService`注册的定时发送水位的任务。
    
    1.  本方法执行是周期性的，调用周期就是`autoWatermarkInterval`。
    2.  这里会先尝试提升水位。
    3.  如果设置了`idleTimeout`且检测到系统时间减去处理的最后一条记录时间已经超过阈值idleTimeout，则会将本channel标记空闲，来忽略本channel发送的水印。
    4.  最后再注册下个周期触发水位发送逻辑的定时任务。这就印证了水位是不断上升不能下降的概念：
    
        public void onProcessingTime(long timestamp) throws Exception { 
        	advanceWatermark();
        
        	if (idleTimeout > 0) { 
        		final long currentTime = getProcessingTimeService().getCurrentProcessingTime();
        		if (currentTime - lastRecordTime > idleTimeout) { 
        			// mark the channel as idle to ignore watermarks from this channel
        			streamStatusMaintainer.toggleStreamStatus(StreamStatus.IDLE);
        		}
        	}
        
        	// register next timer
        	long now = getProcessingTimeService().getCurrentProcessingTime();
        	getProcessingTimeService().registerTimer(now + watermarkInterval, this);
        }
 *  WatermarkAssignerOperator小结  
    其实每条记录都会通过设置的event\_time字段计算出水位，如果比当前系统记录的`currentWatermark`更大就会更新`currentWatermark`为计算的值。
    
    而有个周期性任务（周期由`autoWatermarkInterval`决定），会判断`currentWatermark`是否大于`lastWatermark`，大于就更新lastWatermark并发送水位到下游。

### 2.1.4 WindowOperator对Watermark处理 ###

#### 2.1.4.1 概述 ####

收到水位后，`WindowOperator`必须先将由水位导致的计算全部做完并产生输出，最后再计算、发送水位到下游。

WindowOperator会先评估所有会被触发的窗口，在计算后将结果下发，最后再发送水位到下游。

#### 2.1.4.2 源码分析 ####

相关代码可以参考`org.apache.flink.streaming.api.operators.AbstractStreamOperator`，他是WindowOperator的父类，以下方法被算子用来处理收到收到来自上游发送的水位。

public void processWatermark(Watermark mark) throws Exception { 
    	if (timeServiceManager != null) { 
    		// 1. 提升当前算子水位
    		// 2. 遍历已注册的Timer的优先级队列，调用triggerTarget.onEventTime(timer)
    		// 3. 比如WindowOperator就是尝试触发window计算
    		timeServiceManager.advanceWatermark(mark);
    	}
    	// 最后发送水位到下游
    	output.emitWatermark(mark);
    }
    
    public void processWatermark1(Watermark mark) throws Exception { 
    	input1Watermark = mark.getTimestamp();
    	long newMin = Math.min(input1Watermark, input2Watermark);
    	if (newMin > combinedWatermark) { 
    		combinedWatermark = newMin;
    		processWatermark(new Watermark(combinedWatermark));
    	}
    }
    
    public void processWatermark2(Watermark mark) throws Exception { 
    	input2Watermark = mark.getTimestamp();
    	long newMin = Math.min(input1Watermark, input2Watermark);
    	if (newMin > combinedWatermark) { 
    		combinedWatermark = newMin;
    		processWatermark(new Watermark(combinedWatermark));
    	}
    }

下面看看`InternalTimerServiceImpl#advanceWatermark`

public void advanceWatermark(long time) throws Exception { 
    	// 提升当前算子水位
    	currentWatermark = time;
    
    	InternalTimer<K, N> timer;
    	
    	// 遍历已注册的Timer的优先级队列，调用triggerTarget.onEventTime(timer)
    	while ((timer = eventTimeTimersQueue.peek()) != null && timer.getTimestamp() <= time) { 
    		eventTimeTimersQueue.poll();
    		keyContext.setCurrentKey(timer.getKey());
    		// 触发onEventTime方法，如 WindowOperator#onEventTime
    		triggerTarget.onEventTime(timer);
    	}
    }

这里再看看`org.apache.flink.table.runtime.operators.window.WindowOperator#onEventTime`:

@Override
    public void onEventTime(InternalTimer<K, W> timer) throws Exception { 
    	setCurrentKey(timer.getKey());
    
    	triggerContext.window = timer.getNamespace();
    	
    	// 利用Trigger，根据时间戳判断是否需要触发window
    	if (triggerContext.onEventTime(timer.getTimestamp())) { 
    		// fire
    		emitWindowResult(triggerContext.window);
    	}
    
    	if (windowAssigner.isEventTime()) { 
    		windowFunction.cleanWindowIfNeeded(triggerContext.window, timer.getTimestamp());
    	}
    }

`org.apache.flink.table.runtime.operators.window.WindowOperator.TriggerContext#onEventTime`

boolean onEventTime(long time) throws Exception { 
    	// 如调用AfterEndOfWindow#onEventTime
    	// 判断窗口是否触发
    	return trigger.onEventTime(time, window);
    }

### 2.1.5 水位与窗口触发小结 ###

1.  在Source使用`ctx.collectWithTimestamp`收集timestamp，为元素分配时间戳，并在元素有水位标志时使用`ctx.emitWatermark`发送水位；
2.  或使用`WatermarkStrategy`接口的`createTimestampAssigner`方法创建`TimestampAssigner`来使用`extractTimestamp`方法从元素中获取时间戳；调用`createWatermarkGenerator`创建`WatermarkGenerator`，使用其`currentWatermark`方法来从记录中生成水位。
3.  收到一条记录时，先将记录发送到下游，随后提取时间戳，计算水位，最后根据具体情况决定是否`advanceWatermark`并将水位发送到下游
4.  `WindowOperator`收到水位后，调用`AbstractStreamOperator#processWatermark`方法进行处理：
    
    1.  提升算子当前水位
    2.  遍历已注册的Timer的优先级队列，调用`triggerTarget.onEventTime(timer)`  
        比如WindowOperator就是尝试触发window计算，会根据`trigger#onEventTime(time, window)`判断是否应该触发窗口计算，如果需要就立刻触发计算，并将结果封装为`GenericRowData`发送到下游
5.  `WindowOperator#processElement`在处理每条收到的数据时：
    
    1.  会先取出时间戳（EventTime就根据定义的列取，处理时间就直接取当前系统时间）
    2.  分配该条数据到窗口
    3.  遍历所有分配的窗口，调用`Trigger#onElement`方法判断是否该窗口应该触发  
        比如`EventTimeTriggers.AfterEndOfWindow`就是当窗口最大时间戳小于等于当前水位时就应该触发，如果不应触发就注册一个 `TimerHeapInternalTimer`（带有当前窗口信息）到`InternalTimerServiceImpl.eventTimeTimersQueue`的优先级队列
    4.  如果需要触发，就触发窗口计算并发送到下游
    5.  最后为该窗口注册CleanupTimer  
        如果是EventTime，则使用`window.maxTimestamp() + allowedLateness`；否则使用`window.maxTimestamp()`
    6.  如果是迟到事件，就抛弃或放入`sideOutput`

# 3 滑动窗口 #

# 4 窗口与状态和Checkpoint #

*  非Window下，普通状态属于`VoidNamespace`。
 *  Window场景时，每个Window 会属于不同的 namespace，由 State/Checkpoint 来保证数据的`Exactly Once`语义

Checkpoint发生时记录下当前窗口的部分数据。

恢复时，将恢复数据还是放在这个窗口，新来的数据也放入该窗口即可，依然可以保证exactly-once（EventTime Window）。

可看看窗口算子处理每条元素的`org.apache.flink.streaming.runtime.operators.windowing.WindowOperator#processElement`：

@Override
    public void processElement(StreamRecord<IN> element) throws Exception { 
    	// 将元素分配到多个window
    	final Collection<W> elementWindows = windowAssigner.assignWindows(
    		element.getValue(), element.getTimestamp(), windowAssignerContext);
    
    	//if element is handled by none of assigned elementWindows
    	boolean isSkippedElement = true;
    
    	final K key = this.<K>getKeyedStateBackend().getCurrentKey();
    
    	if (windowAssigner instanceof MergingWindowAssigner) { 
    		MergingWindowSet<W> mergingWindows = getMergingWindowSet();
    
    		for (W window: elementWindows) { 
    
    			// adding the new window might result in a merge, in that case the actualWindow
    			// is the merged window and we work with that. If we don't merge then
    			// actualWindow == window
    			W actualWindow = mergingWindows.addWindow(window, new MergingWindowSet.MergeFunction<W>() { 
    				@Override
    				public void merge(W mergeResult,
    						Collection<W> mergedWindows, W stateWindowResult,
    						Collection<W> mergedStateWindows) throws Exception { 
    
    					if ((windowAssigner.isEventTime() && mergeResult.maxTimestamp() + allowedLateness <= internalTimerService.currentWatermark())) { 
    						throw new UnsupportedOperationException("The end timestamp of an " +
    								"event-time window cannot become earlier than the current watermark " +
    								"by merging. Current watermark: " + internalTimerService.currentWatermark() +
    								" window: " + mergeResult);
    					} else if (!windowAssigner.isEventTime()) { 
    						long currentProcessingTime = internalTimerService.currentProcessingTime();
    						if (mergeResult.maxTimestamp() <= currentProcessingTime) { 
    							throw new UnsupportedOperationException("The end timestamp of a " +
    								"processing-time window cannot become earlier than the current processing time " +
    								"by merging. Current processing time: " + currentProcessingTime +
    								" window: " + mergeResult);
    						}
    					}
    
    					triggerContext.key = key;
    					triggerContext.window = mergeResult;
    
    					triggerContext.onMerge(mergedWindows);
    
    					for (W m: mergedWindows) { 
    						triggerContext.window = m;
    						triggerContext.clear();
    						deleteCleanupTimer(m);
    					}
    
    					// merge the merged state windows into the newly resulting state window
    					windowMergingState.mergeNamespaces(stateWindowResult, mergedStateWindows);
    				}
    			});
    
    			// drop if the window is already late
    			if (isWindowLate(actualWindow)) { 
    				mergingWindows.retireWindow(actualWindow);
    				continue;
    			}
    			isSkippedElement = false;
    
    			W stateWindow = mergingWindows.getStateWindow(actualWindow);
    			if (stateWindow == null) { 
    				throw new IllegalStateException("Window " + window + " is not in in-flight window set.");
    			}
    
    			windowState.setCurrentNamespace(stateWindow);
    			windowState.add(element.getValue());
    
    			triggerContext.key = key;
    			triggerContext.window = actualWindow;
    
    			TriggerResult triggerResult = triggerContext.onElement(element);
    
    			if (triggerResult.isFire()) { 
    				ACC contents = windowState.get();
    				if (contents == null) { 
    					continue;
    				}
    				emitWindowContents(actualWindow, contents);
    			}
    
    			if (triggerResult.isPurge()) { 
    				windowState.clear();
    			}
    			registerCleanupTimer(actualWindow);
    		}
    
    		// need to make sure to update the merging state in state
    		mergingWindows.persist();
    	} else { 
    		// 非合并窗口（SessionWindow就是合并窗口）
    		
    		// 遍历当前元素所分配到的多个window
    		for (W window: elementWindows) { 
    
    			// drop if the window is already late
    			// 使用eventtime且窗口右边界时间小于等于当前水位
    			// 说明窗口已经过期，忽略
    			if (isWindowLate(window)) { 
    				continue;
    			}
    			isSkippedElement = false;
    			
    			// 存到state，注意namespace是当前window
    			windowState.setCurrentNamespace(window);
    			windowState.add(element.getValue());
    
    			triggerContext.key = key;
    			triggerContext.window = window;
    			
    			// 判断当前元素是否应该触发所属的当前遍历的那个window
    			TriggerResult triggerResult = triggerContext.onElement(element);
    
    			if (triggerResult.isFire()) { 
    				// 触发计算，得到结果
    				ACC contents = windowState.get();
    				if (contents == null) { 
    					continue;
    				}
    				// 发送窗口计算结果
    				emitWindowContents(window, contents);
    			}
    
    			if (triggerResult.isPurge()) { 
    				windowState.clear();
    			}
    			registerCleanupTimer(window);
    		}
    	}
    
    	// side output input event if
    	// element not handled by any window
    	// late arriving tag has been set
    	// windowAssigner is event time and current timestamp + allowed lateness no less than element timestamp
    	if (isSkippedElement && isElementLate(element)) { 
    		if (lateDataOutputTag != null){ 
    			sideOutput(element);
    		} else { 
    			this.numLateRecordsDropped.inc();
    		}
    	}
    }

[Flink]: https://blog.csdn.net/baichoufei90/article/details/105674793
[Link 1]: https://blog.csdn.net/baichoufei90/article/details/90264272
[Flink_1-_-]: https://blog.csdn.net/baichoufei90/article/details/82875496#window
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center]: /images/20230521/2ee34b1f32ff4e929c5242d0a4ad5654.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70]: /images/20230521/2e0edac0c85d497389079f4b7e925718.png
[Flink-]: https://blog.csdn.net/baichoufei90/article/details/108164777