建站
咨詢
售后
建站咨詢
新聞中心
我們經(jīng)常需要在一個時間窗口維度上對數(shù)據(jù)進行聚合，窗口是流處理應用中經(jīng)常需要解決的問題。Flink的窗口算子為我們提供了方便易用的API，我們可以將數(shù)據(jù)流切分成一個個窗口，對窗口內(nèi)的數(shù)據(jù)進行處理，下面為大家詳細講解一下Flink的窗口操作。
資源ssl適用于網(wǎng)站、小程序/APP、API接口等需要進行數(shù)據(jù)傳輸應用場景，ssl證書未來市場廣闊！成為成都創(chuàng)新互聯(lián)公司的ssl證書銷售渠道，可以享受市場價格4-6折優(yōu)惠！如果有意向歡迎電話聯(lián)系或者加微信：18980820575（備注：SSL證書合作）期待與您的合作！
一、窗口（window）的類型
對于窗口的操作主要分為兩種，分別對于Keyedstream和Datastream。他們的主要區(qū)別也僅僅在于建立窗口的時候一個為.window(…)，一個為.windowAll(…)。對于Keyedstream的窗口來說，他可以使得多任務并行計算，每一個logical key stream將會被獨立的進行處理。
stream
.keyBy(...) "assigner"
[.trigger(...)] "trigger" (else default trigger)
[.evictor(...)] "evictor" (else no evictor)
[.allowedLateness(...)] "lateness" (else zero)
[.sideOutputLateData(...)] "output tag" (else no side output for late data)
.reduce/aggregate/fold/apply() "function"
[.getSideOutput(...)] "output tag"
按照窗口的Assigner來分，窗口可以分為
Tumbling window， sliding window，session window，global window，custom window
每種窗口又可分別基于processing time和event time，這樣的話，窗口的類型嚴格來說就有很多。
還有一種window叫做count window，依據(jù)元素到達的數(shù)量進行分配，之后也會提到。
窗口的生命周期開始在第一個屬于這個窗口的元素到達的時候，結(jié)束于第一個不屬于這個窗口的元素到達的時候。
二、窗口的操作
2.1 Tumbling window
固定相同間隔分配窗口，每個窗口之間沒有重疊看圖一眼明白。
下面的例子定義了每隔3毫秒一個窗口的流：
WindowedStream
Rates = rates .keyBy(MovieRate::getUserId) .window(TumblingEventTimeWindows.of(Time.milliseconds(3)));
2.2 Sliding Windows
跟上面一樣，固定相同間隔分配窗口，只不過每個窗口之間有重疊。窗口重疊的部分如果比窗口小，窗口將會有多個重疊，即一個元素可能被分配到多個窗口里去。
下面的例子給出窗口大小為10毫秒，重疊為5毫秒的流：
WindowedStream
Rates = rates .keyBy(MovieRate::getUserId) .window(SlidingEventTimeWindows.of(Time.milliseconds(10), Time.milliseconds(5)));
2.3 Session window
這種窗口主要是根據(jù)活動的事件進行窗口化，他們通常不重疊，也沒有一個固定的開始和結(jié)束時間。一個session window關閉通常是由于一段時間沒有收到元素。在這種用戶交互事件流中，我們首先想到的是將事件聚合到會話窗口中（一段用戶持續(xù)活躍的周期），由非活躍的間隙分隔開。
// 靜態(tài)間隔時間
WindowedStream
Rates = rates .keyBy(MovieRate::getUserId) .window(EventTimeSessionWindows.withGap(Time.milliseconds(10))); // 動態(tài)時間 WindowedStream
Rates = rates .keyBy(MovieRate::getUserId) .window(EventTimeSessionWindows.withDynamicGap(()));
2.4 Global window
將所有相同keyed的元素分配到一個窗口里。好吧，就這樣：
WindowedStream
Rates = rates .keyBy(MovieRate::getUserId) .window(GlobalWindows.create());
三、窗口函數(shù)
窗口函數(shù)就是這四個：ReduceFunction，AggregateFunction，F(xiàn)oldFunction，ProcessWindowFunction。前兩個執(zhí)行得更有效，因為Flink可以增量地聚合每個到達窗口的元素。
Flink必須在調(diào)用函數(shù)之前在內(nèi)部緩沖窗口中的所有元素，所以使用ProcessWindowFunction進行操作效率不高。不過ProcessWindowFunction可以跟其他的窗口函數(shù)結(jié)合使用，其他函數(shù)接受增量信息，ProcessWindowFunction接受窗口的元數(shù)據(jù)。
舉一個AggregateFunction的例子吧，下面代碼為MovieRate按user分組，且分配5毫秒的Tumbling窗口，返回每個user在窗口內(nèi)評分的所有分數(shù)的平均值。
DataStream
> Rates = rates .keyBy(MovieRate::getUserId) .window(TumblingEventTimeWindows.of(Time.milliseconds(5))) .aggregate(new AggregateFunction
>() { @Override public AverageAccumulator
createAccumulator() {
return new AverageAccumulator(); } @Override public AverageAccumulator add(MovieRate movieRate, AverageAccumulator acc) { acc.userId = movieRate.userId; acc.sum += movieRate.rate; acc.count++;
return acc; } @Override public Tuple2
getResult(AverageAccumulator acc) {
return Tuple2.of(acc.userId, acc.sum/(double)acc.count); } @Override public AverageAccumulator merge(AverageAccumulator acc0, AverageAccumulator acc1) { acc0.count += acc1.count; acc0.sum += acc1.sum;
return acc0; } }); public static class AverageAccumulator{ int userId; int count; double sum; }
以下是部分輸出：
...
1> (44,3.0)
4> (96,0.5)
2> (51,0.5)
3> (90,2.75)
...
看上面的代碼，會發(fā)現(xiàn)add()函數(shù)特別生硬，因為我們想返回Tuple2 類型，即Integer為key，但AggregateFunction似乎沒有提供這個機制可以讓AverageAccumulator的構(gòu)造函數(shù)提供參數(shù)。所以，這里引入ProcessWindowFunction與AggregateFunction的結(jié)合版，AggregateFunction進行增量疊加，當窗口關閉時，ProcessWindowFunction將會被提供AggregateFunction返回的結(jié)果，進行Tuple封裝：
DataStream
> Rates = rates .keyBy(MovieRate::getUserId) .window(TumblingEventTimeWindows.of(Time.milliseconds(5))) .aggregate(new MyAggregateFunction(), new MyProcessWindowFunction()); public static class MyAggregateFunction implements AggregateFunction
{ @Override public AverageAccumulator
createAccumulator() {
return new AverageAccumulator(); } @Override public AverageAccumulator add(MovieRate movieRate, AverageAccumulator acc) { acc.sum += movieRate.rate; acc.count++;
return acc; } @Override public Double getResult(AverageAccumulator acc) {
return acc.sum/(double)acc.count; } @Override public AverageAccumulator merge(AverageAccumulator acc0, AverageAccumulator acc1) { acc0.count += acc1.count; acc0.sum += acc1.sum;
return acc0; } } public static class MyProcessWindowFunction extends ProcessWindowFunction
, Integer, TimeWindow> { @Override public void process(Integer key, Context context, Iterable
results, Collector
> out) throws Exception { Double result = results.iterator().next(); out.collect(new Tuple2(key, result)); } } public static class AverageAccumulator{ int count; double sum; }
可以得到，結(jié)果與上面一樣，但代碼好看了很多。
四、其他操作
4.1 Triggers（觸發(fā)器）
觸發(fā)器定義了窗口何時準備好被窗口處理。每個窗口分配器默認都有一個觸發(fā)器，如果默認的觸發(fā)器不符合你的要求，就可以使用trigger(…)自定義觸發(fā)器。
通常來說，默認的觸發(fā)器適用于多種場景。例如，多有的event-time窗口分配器都有一個EventTimeTrigger作為默認觸發(fā)器。該觸發(fā)器在watermark通過窗口末尾時出發(fā)。
PS：GlobalWindow默認的觸發(fā)器時NeverTrigger，該觸發(fā)器從不出發(fā)，所以在使用GlobalWindow時必須自定義觸發(fā)器。
4.2 Evictors（驅(qū)逐器）
Evictors可以在觸發(fā)器觸發(fā)之后以及窗口函數(shù)被應用之前和/或之后可選擇的移除元素。使用Evictor可以防止預聚合，因為窗口的所有元素都必須在應用計算邏輯之前先傳給Evictor進行處理
4.3 Allowed Lateness
當使用event-time窗口時，元素可能會晚到，例如Flink用于跟蹤event-time進度的watermark已經(jīng)超過了窗口的結(jié)束時間戳。
默認來說，當watermark超過窗口的末尾時，晚到的元素會被丟棄。但是flink也允許為窗口operator指定最大的allowed lateness，以至于可以容忍在徹底刪除元素之前依然接收晚到的元素，其默認值是0。
為了支持該功能，F(xiàn)link會保持窗口的狀態(tài)，知道allowed lateness到期。一旦到期，flink會刪除窗口并刪除其狀態(tài)。
把晚到的元素當作side output。
SingleOutputStreamOperator
result = input .keyBy(
) .window(
) .allowedLateness(
) .sideOutputLateData(lateOutputTag) .
(
function>);
本文名稱：詳解Flink的窗口操作
文章出自：http://www.dlmjj.cn/article/dpidpsh.html
