flink Window的Timestamps/Watermarks和allowedLateness的區別

 Watermartks是經過additional的時間戳來控制窗口激活的時間，allowedLateness來控制窗口的銷燬時間。    

  注： 由於此特性包括官方文檔在1.3～1.5版本均未作改變，因此此處使用1.5版的文檔    

 

在EventTime的狀況下，       

 

1. 一條記錄的事件時間來控制此條記錄屬於哪個窗口，Watermarks來控制這個窗口何時激活。

   

2. 假如一個窗口時間爲00:00:00～00:00:05，Watermarks爲5秒，那麼當flink收到事件事件爲00:00:10秒的數據時，即Watermarks到達00:00:05，激活這個窗口。

 

 

3. Watermarks激活窗口的方式，官方文檔推薦爲複寫AssignerWithPeriodicWatermarks，與咱們當前項目實現方式一致[ https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/event_timestamps_watermarks.html]

 

4. 或者也能夠使用咱們項目中已經用到的在env級別下的config中設置watermark的方式

 

     env.getConfig().setAutoWatermarkInterval(applConfig.getWatermarkInterval());

 

 

當窗口被激活且運行完畢之後，此時這個窗口不必定被銷燬，窗口狀態有可能會被繼續保持，這一點取決於allowedLateness

 

 

[ https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/operators/windows.html#allowed-lateness]

 

 

In a nutshell, a window is created as soon as the first element that should belong to this window arrives, and the window is completely removed when the time (event or processing time) passes its end timestamp plus the user-specified allowed lateness (see Allowed Lateness). Flink guarantees removal only for time-based windows and not for other types, e.g. global windows (see Window Assigners). For example, with an event-time-based windowing strategy that creates non-overlapping (or tumbling) windows every 5 minutes and has an allowed lateness of 1 min, Flink will create a new window for the interval between 12:00 and 12:05 when the first element with a timestamp that falls into this interval arrives, and it will remove it when the watermark passes the 12:06 timestamp.

 

In addition, each window will have a Trigger (see Triggers) and a function (ProcessWindowFunction, ReduceFunction, AggregateFunction or FoldFunction) (see Window Functions) attached to it. The function will contain the computation to be applied to the contents of the window, while the Trigger specifies the conditions under which the window is considered ready for the function to be applied. A triggering policy might be something like 「when the number of elements in the window is more than 4」, or 「when the watermark passes the end of the window」. A trigger can also decide to purge a window’s contents any time between its creation and removal. Purging in this case only refers to the elements in the window, and not the window metadata. This means that new data can still be added to that window.

 

Apart from the above, you can specify an Evictor (see Evictors) which will be able to remove elements from the window after the trigger fires and before and/or after the function is applied.[ https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/operators/windows.html#window-lifecycle]

 

       

1. 假如設置allowedLateness爲60秒，那麼窗口的狀態會一直保持到事件時間爲00:01:05的數據到達，或者若是最後一條數據早於00:01:05秒，則等到最後一條數據到達後再等待此數據於00:01:05的差值時間。    

 

2. 那麼在窗口被銷燬前，能夠經過一些方式再次激活。注意，allowedLateness只能控制窗口銷燬行爲，並不能控制窗口再次激活的行爲，這是獨立的兩部分行爲。

 

3. 官方文檔推薦的方式爲Getting late data as a side output，能夠單獨得到再次被激活的窗口流 https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/operators/windows.html#getting-late-data-as-a-side-output

目前不肯定原始流內是否也包含了再次被激活的窗口數據，待測試，從代碼上看應該也包含在內。    

   已確認，原始流內窗口也會被從新激活一次

 

final OutputTag<T> lateOutputTag = new OutputTag<T>("late-data"){};

DataStream<T> input = ...;

SingleOutputStreamOperator<T> result = input

    .keyBy(<key selector>)

    .window(<window assigner>)

    .allowedLateness(<time>)

    .sideOutputLateData(lateOutputTag)

    .<windowed transformation>(<window function>);

DataStream<T> lateStream = result.getSideOutput(lateOutputTag);

 

 

4. 或者複寫Triggers[ https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/operators/windows.html#triggers]