flink強(qiáng)大的窗口功能懊悯，是相較于其他流計(jì)算引擎比較有優(yōu)勢(shì)的地方嫉父。
flink中窗口是如何設(shè)計(jì)的隔盛？
一共有四個(gè)要素。
1）window assigner破托。對(duì)于流式的數(shù)據(jù)，如何來界定一個(gè)window歧蒋，也就是把數(shù)據(jù)劃分給正確的窗口土砂。
2） 窗口觸發(fā)器，trigger疏尿。提供靈活的cut point 瘟芝，滿足不同觸發(fā)邏輯的需求。
3） 窗口Incepter褥琐，evict锌俱。提供靈活的cut point，在串口計(jì)算前后可以做一些定制化的事情敌呈。
4） 窗口處理行數(shù)贸宏，processer。提供強(qiáng)大的處理接口磕洪，可以對(duì)窗口中的數(shù)據(jù)進(jìn)行運(yùn)算吭练。
1）首先看window assigner.
通常我們想把數(shù)據(jù)按照一定區(qū)間做一個(gè)劃分，比如10分鐘析显。每一個(gè)數(shù)據(jù)有一個(gè) event_time和process_time的時(shí)間概念鲫咽。
process_time 位系統(tǒng)處理這個(gè)數(shù)據(jù)的時(shí)間，而event——time則是用戶可以自己定義的和業(yè)務(wù)邏輯相關(guān)的時(shí)間。
比如分尸，處理一個(gè)股票trade流锦聊， trade 的交易所發(fā)生時(shí)間可以作為event——time，flink 處理這個(gè)流的時(shí)間位process——time箩绍。
對(duì)于使用event_time的流數(shù)據(jù)孔庭，因?yàn)榫W(wǎng)絡(luò)延遲的原因，通常會(huì)存在亂序材蛛，比如我定義一個(gè)【00：00圆到，00:10】的窗口
那么有些00：09 的數(shù)據(jù)，可能會(huì)在00：11的時(shí)候到達(dá)卑吭，flink 的窗口機(jī)制芽淡，必須要給與一定的容錯(cuò)性來處理這樣的亂序問題。
flink采用一種watermark水位線的機(jī)制陨簇，watermark作為流數(shù)據(jù)中的特殊事件存在吐绵。對(duì)于watermark，通常使用periodical 的機(jī)制生成河绽，
這樣可以提高效率己单，下面舉個(gè)例子說明具體的算法。
容忍的延遲位1分鐘
event_time, watermark
00:01 00:01-1=00:00
00:05 00:05-1 =00:04
00:02 00:04(取當(dāng)前最大)
00:03 00:04(取當(dāng)前最大)
00:06 00:05-1 =00:05
00:10 00:10-1 =00:09
00:11 00:11-1 =00:10
因?yàn)閣atermark到達(dá) 窗口邊界耙饰，這個(gè)時(shí)候就要觸發(fā)窗口計(jì)算了纹笼。后面來的數(shù)據(jù)要么被丟棄，
要么就是進(jìn)入用戶自定義的output中苟跪。
對(duì)于多并行度的流廷痘，每個(gè)子并行流都需要處理窗口的問題，達(dá)到觸發(fā)提交進(jìn)行窗口閉合件已。
2）trigger笋额，一般默認(rèn)的窗口觸發(fā)是 watermark 和窗口的 endTime 相等的時(shí)候，
flink 提供靈活的自定義方式篷扩，用戶可以任意使用窗口觸發(fā)機(jī)制來滿足自己的需求兄猩。
需要對(duì)窗口制定一個(gè)trigger，如下的CountTriggerDebug鉴未，當(dāng)窗口中的數(shù)據(jù)達(dá)到2條就觸發(fā)
DataStream<Tuple2<String, List<Integer>>> reduce = map.keyBy("f0")
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.trigger(CountTriggerDebug.of(2))
.reduce((new ReduceFunction<Tuple2<String, List<Integer>>>() {
@Override
public Tuple2<String, List<Integer>> reduce(Tuple2<String, List<Integer>> value1, Tuple2<String, List<Integer>> value2) {
value1.f1.add(value2.f1.get(0));
return value1;
}
}));
@Override
public TriggerResult onElement(Object element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
ReducingState<Long> count = ctx.getPartitionedState(stateDesc);
count.add(1L);
if (count.get() >= maxCount) {
System.out.println("觸發(fā)器觸發(fā)窗口函數(shù)對(duì)該窗口計(jì)算,同時(shí)清除該窗口的計(jì)數(shù)狀態(tài),--" + count.get());
count.clear();
// return TriggerResult.FIRE;
return TriggerResult.FIRE_AND_PURGE;
}
System.out.println("觸發(fā)器僅對(duì)該窗口的計(jì)數(shù)狀態(tài)進(jìn)行加一操作--" + count.get());
return TriggerResult.CONTINUE;
}
使用trigger的時(shí)候枢冤，需要關(guān)注狀態(tài) 是使用 FIRE_AND_PURGE 還是 FIRE。
如果只是FIRE的話铜秆，窗口中所有數(shù)據(jù)會(huì)傳到下游淹真。如果使用FIRE_PURGE,則下游只會(huì)收到增量數(shù)據(jù)，不至于影響性能连茧。
3）Evictor
Evictor機(jī)制提供給在窗口計(jì)算前和計(jì)算后做一些事情的能力核蘸。比如刪除一些特定的數(shù)據(jù)等巍糯。
指定一個(gè)實(shí)現(xiàn)了Evictor接口的類就可以達(dá)到這個(gè)效果。
DataStream<Tuple2<String, List<Integer>>> reduce = keyedStream
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.evictor(CountEvictor.of(3))
.reduce((new ReduceFunction<Tuple2<String, List<Integer>>>() {
@Override
public Tuple2<String, List<Integer>> reduce(Tuple2<String, List<Integer>> value1, Tuple2<String, List<Integer>> value2) {
value1.f1.add(value2.f1.get(0));
return value1;
}
}));
public interface Evictor<T, W extends Window> extends Serializable {
void evictBefore(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);
void evictAfter(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);
4） Processer
window提供了 reduce值纱，aggregate和 process函數(shù)鳞贷。需要注意的是
reduce和aggreate 是可以增量計(jì)算的坯汤，也就是不用等待所有的數(shù)據(jù)都到了窗口之后在計(jì)算虐唠。
可以預(yù)先計(jì)算好一部分，比如一些統(tǒng)計(jì)功能是可以增量運(yùn)算的惰聂，這樣可以很好的提高性能疆偿。
而對(duì)于一些復(fù)雜的功能，是需要全量數(shù)據(jù)作為先決條件的搓幌，這就需要用process 函數(shù)杆故。
