【趙強老師】Flink的Watermark機制（基於Flink 1.11.0實現）
在使用eventTime的時候如何處理亂序資料？我們知道，流處理從事件產生，到流經source，再到operator，中間是有一個過程和時間的。雖然大部分情況下，流到operator的資料都是按照事件產生的時間順序來的，但是也不排除由於網路延遲等原因，導致亂序的產生，特別是使用kafka的話，多個分割槽的資料無法保證有序。所以在進行window計算的時候，我們又不能無限期的等下去，必須要有個機制來保證一個特定的時間後，必須觸發window去進行計算了。這個特別的機制，就是watermark。Watermark是用於處理亂序事件的，用於衡量Event Time進展的機制。watermark可以翻譯為水位線。
一、Watermark的核心原理
Watermark的核心本質可以理解成一個延遲觸發機制。
在 Flink 的視窗處理過程中，如果確定全部資料到達，就可以對 Window 的所有資料做 視窗計算操作（如彙總、分組等），如果資料沒有全部到達，則繼續等待該視窗中的資料全 部到達才開始處理。這種情況下就需要用到水位線（WaterMarks）機制，它能夠衡量資料處 理進度（表達資料到達的完整性），保證事件資料（全部）到達 Flink 系統，或者在亂序及 延遲到達時，也能夠像預期一樣計算出正確並且連續的結果。當任何 Event 進入到 Flink 系統時，會根據當前最大事件時間產生 Watermarks 時間戳。
那麼 Flink 是怎麼計算 Watermak 的值呢？
Watermark =進入Flink 的最大的事件時間(mxtEventTime)-指定的延遲時間(t)
那麼有 Watermark 的 Window 是怎麼觸發視窗函式的呢？
如果有視窗的停止時間等於或者小於 maxEventTime - t(當時的warkmark)，那麼這個視窗被觸發執行。
其核心處理流程如下圖所示。
二、Watermark的三種使用情況
1、本來有序的Stream中的 Watermark
如果資料元素的事件時間是有序的，Watermark 時間戳會隨著資料元素的事件時間按順 序生成，此時水位線的變化和事件時間保持一直（因為既然是有序的時間，就不需要設定延遲了，那麼t就是 0。所以 watermark=maxtime-0 = maxtime），也就是理想狀態下的水位 線。當 Watermark 時間大於 Windows 結束時間就會觸發對 Windows 的資料計算，以此類推， 下一個 Window 也是一樣。這種情況其實是亂序資料的一種特殊情況。
2、亂序事件中的Watermark
現實情況下資料元素往往並不是按照其產生順序接入到 Flink 系統中進行處理，而頻繁 出現亂序或遲到的情況，這種情況就需要使用 Watermarks 來應對。比如下圖，設定延遲時間t為2。
3、並行資料流中的Watermark
在多並行度的情況下，Watermark 會有一個對齊機制，這個對齊機制會取所有 Channel 中最小的 Watermark。
三、設定Watermark的核心程式碼
1、首先，正確設定事件處理的時間語義，一般都是採用Event Time。
sEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
2、其次，指定生成Watermark的機制，包括：延時處理的時間和EventTime對應的欄位。如下：
注意：不管是資料是否有序，都可以使用上面的程式碼。有序的資料只是無序資料的一種特殊情況。
四、Watermark程式設計案例
測試資料：基站的手機通話資料，如下：
需求：按基站，每5秒統計通話時間最長的記錄。
- StationLog用於封裝基站資料
package watermark;
//station1,18688822219,18684812319,10,1595158485855
public class StationLog {
private String stationID; //基站ID
private String from; //呼叫放
private String to; //被叫方
private long duration; //通話的持續時間
private long callTime; //通話的呼叫時間
public StationLog(String stationID, String from,
String to, long duration,
long callTime) {
this.stationID = stationID;
this.from = from;
this.to = to;
this.duration = duration;
this.callTime = callTime;
}
public String getStationID() {
return stationID;
}
public void setStationID(String stationID) {
this.stationID = stationID;
}
public long getCallTime() {
return callTime;
}
public void setCallTime(long callTime) {
this.callTime = callTime;
}
public String getFrom() {
return from;
}
public void setFrom(String from) {
this.from = from;
}
public String getTo() {
return to;
}
public void setTo(String to) {
this.to = to;
}
public long getDuration() {
return duration;
}
public void setDuration(long duration) {
this.duration = duration;
}
}
- 程式碼實現：WaterMarkDemo用於完成計算（注意：為了方便咱們測試設定任務的並行度為1）
package watermark;
import java.time.Duration;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
//每隔五秒，將過去是10秒內，通話時間最長的通話日誌輸出。
public class WaterMarkDemo {
public static void main(String[] args) throws Exception {
//得到Flink流式處理的執行環境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.setParallelism(1);
//設定週期性的產生水位線的時間間隔。當資料流很大的時候，如果每個事件都產生水位線，會影響效能。
env.getConfig().setAutoWatermarkInterval(100);//預設100毫秒
//得到輸入流
DataStreamSource<String> stream = env.socketTextStream("bigdata111", 1234);
stream.flatMap(new FlatMapFunction<String, StationLog>() {
public void flatMap(String data, Collector<StationLog> output) throws Exception {
String[] words = data.split(",");
// 基站ID from to 通話時長 callTime
output.collect(new StationLog(words[0], words[1],words[2], Long.parseLong(words[3]), Long.parseLong(words[4])));
}
}).filter(new FilterFunction<StationLog>() {
@Override
public boolean filter(StationLog value) throws Exception {
return value.getDuration() > 0?true:false;
}
}).assignTimestampsAndWatermarks(WatermarkStrategy.<StationLog>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner(new SerializableTimestampAssigner<StationLog>() {
@Override
public long extractTimestamp(StationLog element, long recordTimestamp) {
return element.getCallTime(); //指定EventTime對應的欄位
}
})
).keyBy(new KeySelector<StationLog, String>(){
@Override
public String getKey(StationLog value) throws Exception {
return value.getStationID(); //按照基站分組
}}
).timeWindow(Time.seconds(5)) //設定時間視窗
.reduce(new MyReduceFunction(),new MyProcessWindows()).print();
env.execute();
}
}
//用於如何處理視窗中的資料，即：找到視窗內通話時間最長的記錄。
class MyReduceFunction implements ReduceFunction<StationLog> {
@Override
public StationLog reduce(StationLog value1, StationLog value2) throws Exception {
// 找到通話時間最長的通話記錄
return value1.getDuration() >= value2.getDuration() ? value1 : value2;
}
}
//視窗處理完成後，輸出的結果是什麼
class MyProcessWindows extends ProcessWindowFunction<StationLog, String, String, TimeWindow> {
@Override
public void process(String key, ProcessWindowFunction<StationLog, String, String, TimeWindow>.Context context,
Iterable<StationLog> elements, Collector<String> out) throws Exception {
StationLog maxLog = elements.iterator().next();
StringBuffer sb = new StringBuffer();
sb.append("視窗範圍是:").append(context.window().getStart()).append("----").append(context.window().getEnd()).append("\n");;
sb.append("基站ID：").append(maxLog.getStationID()).append("\t")
.append("呼叫時間：").append(maxLog.getCallTime()).append("\t")
.append("主叫號碼：").append(maxLog.getFrom()).append("\t")
.append("被叫號碼：") .append(maxLog.getTo()).append("\t")
.append("通話時長：").append(maxLog.getDuration()).append("\n");
out.collect(sb.toString());
}
}