「Flink」事件時間與水印

咱們先來以滾動時間窗口爲例，來看一下窗口的幾個時間參數與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。

不一樣時間類型的窗口時間計算

一、當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會直接拋棄掉這個時間窗口，新來的事件消息會到其餘的時間窗口中計算。

二、當TimeCharacteristic爲IngestionTime時

窗口的開始時間：與source operator接收到的第一條消息有關。例如：source接收到這條消息的時間是2020-2-6 22:14:50，那麼窗口的開始時間就是2020-2-6 22:14:50

窗口的結束時間：與ProcessTime一致

窗口的觸發計算時間：假設有一條新的消息到達source operator，那麼此時的時間若是大於結束時間，就會觸發計算。

除了窗口的開始時間、觸發時間都是與source operator算子有關，其餘與Processing Time是相似的。

三、但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