Flink EventTime和Watermarks案例分析

迈不过友情╰ 2022-05-04 01:26 209阅读 0赞

**目录**

解释：

1：实现watermark相关代码

1.1：程序说明

1.2：代码如下

1.3：程序详解

2：通过数据跟踪watermark的时间

3：watermark+window处理乱序数据

4：late element(延迟数据)的处理

4.1：丢弃(默认)

4.2：allowedLateness 指定允许数据延迟的时间

4.3：sideOutputLateData 收集迟到的数据

--------------------

针对数据乱序需求的案例分析，需要使用eventtime和watermark来解决

# **解释：** #

watermarks的生成方式有两种

1：With Periodic Watermarks：周期性的触发watermark的生成和发送

2：With Punctuated Watermarks：基于某些事件触发watermark的生成和发送

第一种方式比较常用，所以在这里我们使用第一种方式进行分析。

参考官网文档中With Periodic Watermarks的使用方法

![70][]

代码中的extractTimestamp方法是从数据本身中提取EventTime

getCurrentWatermar方法是获取当前水位线，利用currentMaxTimestamp - maxOutOfOrderness

这里的maxOutOfOrderness表示是允许数据的最大乱序时间

所以在这里我们使用的话也实现接口AssignerWithPeriodicWatermarks。

![70 1][]

# **1：实现watermark相关代码** #

## **1.1：程序说明** ##

从socket模拟接收数据，然后使用map进行处理，后面再调用assignTimestampsAndWatermarks方法抽取timestamp并生成watermark。最后再调用window打印信息来验证window被触发的时机。

## **1.2：代码如下** ##

**![70 2][]**

![70 3][]

![70 4][]

![70 5][]

完整代码如下：

package xuwei.tech.streaming.streamApiDemo;
 
 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.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 javax.annotation.Nullable;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
 
 /**
 *
 * Watermark 案例
 *
 * Created by xuwei.tech.
 */
 public class StreamingWindowWatermark {
 
 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;
 }
 });
 
 //分组，聚合
 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 = 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);
 }
 });
 //测试-把结果打印到控制台即可
 window.print();
 
 //注意：因为flink是懒加载的，所以必须调用execute方法，上面的代码才会执行
 env.execute("eventtime-watermark");
 
 }
 
 
 
 }

## **1.3：程序详解** ##

1. 接收socket数据
2. 将每行数据按照逗号分隔，每行数据调用map转换成tuple<String,Long>类型。其中tuple中的第一个元素代表具体的数据，第二行代表数据的eventtime
3. 抽取timestamp，生成watermar，允许的最大乱序时间是10s，并打印（key,eventtime,currentMaxTimestamp,watermark）等信息
4. 分组聚合，window窗口大小为3秒，输出（key，窗口内元素个数，窗口内最早元素的时间，窗口内最晚元素的时间，窗口自身开始时间，窗口自身结束时间）

# **2：通过数据跟踪watermark的时间** #

在这里重点查看watermark和timestamp的时间，通过数据的输出来确定window的触发时机

首先我们开启socker，输入第一条数据

输出的内容如下：

![70 6][]

为了查看方便，我们把输入内容汇总到表格中

此时，wartermark的时间，已经落后于currentMaxTimestamp10秒了。我们继续输入

![70 7][]

此时，输入内容如下：

![70 8][]

我们再次汇总，如下表：

继续输入：

![70 9][]

输出内容如下：

![70 10][]

汇总如下：

<table style="width:426.1pt;"> 
 <tbody> 
 <tr> 
 <td style="width:65.4pt;"> Key </td> 
 <td style="width:118.85pt;"> Event Time </td> 
 <td style="width:122.3pt;"> CurrentMaxTimeStamp </td> 
 <td style="width:119.55pt;"> WaterMark </td> 
 </tr> 
 <tr> 
 <td style="width:65.4pt;"> 0001 </td> 
 <td style="vertical-align:top;width:118.85pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 1538359872000 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:118.85pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 2018-10-01 10:11:12.000 </td> 
 </tr> 
 <tr> 
 <td style="width:65.4pt;"> 0001 </td> 
 <td style="vertical-align:top;width:118.85pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 1538359876000 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:118.85pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 2018-10-01 10:11:16.000 </td> 
 </tr> 
 <tr> 
 <td style="width:65.4pt;"> 0001 </td> 
 <td style="vertical-align:top;width:118.85pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 1538359882000 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:118.85pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 2018-10-01 10:11:22.000 </td> 
 </tr> 
 </tbody> 
</table>

到这里，window仍然没有被触发，此时watermark的时间已经等于了第一条数据的Event Time了。那么window到底什么时候被触发呢？

我们再次输入：

![70 11][]

输出内容如下：

![70 12][]

汇总如下：

<table style="width:426.1pt;"> 
 <tbody> 
 <tr> 
 <td style="width:65.4pt;"> Key </td> 
 <td style="width:118.85pt;"> Event Time </td> 
 <td style="width:122.3pt;"> CurrentMaxTimeStamp </td> 
 <td style="width:119.55pt;"> WaterMark </td> 
 </tr> 
 <tr> 
 <td style="width:65.4pt;"> 0001 </td> 
 <td style="vertical-align:top;width:118.85pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 1538359872000 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:118.85pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 2018-10-01 10:11:12.000 </td> 
 </tr> 
 <tr> 
 <td style="width:65.4pt;"> 0001 </td> 
 <td style="vertical-align:top;width:118.85pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 1538359876000 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:118.85pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 2018-10-01 10:11:16.000 </td> 
 </tr> 
 <tr> 
 <td style="width:65.4pt;"> 0001 </td> 
 <td style="vertical-align:top;width:118.85pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 1538359882000 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:118.85pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 2018-10-01 10:11:22.000 </td> 
 </tr> 
 <tr> 
 <td style="width:65.4pt;"> 0001 </td> 
 <td style="vertical-align:top;width:118.85pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 1538359883000 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:118.85pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:122.3pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:119.55pt;"> 2018-10-01 10:11:23.000 </td> 
 </tr> 
 </tbody> 
</table>

window仍然没有触发，此时，我们的数据已经发到2018-10-01 10:11:33.000了，根据eventtime来算，最早的数据已经过去了11秒了，window还没有开始计算，那到底什么时候会触发window呢？

我们再次增加1秒，输入：

![70 13][]

输出：

![70 14][]

汇总如下：

<table style="width:426.05pt;"> 
 <tbody> 
 <tr> 
 <td style="width:34.35pt;"> Key </td> 
 <td style="width:81.25pt;"> Event Time </td> 
 <td style="width:81.9pt;"> CurrentMaxTimeStamp </td> 
 <td style="width:83.1pt;"> WaterMark </td> 
 <td style="width:71.9pt;"> window_start_time </td> 
 <td style="width:73.55pt;"> window_end_time </td> 
 </tr> 
 <tr> 
 <td style="width:34.35pt;"> 0001 </td> 
 <td style="vertical-align:top;width:81.25pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 1538359872000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:73.55pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:81.25pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 2018-10-01 10:11:12.000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:73.55pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:34.35pt;"> 0001 </td> 
 <td style="vertical-align:top;width:81.25pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 1538359876000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:73.55pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:81.25pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 2018-10-01 10:11:16.000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:73.55pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:34.35pt;"> 0001 </td> 
 <td style="vertical-align:top;width:81.25pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:73.55pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:81.25pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:73.55pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:34.35pt;"> 0001 </td> 
 <td style="vertical-align:top;width:81.25pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 1538359883000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:73.55pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:81.25pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 2018-10-01 10:11:23.000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:73.55pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:34.35pt;"> 0001 </td> 
 <td style="vertical-align:top;width:81.25pt;"> 1538359894000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 1538359894000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 1538359884000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:73.55pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:81.25pt;"> 2018-10-01 10:11:34.000 </td> 
 <td style="vertical-align:top;width:81.9pt;"> 2018-10-01 10:11:34.000 </td> 
 <td style="vertical-align:top;width:83.1pt;"> 2018-10-01 10:11:24.000 </td> 
 <td style="vertical-align:top;width:71.9pt;"> [10:11:21.000 </td> 
 <td style="vertical-align:top;width:73.55pt;"> 10:11:24.000) </td> 
 </tr> 
 </tbody> 
</table>

到这里，我们做一个说明：   
window的触发机制，是先按照自然时间将window划分，如果window大小是3秒，那么1分钟内会把window划分为如下的形式【左闭右开】:

[00:00:00,00:00:03)
    [00:00:03,00:00:06)
    [00:00:06,00:00:09)
    [00:00:09,00:00:12)
    [00:00:12,00:00:15)
    [00:00:15,00:00:18)
    [00:00:18,00:00:21)
    [00:00:21,00:00:24)
    [00:00:24,00:00:27)
    [00:00:27,00:00:30)
    [00:00:30,00:00:33)
    [00:00:33,00:00:36)
    [00:00:36,00:00:39)
    [00:00:39,00:00:42)
    [00:00:42,00:00:45)
    [00:00:45,00:00:48)
    [00:00:48,00:00:51)
    [00:00:51,00:00:54)
    [00:00:54,00:00:57)
    [00:00:57,00:01:00)
    ...

window的设定无关数据本身，而是系统定义好了的。

输入的数据中，根据自身的Event Time，将数据划分到不同的window中，如果window中有数据，则当watermark时间>=Event Time时，就符合了window触发的条件了，最终决定window触发，还是由数据本身的Event Time所属的window中的window\_end\_time决定。

上面的测试中，最后一条数据到达后，其水位线已经升至10:11:24秒，正好是最早的一条记录所在window的window\_end\_time，所以window就被触发了。

为了验证window的触发机制，我们继续输入数据:

![70 15][]

输出：

![70 16][]

汇总如下：

此时，watermark时间虽然已经达到了第二条数据的时间，但是由于其没有达到第二条数据所在window的结束时间，所以window并没有被触发。那么，第二条数据所在的window时间是:

[00:00:24,00:00:27)

也就是说，我们必须输入一个10:11:27秒的数据，第二条数据所在的window才会被触发。我们继续输入：

![70 17][]

输出：

![70 18][]

汇总如下：

<table style="width:426.05pt;"> 
 <tbody> 
 <tr> 
 <td style="width:32.5pt;"> Key </td> 
 <td style="width:80pt;"> Event Time </td> 
 <td style="width:80.6pt;"> CurrentMaxTimeStamp </td> 
 <td style="width:80.25pt;"> WaterMark </td> 
 <td style="width:76.35pt;"> window_start_time </td> 
 <td style="width:76.35pt;"> window_end_time </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359872000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:12.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359876000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:16.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359883000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:23.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359894000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359894000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359884000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:34.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:34.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:24.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> [10:11:21.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> 10:11:24.000) </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359896000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359896000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:36.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:36.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359897000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359897000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359887000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:37.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:37.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:27.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> [10:11:24.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> 10:11:27.000) </td> 
 </tr> 
 </tbody> 
</table>

此时，我们已经看到，window的触发要符合以下几个条件：

1、watermark时间 >= window_end_time 
    2、在[window_start_time,window_end_time)区间中有数据存在，注意是左闭右开的区间

同时满足了以上2个条件，window才会触发。

# **3：watermark+window处理乱序数据** #

我们上面的测试，数据都是按照时间顺序递增的，现在，我们输入一些乱序的（late）数据，看看watermark结合window机制，是如何处理乱序的。

输入两行数据：

![70 19][]

输出：

![70 20][]

汇总如下：

<table style="width:426.05pt;"> 
 <tbody> 
 <tr> 
 <td style="width:32.5pt;"> Key </td> 
 <td style="width:80pt;"> Event Time </td> 
 <td style="width:80.6pt;"> CurrentMaxTimeStamp </td> 
 <td style="width:80.25pt;"> WaterMark </td> 
 <td style="width:76.35pt;"> window_start_time </td> 
 <td style="width:76.35pt;"> window_end_time </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359872000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:12.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359876000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:16.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359883000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:23.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359894000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359894000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359884000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:34.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:34.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:24.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> [10:11:21.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> 10:11:24.000) </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359896000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359896000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:36.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:36.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359897000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359897000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359887000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:37.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:37.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:27.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> [10:11:24.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> 10:11:27.000) </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359899000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359899000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359889000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:39.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:39.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:29.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359891000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359899000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359889000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:31.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:39.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:29.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 </tbody> 
</table>

可以看到，虽然我们输入了一个10:11:31的数据，但是currentMaxTimestamp和watermark都没变。此时，按照我们上面提到的公式：

1、watermark时间 >= window_end_time
    2、在[window_start_time,window_end_time)中有数据存在

watermark时间（10:11:29） < window\_end\_time（10:11:33），因此不能触发window。

那如果我们再次输入一条10:11:43的数据，此时watermark时间会升高到10:11:33，这时的window一定就会触发了，我们试一试：   
输入：

![70 21][]

输出：

![70 22][]

汇总如下：

<table style="width:426.05pt;"> 
 <tbody> 
 <tr> 
 <td style="width:32.5pt;"> Key </td> 
 <td style="width:80pt;"> Event Time </td> 
 <td style="width:80.6pt;"> CurrentMaxTimeStamp </td> 
 <td style="width:80.25pt;"> WaterMark </td> 
 <td style="width:76.35pt;"> window_start_time </td> 
 <td style="width:76.35pt;"> window_end_time </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359872000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:12.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359876000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:16.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359892000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359882000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:32.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:22.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359883000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:23.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359894000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359894000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359884000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:34.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:34.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:24.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> [10:11:21.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> 10:11:24.000) </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359896000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359896000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359886000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:36.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:36.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:26.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359897000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359897000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359887000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:37.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:37.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:27.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> [10:11:24.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> 10:11:27.000) </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359899000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359899000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359889000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:39.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:39.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:29.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359891000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359899000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359889000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:31.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:39.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:29.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:32.5pt;"> 0001 </td> 
 <td style="vertical-align:top;width:80pt;"> 1538359903000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 1538359903000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 1538359893000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:76.35pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:80pt;"> 2018-10-01 10:11:43.000 </td> 
 <td style="vertical-align:top;width:80.6pt;"> 2018-10-01 10:11:43.000 </td> 
 <td style="vertical-align:top;width:80.25pt;"> 2018-10-01 10:11:33.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> [10:11:30.000 </td> 
 <td style="vertical-align:top;width:76.35pt;"> 10:11:33.000) </td> 
 </tr> 
 </tbody> 
</table>

这里，我么看到，窗口中有2个数据，10:11:31和10:11:32，但是没有10:11:33的数据，原因是窗口是一个前闭后开的区间，10:11:33的数据是属于\[10:11:33,10:11:36)的窗口的。

Flink应该如何设置最大乱序时间 ?

这个要结合自己的业务以及数据情况去设置。如果maxOutOfOrderness设置的太小，而自身数据发送时由于网络等原因导致乱序或者late太多，那么最终的结果就是会有很多单条的数据在window中被触发，数据的正确性影响太大 
    对于严重乱序的数据，需要严格统计数据最大延迟时间，才能保证计算的数据准确，延时设置太小会影响数据准确性，延时设置太大不仅影响数据的实时性，更加会加重Flink作业的负担，不是对eventTime要求特别严格的数据，尽量不要采用eventTime方式来处理，会有丢数据的风险。

上边的结果，已经表明，对于out-of-order的数据，Flink可以通过watermark机制结合window的操作，来处理一定范围内的乱序数据。那么对于“迟到(late element)”太多的数据，Flink是怎么处理的呢？

# **4：late element(延迟数据)的处理** #

延迟数据的三种处理方案

## **4.1：丢弃(默认)** ##

我们输入一个乱序很多的（其实只要Event Time < watermark时间）数据来测试下：

输入：【输入两条内容】

![70 23][]

输出：

![70 24][]

汇总如下：

注意：此时watermark是2018-10-01 10:11:33.000

下面我们再输入几个eventtime小于watermark的时间

输入：【输入了三行内容】

![70 25][]

输出：

![70 26][]

注意：此时并没有触发window。因为输入的数据所在的窗口已经执行过了，flink默认对这些迟到的数据的处理方案就是丢弃。

## **4.2：allowedLateness 指定允许数据延迟的时间** ##

在某些情况下，我们希望对迟到的数据再提供一个宽容的时间。

Flink提供了allowedLateness方法可以实现对迟到的数据设置一个延迟时间，在指定延迟时间内到达的数据还是可以触发window执行的。

修改代码：

![70 27][]

下面我们来验证一下：

输入：【输入两行内容】

![70 28][]

输出：

![70 29][]

正常触发window，没什么问题。

汇总：

此时watermark是2018-10-01 10:11:33.000

那么现在我们输入几条eventtime<watermark的数据验证一下效果

输入：【输入三行内容】

![70 30][]

输出：

![70 31][]

在这里看到每条数据都触发了window执行。

汇总：

我们再输入一条数据，把water调整到10:11:34

输入：

![70 32][]

输出：

![70 33][]

汇总如下：

此时，把water上升到了10:11:34，我们再输入几条eventtime<watermark的数据验证一下效果

输入：

![70 34][]

输出：

![70 35][]

发现输入的三行数据都触发了window的执行。

我们再输入一条数据，把water调整到10:11:35

输入：

![70 36][]

输出：

![70 37][]

此时，watermark上升到了10:11:35

我们再输入几条eventtime<watermark的数据验证一下效果

输入：

![70 38][]

输出：

![70 39][]

发现这几条数据都没有触发window。

分析：

当watemark等于10:11:33的时候，正好是window\_end\_time，所以会触发\[10:11:30~10:11:33) 的window执行。

当窗口执行过后，我们输入\[10:11:30~10:11:33) window内的数据会发现window是可以被触发的。

当watemark提升到10:11:34的时候，我们输入\[10:11:30~10:11:33)window内的数据会发现window也是可以被触发的。

当watemark提升到10:11:35的时候，我们输入\[10:11:30~10:11:33)window内的数据会发现window不会被触发了。

由于我们在前面设置了allowedLateness(Time.seconds(2))，可以允许延迟在2s内的数据继续触发window执行。

所以当watermark是10:11:34的时候可以触发window，但是10:11:35的时候就不行了。

总结：

对于此窗口而言，允许2秒的迟到数据，即第一次触发是在watermark >=window\_end\_time时

第二次（或多次）触发的条件是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执行了。

## **4.3：sideOutputLateData 收集迟到的数据** ##

通过sideOutputLateData 可以把迟到的数据统一收集，统计存储，方便后期排查问题。

需要先调整代码：

![70 40][]

我们来输入一些数据验证一下

输入：

![70 41][]

输出：  
![70 42][]

此时，window被触发执行了，此时watermark是10:11:33

下面我们再输入几个eventtime小于watermark的数据测试一下

输入：

![70 43][]

输出：

![70 44][]

此时，针对这几条迟到的数据，都通过sideOutputLateData保存到了outputTag中。

更多资源和问题请加群沟通学习：

**获取更多大数据资料，视频以及技术交流请加群：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1NDcwNDM4MDAw_size_16_color_FFFFFF_t_70][]

[70]: /images/20220504/f97cf662b3b645cd881c1ff176fde212.png
[70 1]: /images/20220504/2f50c6e18eee425cabde9ff68cb4cb9e.png
[70 2]: /images/20220504/2f0d81c134fd4a33ba8201ae80c70633.png
[70 3]: /images/20220504/62e78027c193409fb3b862c7096e48b9.png
[70 4]: /images/20220504/4ef3d29014eb44d6b62bf4ee472d3137.png
[70 5]: /images/20220504/1fe73f1f33b14b5f8372dc6036e5f494.png
[70 6]: /images/20220504/8463b5b89d84498b8fda4a05c0d886be.png
[70 7]: /images/20220504/a13bfb6818cd4a41b23d6401955ead34.png
[70 8]: /images/20220504/60aefe8251b3430a921c1b2055e1ed9e.png
[70 9]: /images/20220504/dca23ade1437486dab5299f426166a22.png
[70 10]: /images/20220504/7f76e4de264e435cb9630e5a2a72bb07.png
[70 11]: /images/20220504/61cea5608a8947899f7d7dc398bf6ff9.png
[70 12]: /images/20220504/cf2f13daef2642f8802e982120f53ee5.png
[70 13]: /images/20220504/47ac5722d7f84545a879f8e341ab59c8.png
[70 14]: /images/20220504/e6246037659e4d3fbf35d326b38ba346.png
[70 15]: /images/20220504/400d32a4298040319ad5fe7c56aaeba0.png
[70 16]: /images/20220504/8dcfb2a7bbd3489791e09c9633a72cc1.png
[70 17]: /images/20220504/403a10cf030e447e995414e408d25c3d.png
[70 18]: /images/20220504/83462cf837f34492b04c250ae0da3a7b.png
[70 19]: /images/20220504/f52ffb21fd9349cdb581ff3970a27a34.png
[70 20]: /images/20220504/9d70985a650744c2a51d14e8b20a0708.png
[70 21]: /images/20220504/0e38027d638143e5a16e6d4ed3339fdf.png
[70 22]: /images/20220504/0c5d5934c490460996dc8013044a314e.png
[70 23]: /images/20220504/c525f10fc8e84212b4e1b9a23fb626b9.png
[70 24]: /images/20220504/1528b3990816470b98d0a4feac371800.png
[70 25]: /images/20220504/5998fd1e859849c4a002d7d90293e68f.png
[70 26]: /images/20220504/72b95904bb2649cf87548ecba90e3abf.png
[70 27]: /images/20220504/b6cef5ee7ae74e7cbc91aa34245011d4.png
[70 28]: /images/20220504/cff048bdb8884f99a863ed6abcadb150.png
[70 29]: /images/20220504/a116f3de57b1401c9d5de7c742b0d004.png
[70 30]: /images/20220504/656f5c30ec6946478574a1f87078358e.png
[70 31]: /images/20220504/f790c9e127b54f3ab3ea7c88ce40ec9d.png
[70 32]: /images/20220504/f9c243a08b0b4ae5adf17a19b2ce0153.png
[70 33]: /images/20220504/a29ce06a0bd04959aba1910709582686.png
[70 34]: /images/20220504/0ae5fbfd691c439d85d80edbe17a1d14.png
[70 35]: /images/20220504/18ab5205c626482baea153d1b610c6b7.png
[70 36]: /images/20220504/8df8953cb56c4bdaa51ae439a167bb41.png
[70 37]: /images/20220504/2db287c419f84e73b3010d38c5dec332.png
[70 38]: /images/20220504/eeb1c52c75fd4c9386658f0a1ad5eae3.png
[70 39]: /images/20220504/54412052c57045e1b31369ae85465207.png
[70 40]: /images/20220504/875b3d84c0cf4d3c95d8acd3cbfeb54c.png
[70 41]: /images/20220504/5f81d3c6fb1b41d9afed79f8ab9dda62.png
[70 42]: /images/20220504/fb6c9fc211a54a8194f4ca9b69a81eb3.png
[70 43]: /images/20220504/af63830527054892863138e9766661e7.png
[70 44]: /images/20220504/774a2efb21fb4983b7d990b4f7b989c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1NDcwNDM4MDAw_size_16_color_FFFFFF_t_70]: /images/20220504/eb2f35f8b38a4cf1889b66bf54d78235.png