2.5.2 WindowedStream的设计与实现
和其他DataStream操作相比，WindowedStream转换操作相对复杂一些，在本节我们结合前面学习的内容，继续了解WindowedStream转换操作的实现。
我们知道，如果将DataStream根据Key进行分组，生成KeyedStream数据集，然后在KeyedStream上执行window()转换操作，就会生成WindowedStream数据集。如果直接调用DataStream.windowAll()方法进行转换，就会生成AllWindowedStream数据集。WindowedStream和AllWindowedStream的主要区别在于是否按照Key进行分区处理，这里我们以WindowedStream为例讲解窗口转换操作的具体实现。
1.WindowAssigner设计与实现
如代码清单2-39所示，当用户调用KeyedStream.window()方法时，会创建WindowedStream转换操作。通过window()方法可以看出，此时需要传递WindowAssigner作为窗口数据元素的分配器，通过WindowAssigner组件，可以根据指定的窗口类型将数据元素分配到指定的窗口中。
代码清单2-39 KeyedStream.window()方法定义
public <W extends Window> WindowedStream<T, KEY, W> window(WindowAssigner<?
super T, W> assigner) {
return new WindowedStream<>(this, assigner);
}
接下来我们看WindowAssigner的具体实现。如图2-17所示，WindowAssigner作为抽象类，其子类实现是非常多的，例如基于事件时间实现的SlidingEventTimeWindows、基于处理时间实现的TumblingProcessingTimeWindows等。这些WindowAssigner根据窗口类型进行区分，且属于DataStream API中内置的窗口分配器，用户可以直接调用它们创建不同类型的窗口转换。
从图2-17中可以看出，SessionWindow类型的窗口比较特殊，在WindowAssigner的基础上又实现了MergingWindowAssigner抽象类，在MergingWindowAssigner抽象类中定义了MergeCallback接口。这样做的原因是SessionWindow的窗口长度不固定，SessionWindow窗口的长度取决于指定时间范围内是否有数据元素接入，然后动态地将接入数据切分成独立的窗口，最后完成窗口计算。此时涉及对窗口中的元素进行动态Merge操作，这里主要借助MergingWindowAssigner提供的mergeWindows()方法来实现。
图2-17 WindowAssigner UML关系图
在WindowAssigner中通过提供WindowAssignerContext上下文获取CurrentProcessingTime等时间信息。在WindowAssigner抽象类中提供了以下方法供子类选择。
·assignWindows()：定义将数据元素分配到对应窗口的逻辑。
·getDefaultTrigger()：获取默认的Trigger，也就是默认窗口触发器，例如EventTimeTrigger。
·getWindowSerializer()：获取WindowSerializer实现，默认为TimeWindow.Serializer()。
·isEventTime()：判断是否为基于EventTime时间类型实现的窗口。
如代码清单2-40所示，我们以SlidingEventTimeWindows为例进行说明。
代码清单2-40 SlidingEventTimeWindows.assignWindows()方法定义
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(...)'?");
}
}
在SlidingEventTimeWindows.assignWindows()方法中可以看出，assignWindows()方法的参数包含了当前数据元素element、timestamp和WindowAssignerContext的上下文信息，且方法主要包含如下逻辑。
·判断timestamp是否有效，然后根据窗口长度和滑动时间计算数据元素所属窗口的数量，再根据窗口数量创建窗口列表。
·调用TimeWindow.getWindowStartWithOffset()方法，确定窗口列表中最晚的窗口对应的WindowStart时间，并赋值给lastStart变量；然后从lastStart开始遍历，每次向前移动固定的slide长度；最后向windows窗口列表中添加创建的TimeWindow，在TimeWindow中需要指定窗口的起始时间和结束时间。
·返回创建的窗口列表windows，也就是当前数据元素所属的窗口列表。
创建的WindowAssigner实例会在WindowOperator中使用，输入一条数据元素时会调用WindowAssigner.assignWindows()方法为接入的数据元素分配窗口，WindowOperator会根据元素所属的窗口分别对数据元素进行处理。
当然还有其他类型的WindowAssigner实现，基本功能都是一样的，主要是根据输入的元素确定和分配窗口。对于SlidingWindow类型的窗口来讲，同一个数据元素可能属于多个窗口，主要取决于窗口大小和滑动时间长度；而对于TumpleWindow类型来讲，每个数据元素仅属于一个窗口。
2.Window Trigger的核心实现
Window Trigger决定了窗口触发WindowFunction计算的时机，当接入的数据元素通过WindowAssigner分配到不同的窗口后，数据元素会被不断地累积在窗口状态中。当满足窗口触发条件时，会取出当前窗口中的所有数据元素，基于指定的WindowFunction对窗口中的数据元素进行运算，最后产生窗口计算结果并发送到下游的算子中。
如图2-18所示，在DataStream API中，所有定义的Window Trigger继承自Trigger基本实现类。每种窗口的触发策略不同，相应的Trigger触发器也有所不同。例如TumblingProcessingTimeWindows对应的默认Trigger为ProcessingTimeTrigger，而SlidingEventTimeWindows默认对应的是EventTimeTrigger。
数据元素接入WindowOperator后，调用窗口触发器的onElement()方法，判断窗口是否满足触发条件。如果满足，则触发窗口计算操作。我们以EventTimeTrigger为例介绍Trigger的核心实现，如代码清单2-41所示。
图2-18 Trigger UML关系图
·当数据元素接入后，根据窗口中maxTimestamp是否大于当前算子中的Watermark决定是否触发窗口计算。如果符合触发条件，则返回TriggerResult.FIRE事件，这里的maxTimestamp实际上是窗口的结束时间减1，属于该窗口的最大时间戳。
·如果不满足以上条件，就会继续向TriggerContext中注册Timer定时器，等待指定的时间再通过定时器触发窗口计算，此时方法会返回TriggerResult.CONTINUE消息给WindowOperator，表示此时窗口不会触发计算，继续等待新的数据接入。
·当数据元素不断接入WindowOperator，不断更新Watermark时，只要Watermark大于窗口的右边界就会触发相应的窗口计算。
代码清单2-41 EventTimeTrigger.onElement()方法定义
public TriggerResult onElement(Object element, long timestamp, TimeWindow
window, TriggerContext ctx) throws Exception {
if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
// 如果Watermark超过窗口最大时间戳，则立即执行
return TriggerResult.FIRE;
} else {
ctx.registerEventTimeTimer(window.maxTimestamp());
return TriggerResult.CONTINUE;
}
}
在EventTimeTrigger.onElement()方法定义中我们可以看到，当窗口不满足触发条件时，会向TriggerContext中注册EventTimeTimer定时器，指定的触发时间为窗口中的最大时间戳。算子中的Watermark到达该时间戳时，会自动触发窗口计算，不需要等待新的数据元素接入。这里TriggerContext使用到的TimerService实际上就是我们在2.4.2节介绍过的InternalTimerService，EventTimeTimer会基于InternalTimerService的实现类进行存储和管理。
当Timer定时器到达maxTimestamp时就会调用EventTimeTrigger.onEventTime()方法。如代码清单2-42所示，在EventTimeTrigger.onEventTime()方法中，实际上会判断传入的事件时间和窗口的maxTimestamp是否相等，如果相等则返回TriggerResult.FIRE并触发窗口的统计计算。
代码清单2-42 EventTimeTrigger.onEventTime()方法定义
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext
ctx) {
return time == window.maxTimestamp() ?
TriggerResult.FIRE :
TriggerResult.CONTINUE;
}
对于其他类型的窗口触发器，在原理上和EventTimeTrigger基本相同，感兴趣的读者可以阅读相关代码实现。
3.WindowFunction的设计与实现
经过以上几个步骤，基本上就能够确认窗口的类型及相应的触发时机了。窗口符合触发条件之后，就会对窗口中已经积蓄的数据元素进行统计计算，以得到最终的统计结果。对窗口元素的计算逻辑定义则主要通过窗口函数来实现。
在WindowStream的计算中，将窗口函数分为两种类型：用户指定的聚合函数AggregateFunction和专门用于窗口计算的WindowFunction。对于大部分用户来讲，基本都是基于窗口做聚合类型的统计运算，因此只需要在WindowStream中指定相应的聚合函数，如ReduceFunction和AggregateFunction。而在WindowStream的计算过程中，实际上会通过WindowFunction完成更加复杂的窗口计算。
如图2-19所示，WindowFunction继承了Function接口，同时又被不同类型的聚合函数实现，例如实现窗口关联计算的CoGroupWindowFunction、在窗口中对元素进行Reduce操作的ReduceApplyWindowFunction。这些函数同时继承自WrappingFunction，WrappingFunction对WindowFunction进行了一层封装，主要通过继承AbstractRichFunction抽象类，拓展和实现了RichFunction提供的能力。
图2-19 WindowFunction UML类图
总而言之，窗口中的函数会将用户定义的聚合函数和WindowFunction进行整合，形成统一的RichWindowFunction，然后基于RichWindowFunction进行后续的操作。
如代码清单2-43所示，用户创建WindowStream后，将ReduceFunction传递给WindowStream.reduce()方法。在WindowStream.reduce()方法中可以看出，还需要将WindowFunction作为参数，但这里的WindowFunction会在WindowStream中创建PassThroughWindowFunction默认实现类。
代码清单2-43 WindowStream.reduce()方法定义
public <R> SingleOutputStreamOperator<R> reduce(
ReduceFunction<T> reduceFunction,
WindowFunction<T, R, K, W> function,
TypeInformation<R> resultType) {
if (reduceFunction instanceof RichFunction) {
throw new UnsupportedOperationException("ReduceFunction of reduce can
not be a RichFunction.");
}
//清理函数闭包
function = input.getExecutionEnvironment().clean(function);
reduceFunction = input.getExecutionEnvironment().clean(reduceFunction);
final String opName = generateOperatorName(windowAssigner, trigger,
evictor, reduceFunction, function);
KeySelector<T, K> keySel = input.getKeySelector();
OneInputStreamOperator<T, R> operator;
if (evictor != null) {
@SuppressWarnings({"unchecked", "rawtypes"})
TypeSerializer<StreamRecord<T>> streamRecordSerializer =
(TypeSerializer<StreamRecord<T>>) new StreamElementSerializer(input.
getType().createSerializer(getExecutionEnvironment().getConfig()));
ListStateDescriptor<StreamRecord<T>> stateDesc =
new ListStateDescriptor<>("window-contents", streamRecordSerializer);
operator =
new EvictingWindowOperator<>(windowAssigner,
windowAssigner.getWindowSerializer(getExecutionEnvironment().
getConfig()),
keySel,
input.getKeyType().createSerializer(getExecutionEnvironment().
getConfig()),
stateDesc,
new InternalIterableWindowFunction<>(new ReduceApplyWindowFunction
<>(reduceFunction, function)),
trigger,
evictor,
allowedLateness,
lateDataOutputTag);
} else {
ReducingStateDescriptor<T> stateDesc = new ReducingStateDescriptor
<>("window-contents",
reduceFunction,
input.getType().createSerializer(getExecutionEnvironment().
getConfig()));
operator =
new WindowOperator<>(windowAssigner,
windowAssigner.getWindowSerializer(getExecutionEnvironment().
getConfig()),
keySel,
input.getKeyType().createSerializer(getExecutionEnvironment().
getConfig()),
stateDesc,
new InternalSingleValueWindowFunction<>(function),
trigger,
allowedLateness,
lateDataOutputTag);
}
return input.transform(opName, resultType, operator);
}
最后实际上就是创建OneInputStreamOperator实例，StreamOperator会根据evictor数据剔除器是否为空，选择创建EvictingWindowOperator还是WindowOperator。在创建EvictingWindowOperator时，通过调用new ReduceApplyWindowFunction <?> (reduceFunction, function)合并ReduceFunction和WindowFunction，然后转换为InternalIterableWindowFunction函数供WindowOperator使用。接下来调用input.transform()方法将创建好的EvictingWindowOperator或WindowOperator实例添加到OneInputTransformation转换操作中。其他的窗口计算函数和Reduce聚合函数基本一致，这里不再赘述。