往期推薦：
本篇終於到了Flink的核心內容：時間與水印。最初接觸這個概念是在Spark Structured Streaming中，一直無法理解水印的作用。直到使用了一段時間Flink之后，對實時流處理有了一定的理解，才想清楚其中的緣由。接下來就來介紹下Flink中的時間和水印，以及基於時間特性支持的窗口處理。
1 時間和水印
1.1 介紹
Flink支持不同的時間類型：
- 事件時間：事件發生的時間，是設備生產或存儲事件的時間，一般都直接存儲在事件上，比如Mysql Binglog中的修改時間；或者用戶訪問日志的訪問時間等。
- 攝入時間：事件進入Flink的時間，這個時間不常用。
- 處理時間：某個特殊的算子處理事件的時間，當不在意事件的順序時，為了保證高吞吐低延遲，會采用這種時間。
比如想要計算給定某天的第一個小時的股票價格趨勢，就需要使用事件時間。如果選擇處理時間進行計算，那么將會按照當前Flink應用處理的時間進行統計，就可能會造成數據一致性問題，歷史數據的分析也很難復現。還有個典型的場景是流式處理往往是7*24小時不間斷的運行，加入使用處理時間，當中間停機進行代碼更新或者BUG處理時，再次啟動，中間未處理的數據會堆積當重啟時間一次性處理，這樣對統計結果就造成大大的干擾。
1.2 使用EventTime
Flink默認使用的是處理時間，可以通過下面的方法修改成事件時間：
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
如果需要使用事件時間，還需要提供時間抽取器和水印生成器，這樣Flink才可以追蹤到事件時間的處理進度。
1.3 水印
通過下面的例子，可以了解為什么需要水印，水印是怎么工作的。在這個例子中，每個事件都帶有一個時間標識，下面的數字就是事件上的時間，很明顯它們是亂序到達的。第一個到達的是4，然后是2：
23 19 22 24 21 14 17 13 12 15 9 11 7 2 4(第一個事件)
加入現在希望對流進行排序，那么每個事件到達的時候，就需要產生一個流，按照時間戳排好序輸出每個到達的事件。
- 上帝視角：第一個到達的事件是4，但是不能立刻就把它當做第一個元素放入排序流中，因為現在事件是亂序的，無法確定前面的事件是否已經到達。當然現在你已經看到完整的事件順序，當然會知道只要再等待一個事件，4之前的事件就都處理完了（這就是上帝視角），但在現實中我們是一條條接收的數據，無法知道4后面出現的是2。
- 緩存和延遲：如果使用緩存，那么很有可能會永遠停止等待。第一個事件是4，第二個事件是2，我們是不是只需要等待一個事件就能保證事件的完整？可能是，也可能不是，比如現在事件就永遠等待不到1。
- 排序策略：對於任何給定的時間事件停止等待之前的數據，直接進行排序。這就是水印的作用：用來定義何時停止等待更早的數據。Flink中的事件時間處理依賴於水印生成器，每當元素進入到Flink，會根據其事件時間，生成一個新的時間戳，即水印。對於t時間的水印，意味着Flink不會再接收t之前的數據，那么t之前的數據就可以進行排序產出順序流了。在上面的例子中，當水印的時間戳到達2時，就會把2事件輸出。
- 水印策略：每當事件延遲到達時，這些延遲都不是固定的，一種簡單的方式是按照最大的延遲事件來判斷。對於大部分的應用，這種固定水印都可以工作的比較好。
1.4 延遲和完整性
在批處理中，用戶可以一次性看到全部的數據，因此可以很容易的知道事件的順序。在流處理中總需要等待一段時間，確定事件完整后才能產生結果。可以很激進的配置一個較短的水印延遲時間，這樣雖然輸入結果不完整（有的時間延遲還未到達就已經開始計算），但是速度會很快。或者設置較長的延遲，數據會相對完整，但是會有一定的延遲。也可以采用混合的策略，剛開始延遲小一點，當處理了部分數據后，延遲增加。
1.5 延時
延時通過水印來定義，Watermark(t)代表了t時間的事件是完整的，即小於t的事件都可以開始處理了。
1.6 使用水印
為了支撐事件時間機制的處理，Flink需要知道每個事件的時間，然后為其產生一個水印。
DataStream<Event> stream = ...
WatermarkStrategy<Event> strategy = WatermarkStrategy
.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(20))
// 選擇時間字段
.withTimestampAssigner((event, timestamp) -> event.timestamp);
DataStream<Event> withTimestampsAndWatermarks =
// 定義水印生成的策略
stream.assignTimestampsAndWatermarks(strategy);
2 窗口
Flink擁有豐富的窗口語義，接下來將會了解到：
- 如何在無限數據流上使用窗口聚合數據
- Flink都支持什么類型的窗口
- 如何實現一個窗口聚合
2.1 介紹
當進行流處理時很自然的想針對一部分數據聚合分析，比如想要統計每分鍾有多少瀏覽、每周每個用戶有多少次會話、每分鍾每個傳感器的最大溫度等。Flink的窗口分析依賴於兩個抽象概念：窗口分配器Assigner（用來指定事件屬於哪個窗口，在必要的時候新建窗口），窗口函數Function（應用於窗口內的數據）。Flink的窗口也有觸發器Trigger的概念，它決定了何時調用窗口函數進行處理；Evictor用於剔除窗口中不需要計算的數據。可以像下面這樣創建窗口：
stream.
.keyBy(<key selector>)
.window(<window assigner>)
.reduce|aggregate|process(<window function>)
也可以在非key數據流上使用窗口，但是一定要小心，因為處理過程將不會並行執行：
stream.
.windowAll(<window assigner>)
.reduce|aggregate|process(<window function>)
2.2 窗口分配器
Flink有幾種內置的窗口分配器:
按照窗口聚合的種類可以大致分為：
- 滾動窗口：比如統計每分鍾的瀏覽量，
TumblingEventTimeWindows.of(Time.minutes(1))
- 滑動窗口：比如每10秒鍾統計一次一分鍾內的瀏覽量，
SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10))
- 會話窗口：統計會話內的瀏覽量，會話的定義是同一個用戶兩次訪問不超過30分鍾，
EventTimeSessionWindows.withGap(Time.minutes(30))
窗口的時間可以通過下面的幾種時間單位來定義：
- 毫秒，
Time.milliseconds(n)
- 秒，
Time.seconds(n)
- 分鍾，
Time.minutes(n)
- 小時，
Time.hours(n)
- 天，
Time.days(n)
基於時間的窗口分配器支持事件時間和處理時間，這兩種類型的時間處理的吞吐量會有差別。使用處理時間優點是延遲很低，但是也存在幾個缺點：無法正確的處理歷史數據；無法處理亂序數據；結果非冪等。當使用基於數量的窗口，如果數量不夠，可能永遠不會觸發窗口操作。沒有選項支持超時處理或部分窗口的處理，當然你可以通過自定義窗口的方式來實現。全局窗口分配器會在一個窗口內，統一分配每個事件。如果需要自定義窗口，一般會基於它來做。不過推薦直接使用ProcessFunction。
2.3 窗口函數
有三種選擇來處理窗口中的內容：
- 當做批處理，使用
ProcessWindowFunction，基於Iterable處理窗口內容
- 增量的使用
ReduceFunction和AggregateFunction依次處理窗口的每個數據
- 上面兩者結合，使用
ReduceFunction和AggregateFunction進行預聚合，然后使用ProcessFunction進行批量處理。
下面給出了方法1和方法3的例子，需求為在每分鍾內尋找到每個傳感器的值，產生<key,>的結果流。
2.3.1 ProcessWindowFunction的例子
DataStream<SensorReading> input = ...
input
.keyBy(x -> x.key)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.process(new MyWastefulMax());
public static class MyWastefulMax extends ProcessWindowFunction<
SensorReading, // input type
Tuple3<String, Long, Integer>, // output type
String, // key type
TimeWindow> { // window type
@Override
public void process(
String key,
Context context,
Iterable<SensorReading> events,
Collector<Tuple3<String, Long, Integer>> out) {
int max = 0;
for (SensorReading event : events) {
max = Math.max(event.value, max);
}
out.collect(Tuple3.of(key, context.window().getEnd(), max));
}
}
有一些內容需要了解：
-
所有窗口分配的時間都在Flink中按照key緩存起來，直到窗口觸發，因此代價很昂貴。
-
ProcessWindowFunction中傳入了Context對象，內部包含了對應的窗口信息，接口類似：
public abstract class Context implements java.io.Serializable {
public abstract W window();
public abstract long currentProcessingTime();
public abstract long currentWatermark();
public abstract KeyedStateStore windowState();
public abstract KeyedStateStore globalState();
}
其中windowState和globalState會為每個key、每個窗口或者全局存儲信息，當需要記錄窗口的某些信息的時候會很有用。
2.3.2 Incremental Aggregation例子
DataStream<SensorReading> input = ...
input
.keyBy(x -> x.key)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.reduce(new MyReducingMax(), new MyWindowFunction());
private static class MyReducingMax implements ReduceFunction<SensorReading> {
public SensorReading reduce(SensorReading r1, SensorReading r2) {
return r1.value() > r2.value() ? r1 : r2;
}
}
private static class MyWindowFunction extends ProcessWindowFunction<
SensorReading, Tuple3<String, Long, SensorReading>, String, TimeWindow> {
@Override
public void process(
String key,
Context context,
Iterable<SensorReading> maxReading,
Collector<Tuple3<String, Long, SensorReading>> out) {
SensorReading max = maxReading.iterator().next();
out.collect(Tuple3.of(key, context.window().getEnd(), max));
}
}
注意iterable只會執行一次，即只有MyReducingMax輸出的值才會傳入這里。
2.4 延遲事件
默認當使用基於事件時間窗口時，延遲事件會直接丟棄。有兩種方法可以處理這個問題：你可以把需要丟棄的事件重新搜集起來輸出到另一個流中，也叫側輸出；或者配置水印的延遲時間。
OutputTag<Event> lateTag = new OutputTag<Event>("late"){};
SingleOutputStreamOperator<Event> result = stream.
.keyBy(...)
.window(...)
.sideOutputLateData(lateTag)
.process(...);
DataStream<Event> lateStream = result.getSideOutput(lateTag);
通過指定允許延遲的間隔時間，當在允許的延遲范圍內，仍然可以分配到對應的窗口(窗口對應的狀態信息將會保留一段時間)。但是會導致對應窗口重新計算（也叫做延遲響應late firing）默認允許的延遲是0，也就是說一旦事件在水印之后就會被丟棄掉。
stream.
.keyBy(...)
.window(...)
.allowedLateness(Time.seconds(10))
.process(...);
當配置延遲后，只有那些在允許的延遲之外的數據會被丟棄或者使用側輸出搜集起來。
3 注意
Flink的窗口處理可能跟你想的不太一樣，基於在flink用戶郵件中常問的問題，整理如下
3.1 滑動窗口造成數據拷貝
滑動窗口會造成大量的窗口對象，並且會拷貝每個對象到對應的窗口中。比如，你的滑動窗口為每15分鍾統計24小時的窗口長度，那么每個時間將會復制到4*24=96個窗口中。
3.2 時間窗口會對齊到系統時間
如果使用1個小時的窗口，那么當應用在12:05啟動時，並不是說第一個窗口的時間范圍是到1:05，事實上第一個窗口的時間是12:05到01:00，只有55分鍾而已。注意，滾動窗口和滑動窗口都支持偏移值的參數配置。
3.3 窗口后面可以接窗口
比如：
stream
.keyBy(t -> t.key)
.timeWindow(<time specification>)
.reduce(<reduce function>)
.timeWindowAll(<same time specification>)
.reduce(<same reduce function>)
這樣的代碼能夠工作主要是因為第一個窗口輸出的內容系統會自動添加一個窗口結束的時間，后面的處理可以基於這個時間再次進行窗口操作，但是需要窗口的配置統一或者整數倍。
3.4 空窗口沒有輸出
只有對應的事件到達時，才會創建對應的窗口。因此如果沒有對應的事件，窗口就不會創建，因此也不會有任何輸出。
3.5 延遲數據造成延遲合並
對於會話窗口，實際上會為每個事件在一開始分配一個新的窗口，當新的事件到達時，會根據時間間隔合並窗口。因此如果事件延遲到達，很有可能會造成窗口的延遲合並。