我們先來以滾動時間視窗為例，來看一下視窗的幾個時間引數與Flink流處理系統時間特性的關系，
獲取視窗開始時間Flink源代碼
獲取視窗的開始時間為以下代碼：
org.apache.flink.streaming.api.windowing.windows.TimeWindow
/**
* Method to get the window start for a timestamp.
*
* @param timestamp epoch millisecond to get the window start.
* @param offset The offset which window start would be shifted by.
* @param windowSize The size of the generated windows.
* @return window start
*/
public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) { return timestamp - (timestamp - offset + windowSize) % windowSize; }
這一段代碼，我們可以認為Flink并不是把時間戳直接作為視窗的開始時間，而是做了一些“對齊”操作，確保時間能夠整除8，
不同時間型別的視窗時間計算
1、當TimeCharacteristic為ProcessingTime時
視窗的開始時間：與視窗接收到的第一條訊息的處理時間有關，例如：window operator是2020-02-06 22:02:33接收到的第一條訊息，那么視窗的開始時間就是2020-02-06 22:02:33，
視窗的結束時間：一旦視窗的開始時間確定了，因為視窗的長度是固定的，那么視窗的結束時間就確定下來了，例如：假設這里的時間視窗是3秒，那么視窗的結束時間就是2020-02-06 22:02:36，
視窗的觸發計算時間：假設有一條新的訊息到達window operator，此時如果對應operator的系統時間，大于結束時間，就會觸發計算，
一旦視窗的開始時間確定了，那么后續視窗的開始時間，也就都確定下來了，
問題：
假設某個時間視窗，2020-2-6 22:12:20 - 2020-2-6 22:12:23，之間沒有任何一條資料進來，Flink會如何處理？
Flink會直接拋棄掉這個時間視窗，新來的事件訊息會到其他的時間視窗中計算，
2、當TimeCharacteristic為IngestionTime時
視窗的開始時間：與source operator接收到的第一條訊息有關，例如：source接收到這條訊息的時間是2020-2-6 22:14:50，那么視窗的開始時間就是2020-2-6 22:14:50
視窗的結束時間：與ProcessTime一致
視窗的觸發計算時間：假設有一條新的訊息到達source operator，那么此時的時間如果大于結束時間，就會觸發計算，
除了視窗的開始時間、觸發時間都是與source operator算子有關，其他與Processing Time是類似的，
3、但TimeCharacteristic為EventTime時
視窗的開始時間：與window operator接收到的第一條訊息的事件時間有關，例如：如果這條訊息的水印時間是2020-2-6 22:17:50，那么視窗的的開始時間就是2020-2-6 22:17:50
視窗的結束時間：與ProcessTime一致
視窗的觸發計算時間：假設有一條新的訊息到達window operator，如果該事件的水印時間大于視窗的結束時間，就會觸發計算，
通常，我們會讓水印時間比事件時間允許延遲幾秒鐘，這樣，如果是因為網路延遲訊息晚到了幾秒，也不會影響到統計結果了，
public class WordCountWindow { public static void main(String[] args) throws Exception { // 1. 初始化流式運行環境 Configuration conf = new Configuration(); StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf); // 2. 設定時間處理型別，這里設定的方式處理時間 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); // 3. 定義資料源，每秒發送一個hadoop單詞 SingleOutputStreamOperator<Tuple2<String, Long>> wordDSWithWaterMark = env.addSource(new RichSourceFunction<Tuple2<String, Long>>() { private boolean isCanaled = false; private int TOTAL_NUM = 20; @Override public void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception { while (!isCanaled) { ctx.collect(Tuple2.of("hadooop", System.currentTimeMillis())); // 列印視窗開始、結束時間 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); System.out.println("事件發送時間:" + sdf.format(System.currentTimeMillis())); Thread.sleep(1000); } } @Override public void cancel() { isCanaled = true; } }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<String, Long>>(Time.seconds(5)) { @Override public long extractTimestamp(Tuple2<String, Long> element) { return element.f1; } }); // 4. 每5秒進行一次，分組統計 // 4.1 轉換為元組 wordDSWithWaterMark.map(word -> { return Tuple2.of(word.f0, 1); }) // 指定回傳型別 .returns(Types.TUPLE(Types.STRING, Types.INT)) // 按照單詞進行分組 .keyBy(t -> t.f0) // 滾動視窗，3秒計算一次 .timeWindow(Time.seconds(3)) .reduce(new ReduceFunction<Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception { return Tuple2.of(value1.f0, value1.f1 + value2.f1); } }, new RichWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>() { @Override public void apply(String word, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<Tuple2<String, Integer>> out) throws Exception { // 列印視窗開始、結束時間 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); System.out.println("視窗開始時間:" + sdf.format(window.getStart()) + " 視窗結束時間：" + sdf.format(window.getEnd()) + " 視窗計算時間:" + sdf.format(System.currentTimeMillis())); int sum = 0; Iterator<Tuple2<String, Integer>> iterator = input.iterator(); while(iterator.hasNext()) { Integer count = iterator.next().f1; sum += count; } out.collect(Tuple2.of(word, sum)); } }).print(); env.execute("app"); } }
輸出結果如下：
事件發送時間:2020-02-06 22:35:08
事件發送時間:2020-02-06 22:35:09
事件發送時間:2020-02-06 22:35:10
事件發送時間:2020-02-06 22:35:11
事件發送時間:2020-02-06 22:35:12
事件發送時間:2020-02-06 22:35:13
事件發送時間:2020-02-06 22:35:14
視窗開始時間:2020-02-06 22:35:06 視窗結束時間：2020-02-06 22:35:09 視窗計算時間:2020-02-06 22:35:14
4> (hadooop,1)事件發送時間:2020-02-06 22:35:15
事件發送時間:2020-02-06 22:35:16
事件發送時間:2020-02-06 22:35:17
視窗開始時間:2020-02-06 22:35:09 視窗結束時間：2020-02-06 22:35:12 視窗計算時間:2020-02-06 22:35:17
4> (hadooop,3)
參考檔案：
https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/dev/event_time.html
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/32576.html
標籤：大數據
上一篇：01-Flink運行架構