Window在流式计算中很重要，因为”流”是一个无终点的持续输入，所以通过window机制来分块，进行聚合等各种处理
Keyed vs Non-Keyed Windows
Non-key window是在整个流上进行分块，没法并行处理
Window Assigners
分为4种窗口类型，分别是：
Tumbling Windows；Sliding Windows；Session Windows；Global Windows
Window Functions
注意区分可以累加计算的Function，和必须把一个window内的所有值都缓存起来最后计算的Fucntion
ReduceFunction；AggregateFunction；FoldFunction
ProcessWindowFunction（需要缓存整个window的数据后再计算）
ProcessWindowFunction with IncrementalAggregation
Using per-window state inProcessWindowFunction
Triggers
什么时候触发window的计算？一般是window结束的时候，涉及到系统时钟时间、watermark等，也可以自定义trigger。
Fire and Purge；Default Triggers of WindowAssigners；Built-inand Custom Triggers
Evictors
用来在window的trigger触发后，排除一些值
Allowed Lateness
一般违反watermark规则的超时数据被丢弃，如果实际场景特殊，也可以设置不丢弃超时情况。不丢弃就意味着对之前的结果做更改，需要缓存之前的结果，所以有一些性能开销。下面是些注意事项，具体细节可以参考官方文档
Gettinglate data as a side output
Lateelements considerations
Workingwith window results
Interactionof watermarks and windows
Consecutivewindowed operations
Useful state size considerations
接下来通过一个实际例子来说明window的使用方法：
滴滴司机每小时的最高收入
我们乘坐滴滴打车的时候，每次付款都会向滴滴公司发送一个消息，里面包括行程情况和所付的费用，如果滴滴想了解每小时司机的收入，最高收入是多少，可以通过下面的代码实现：
// configure thedata source
DataStream<DidiFare> fares = xxxx;
// compute tipsper hour for each driver
DataStream<Tuple3<Long, Long, Float>> hourlyTips = fares
.keyBy((DidiFare fare) ->fare.driverId).timeWindow(Time.hours(1))
.aggregate(new HourlyTipsSolution.AddTips(), newHourlyTipsSolution.WrapWithWindowInfo());
// find thehighest total tips in each hour
DataStream<Tuple3<Long, Long, Float>> hourlyMax = hourlyTips
.timeWindowAll(Time.hours(1)).maxBy(2);
// print theresult on stdout
printOrTest(hourlyMax);
上面的代码首先计算每个司机，每隔一小时的收入之和；然后基于上一步的数据(上一步的数据形成一个新的流)，每隔一小时计算出收入最大的那个。
public static class AddTips implements AggregateFunction<
DidiFare, // input type
Float, // accumulator type
Float // output type
>
{
@Override
public Float createAccumulator() {
return 0F;
}
@Override
public Float add(DidiFare fare, Float aFloat) {
return fare.tip + aFloat;
}
@Override
public Float getResult(Float aFloat) {
return aFloat;
}
@Override
public Float merge(Float aFloat, Float accumulator) {
return aFloat + accumulator;
}
}
上面是个自定义的aggregate函数，在window中不必缓存数据，任一时刻只有一个值。
public static class WrapWithWindowInfo extends ProcessWindowFunction<
Float, Tuple3<Long, Long, Float>, Long, TimeWindow> {
@Override
public void process(Long key, Context context, Iterable<Float> elements,
Collector<Tuple3<Long, Long, Float>> out) throws Exception {
Float sumOfTips = elements.iterator().next();//after aggregation, there is only one value
out.collect(new Tuple3<>(context.window().getEnd(), key, sumOfTips));
}
}
上面的代码是把window结果，结合window的时间熟悉，拼成一个新的流的元素。
欢迎阅读，有问题可以通过邮件kaiyuan0989爱特163.com一起探讨。
