Flink Window分析及Watermark解決亂序資料機制深入剖析-Flink牛刀小試

Socket · 發表 2018-11-25 01:50:34

摘要：版權宣告：本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。版權宣告：禁止轉載，歡迎學習。QQ郵箱地址：[email protected]，如有任何問題，可隨時聯絡。本文決心...

版權宣告：本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。版權宣告：禁止轉載，歡迎學習。QQ郵箱地址：[email protected]，如有任何問題，可隨時聯絡。

本文決心講清楚這個糾結的水印Watermark問題，Come on ！

1 The Time

針對stream資料中的時間，可以分為以下三種：

Event Time：事件產生的時間，它通常由事件中的時間戳描述。
Ingestion time：事件進入Flink的時間
Processing Time：事件被處理時當前系統的時間

Flink中，預設Time類似是ProcessingTime ，可以在程式碼中設定：

1.1 tips（請認真思考下面的話，絕對震聾發揮！）

在水印的處理中，我們一般取事件時間序列的最大值作為水印的生成時間參考。
按照訊號發生的順序，時間是不斷增加的，所以在時間序列上若出現事件時間小於時間序列最大值，一般都是延時的事件，時間序列最大值不會改變。
每處理一條事件資料，watermark時間就生成一次，後面窗的觸發就是依據水印時間。若設定亂序延時為10s，則生成規則就是：
```
final Long maxOutOfOrderness = 10000L;// 最大允許的亂序時間是10s
 new Watermark(currentMaxTimestamp - maxOutOfOrderness)
複製程式碼
```
資料會按照時間進行依次Append,
水印依賴的條件是窗，水印只是決定了窗的觸發時間，若設定最大允許的亂序時間是maxOutOfOrderness=10s，則窗的觸發時機就是：
```
watermark 時間 >= window_end_time
複製程式碼
```
窗觸發時，資料除了正常的時間序列，同時也包含延時到達的序列。在窗觸發前（也就水印時間），計算除了把之前的正常窗資料給觸發了，同時還包含了本來也屬於該窗的延時資料。

2 窗與水印的世紀謎題

事件時間的最大值，也就是當前的實際事件時間，因此需要以此為參考點。
實際窗：意思資料就在那裡Append,窗資料已經準備好，等待觸發時機。
水印時間不受影響：就是每次來的資料的事件時間最大值，不受延遲資料時間影響。
下面例子中，等水印時間為10:11:33時，滿足時間窗 10:11:30 <-> 10:11:33的觸發時機，此時需要處理的資料不僅包含正常資料10:11:21 ,同時還包含亂序資料10:11:31。
再次強調：窗時機到來時，會遍歷亂序資料和原窗資料。
實際窗在流動，只是暫不觸發。
水印也在標記流動
窗時機觸發也在流動。
watermark 時間 >= window_end_time時，觸發歷史窗執行。

3 EventTime和Watermarks 水位線理論碰撞

流處理從事件產生，到流經source，再到operator，中間是有一個過程和時間的。雖然大部分情況下，流到operator的資料都是按照事件產生的時間順序來的，但是也不排除由於網路延遲等原因，導致亂序的產生，特別是使用kafka的話，多個分割槽的資料無法保證有序。所以在進行window計算的時候，我們又不能無限期的等下去，必須要有個機制來保證一個特定的時間後，必須觸發window去進行計算了。這個特別的機制，就是watermark，watermark是用於處理亂序事件的。
通常，在接收到source的資料後，應該立刻生成watermark；但是，也可以在source後，應用簡單的map或者filter操作後，再生成watermark。注意：如果指定多次watermark，後面指定的會覆蓋前面的值。生成方式

With Periodic Watermarks

週期性的觸發watermark的生成和傳送，預設是100ms，每隔N秒自動向流裡
注入一個WATERMARK 時間間隔由ExecutionConfig.setAutoWatermarkInterval 
決定. 每次呼叫getCurrentWatermark 方法, 如果得到的WATERMARK
不為空並且比之前的大就注入流中 
可以定義一個最大允許亂序的時間，這種比較常用
實現AssignerWithPeriodicWatermarks介面
複製程式碼

With Punctuated Watermarks

基於某些事件觸發watermark的生成和傳送基於事件向流裡注入一個WATERMARK，
每一個元素都有機會判斷是否生成一個WATERMARK. 如果得到的WATERMARK
不為空並且比之前的大就注入流中實現AssignerWithPunctuatedWatermarks介面
複製程式碼

多並行度流的watermarks

注意：多並行度的情況下，watermark對齊會取所有channel最小的watermark。

4 With Periodic Watermarks案例實戰

4.1 最樸實的水印方案（基於事件序列最大值）

public class StreamingWindowWatermark {

public static void main(String[] args) throws Exception {
//定義socket的埠號
int port = 9010;
//獲取執行環境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//設定使用eventtime，預設是使用processtime
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//設定並行度為1,預設並行度是當前機器的cpu數量
env.setParallelism(1);
//連線socket獲取輸入的資料
DataStream<String> text = env.socketTextStream("SparkMaster", port, "\n");

//解析輸入的資料
DataStream<Tuple2<String, Long>> inputMap = text.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String value) throws Exception {
String[] arr = value.split(",");
return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
}
});

//抽取timestamp和生成watermark
DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {

Long currentMaxTimestamp = 0L;
final Long maxOutOfOrderness = 10000L;// 最大允許的亂序時間是10s

SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
 
@Nullable
@Override
public Watermark getCurrentWatermark() {
return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
}

//定義如何提取timestamp
@Override
public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
long timestamp = element.f1;
currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
long id = Thread.currentThread().getId();
System.out.println("作者：秦凱新 鍵值 :"+element.f0+",事件事件:[ "+sdf.format(element.f1)+" ],currentMaxTimestamp:[ "+
sdf.format(currentMaxTimestamp)+" ],水印時間:[ "+sdf.format(getCurrentWatermark().getTimestamp())+" ]");
return timestamp;
}
});

DataStream<String> window = waterMarkStream.keyBy(0)
.window(TumblingEventTimeWindows.of(Time.seconds(3)))//按照訊息的EventTime分配視窗，和呼叫TimeWindow效果一樣
.apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
/**
* 對window內的資料進行排序，保證資料的順序
* @param tuple
* @param window
* @param input
* @param out
* @throws Exception
*/
@Override
public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
String key = tuple.toString();
List<Long> arrarList = new ArrayList<Long>();
Iterator<Tuple2<String, Long>> it = input.iterator();
while (it.hasNext()) {
Tuple2<String, Long> next = it.next();
arrarList.add(next.f1);
}
Collections.sort(arrarList);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
String result = "\n 作者：秦凱新 鍵值 : "+ key + "\n觸發窗內資料個數 : " + arrarList.size() + "\n觸發窗起始資料： " + sdf.format(arrarList.get(0)) + "\n觸發窗最後（可能是延時）資料：" + sdf.format(arrarList.get(arrarList.size() - 1))
+ "\n實際窗起始和結束時間： " + sdf.format(window.getStart()) + "《----》" + sdf.format(window.getEnd()) + " \n \n ";

out.collect(result);
}
});
//測試-把結果列印到控制檯即可
window.print();

//注意：因為flink是懶載入的，所以必須呼叫execute方法，上面的程式碼才會執行
env.execute("eventtime-watermark");

}
}
複製程式碼

執行測試資料

0001,15383598820002018-10-01 10:11:22
 0002,15383598860002018-10-01 10:11:26
 0003,15383598920002018-10-01 10:11:32
 0004,15383598930002018-10-01 10:11:33
 0005,15383598940002018-10-01 10:11:34
 0006,15383598960002018-10-01 10:11:36
 0007,15383598970002018-10-01 10:11:37
 
 0008,15383598990002018-10-01 10:11:39
 0009,15383598910002018-10-01 10:11:31
 0010,15383599030002018-10-01 10:11:43
 
 0011,15383598920002018-10-01 10:11:32
 0012,15383598910002018-10-01 10:11:31
 
 0010,15383599060002018-10-01 10:11:46
複製程式碼

第一個窗觸發：2018-10-01 10:11:21.000《----》2018-10-01 10:11:24.000

第二個窗觸發：2018-10-01 10:11:24.000《----》2018-10-01 10:11:27.000

第三個窗觸發：2018-10-01 10:11:30.000《----》2018-10-01 10:11:33.000

第四個窗觸發：10:11:33.000《----》2018-10-01 10:11:36.000

4.2 最霸道的水印設計（allowedLateness與OutputLateData）

在某些情況下，我們希望對遲到的資料再提供一個寬容的時間。 Flink 提供了 allowedLateness 方法可以實現對遲到的資料設定一個延遲時間，在指定延遲時間內到達的資料還是可以觸發 window 執行的。
第二次（或多次）觸發的條件是 watermark < window_end_time + allowedLateness 時間內，這個視窗有 late 資料到達時。
舉例：當 watermark 等於 10:11:34 的時候，我們輸入 eventtime 為 10:11:30、 10:11:31、10:11:32 的資料的時候，是可以觸發的，因為這些資料的 window_end_time 都是 10:11:33，也就是10:11:34<10:11:33+2 為 true。
舉例：但是當 watermark 等於 10:11:35 的時候，我們再輸入 eventtime 為 10:11:30、10:11:31、10:11:32的資料的時候，這些資料的 window_end_time 都是 10:11:33，此時， 10:11:35< 10:11:33+2 為false 了。所以最終這些資料遲到的時間太久了，就不會再觸發 window 執行了，預示著丟棄。

同時注意，對於延遲的資料，我們完全可以把它揪出來作分析。通過設定sideOutputLateData。

public class StreamingWindowWatermark2 {
public static void main(String[] args) throws Exception {
//定義socket的埠號
int port = 9000;
//獲取執行環境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//設定使用eventtime，預設是使用processtime
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

//設定並行度為1,預設並行度是當前機器的cpu數量
env.setParallelism(1);

//連線socket獲取輸入的資料
DataStream<String> text = env.socketTextStream("hadoop100", port, "\n");

//解析輸入的資料
DataStream<Tuple2<String, Long>> inputMap = text.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String value) throws Exception {
String[] arr = value.split(",");
return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
}
});

//抽取timestamp和生成watermark
DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {

Long currentMaxTimestamp = 0L;
final Long maxOutOfOrderness = 10000L;// 最大允許的亂序時間是10s

SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
/**
* 定義生成watermark的邏輯
* 預設100ms被呼叫一次
*/
@Nullable
@Override
public Watermark getCurrentWatermark() {
return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
}

//定義如何提取timestamp
@Override
public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
long timestamp = element.f1;
currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
System.out.println("key:"+element.f0+",eventtime:["+element.f1+"|"+sdf.format(element.f1)+"],currentMaxTimestamp:["+currentMaxTimestamp+"|"+
sdf.format(currentMaxTimestamp)+"],watermark:["+getCurrentWatermark().getTimestamp()+"|"+sdf.format(getCurrentWatermark().getTimestamp())+"]");
return timestamp;
}
});

//儲存被丟棄的資料
OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("late-data"){};
//注意，由於getSideOutput方法是SingleOutputStreamOperator子類中的特有方法，所以這裡的型別，不能使用它的父類dataStream。
SingleOutputStreamOperator<String> window = waterMarkStream.keyBy(0)
.window(TumblingEventTimeWindows.of(Time.seconds(3)))//按照訊息的EventTime分配視窗，和呼叫TimeWindow效果一樣
//.allowedLateness(Time.seconds(2))//允許資料遲到2秒
.sideOutputLateData(outputTag)
.apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
/**
* 對window內的資料進行排序，保證資料的順序
* @param tuple
* @param window
* @param input
* @param out
* @throws Exception
*/
@Override
public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
String key = tuple.toString();
List<Long> arrarList = new ArrayList<Long>();
Iterator<Tuple2<String, Long>> it = input.iterator();
while (it.hasNext()) {
Tuple2<String, Long> next = it.next();
arrarList.add(next.f1);
}
Collections.sort(arrarList);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
String result = key + "," + arrarList.size() + "," + sdf.format(arrarList.get(0)) + "," + sdf.format(arrarList.get(arrarList.size() - 1))
+ "," + sdf.format(window.getStart()) + "," + sdf.format(window.getEnd());
out.collect(result);
}
});
//把遲到的資料暫時列印到控制檯，實際中可以儲存到其他儲存介質中
DataStream<Tuple2<String, Long>> sideOutput = window.getSideOutput(outputTag);
sideOutput.print();
//測試-把結果列印到控制檯即可
window.print();

//注意：因為flink是懶載入的，所以必須呼叫execute方法，上面的程式碼才會執行
env.execute("eventtime-watermark");
}
}
複製程式碼

4.3 多並行度下的 watermark觸發機制

4.3.1 先領會程式碼（感謝 github xuwei）

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import javax.annotation.Nullable;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

public class StreamingWindowWatermark2 {

public static void main(String[] args) throws Exception {
//定義socket的埠號
int port = 9010;
//獲取執行環境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//設定使用eventtime，預設是使用processtime
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

//設定並行度為1,預設並行度是當前機器的cpu數量
env.setParallelism(8);

//連線socket獲取輸入的資料
DataStream<String> text = env.socketTextStream("SparkMaster", port, "\n");

//解析輸入的資料
DataStream<Tuple2<String, Long>> inputMap = text.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String value) throws Exception {
String[] arr = value.split(",");
return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
}
});

//抽取timestamp和生成watermark
DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {

Long currentMaxTimestamp = 0L;
final Long maxOutOfOrderness = 10000L;// 最大允許的亂序時間是10s

SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
/**
* 定義生成watermark的邏輯
* 預設100ms被呼叫一次
*/
@Nullable
@Override
public Watermark getCurrentWatermark() {
return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
}

//定義如何提取timestamp
@Override
public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
long timestamp = element.f1;
currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
long id = Thread.currentThread().getId();
System.out.println("作者：秦凱新 鍵值 :"+element.f0+"執行緒驗證 ："+id+" , 事件事件:[ "+sdf.format(element.f1)+" ],currentMaxTimestamp:[ "+
sdf.format(currentMaxTimestamp)+" ],水印時間:[ "+sdf.format(getCurrentWatermark().getTimestamp())+" ]");return timestamp;
}
});

//儲存被丟棄的資料
OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("late-data"){};
//注意，由於getSideOutput方法是SingleOutputStreamOperator子類中的特有方法，所以這裡的型別，不能使用它的父類dataStream。
SingleOutputStreamOperator<String> window = waterMarkStream.keyBy(0)
.window(TumblingEventTimeWindows.of(Time.seconds(3)))//按照訊息的EventTime分配視窗，和呼叫TimeWindow效果一樣
//.allowedLateness(Time.seconds(2))//允許資料遲到2秒
.sideOutputLateData(outputTag)
.apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
/**
* 對window內的資料進行排序，保證資料的順序
* @param tuple
* @param window
* @param input
* @param out
* @throws Exception
*/
@Override
public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
String key = tuple.toString();
List<Long> arrarList = new ArrayList<Long>();
Iterator<Tuple2<String, Long>> it = input.iterator();
while (it.hasNext()) {
Tuple2<String, Long> next = it.next();
arrarList.add(next.f1);
}
Collections.sort(arrarList);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
String result = "\n 作者：秦凱新 鍵值 : "+ key + "\n觸發窗內資料個數 : " + arrarList.size() + "\n觸發窗起始資料： " + sdf.format(arrarList.get(0)) + "\n觸發窗最後（可能是延時）資料：" + sdf.format(arrarList.get(arrarList.size() - 1))
+ "\n實際窗起始和結束時間： " + sdf.format(window.getStart()) + "《----》" + sdf.format(window.getEnd()) + " \n \n ";
out.collect(result);
}
});
//把遲到的資料暫時列印到控制檯，實際中可以儲存到其他儲存介質中
DataStream<Tuple2<String, Long>> sideOutput = window.getSideOutput(outputTag);
sideOutput.print();
//測試-把結果列印到控制檯即可
window.print();

//注意：因為flink是懶載入的，所以必須呼叫execute方法，上面的程式碼才會執行
env.execute("eventtime-watermark");

}
}
複製程式碼

4.3.2 前面程式碼中設定了並行度為 1：

env.setParallelism(1);
複製程式碼

如果這裡不設定的話，程式碼在執行的時候會預設讀取本機 CPU 數量設定並行度。

下面我們來驗證一下，把程式碼中的並行度調整為 2：

env.setParallelism(2);
複製程式碼

發現玄機如下：在第二條事件時，其實已經達到窗的觸發時機，但是因為並行度為2，只有等到最小
watermark 到的時候才會觸發窗計算。發現執行緒44處理的是001和003 ，執行緒42處理的是0002，所以只有等到執行緒42到達後，水印才會起作用執行2018-10-01 10:11:33.000所在的窗。
```
0001,15383598900002018-10-01 10:11:30
0002,15383599030002018-10-01 10:11:43
0003,15383599080002018-10-01 10:11:48
複製程式碼
```

4.3.3 現在程式碼中設定了並行度為 8：

發現這 7 條資料都是被不同的執行緒處理的。每個執行緒都有一個 watermark。且每一個執行緒都是基於自己接收資料的事件時間最大值。
因此，導致到最後現在還沒獲取到最小的 watermark，所以 window 無法被觸發執行。

只有所有的執行緒的最小watermark都滿足watermark 時間 >= window_end_time時，觸發歷史窗才會執行。

0001,15383598820002018-10-01 10:11:22
0002,15383598860002018-10-01 10:11:26
0003,15383598920002018-10-01 10:11:32
0004,15383598930002018-10-01 10:11:33
0005,15383598940002018-10-01 10:11:34
0006,15383598960002018-10-01 10:11:36
0007,15383598970002018-10-01 10:11:37
複製程式碼

當持續發生事件資料時。一旦所有執行緒都達到最低的窗觸發時機時，就會進行窗觸發執行了。輸入資料如下：

0007,15383598970002018-10-01 10:11:37
0008,15383598970002018-10-01 10:11:37
0009,15383598970002018-10-01 10:11:37
0010,15383598970002018-10-01 10:11:37
0011,15383598970002018-10-01 10:11:37
0012,15383598970002018-10-01 10:11:37
0013,15383598970002018-10-01 10:11:37
0014,15383598970002018-10-01 10:11:37
0015,15383598970002018-10-01 10:11:37
複製程式碼

5 再下一城

截止到201811250141時間點，我的個人技術部落格已經涵蓋：

-《Spark Core 商業原始碼實戰系列目錄》
-《SparkStreaming 商業原始碼實戰系列目錄》
-《SparkSQL 商業原始碼實戰系列目錄》
-《Spark商業應用實戰系列目錄》
-《Spark商業調優實戰系列目錄》
-《Spark商業ML實戰系列目錄》
-《Flink 牛刀小試實戰系列目錄》
-《Hadoop商業環境實戰系列目錄》
-《kafka 商業環境實戰系列目錄》
-《OLAP商業環境實戰系列目錄》
-《DW商業環境實戰系列目錄》
複製程式碼

部落格原創作品達到64篇高質量的精品，在這裡給自己的堅持加油，送上一句豪情萬丈。

秦凱新於深圳