默认情况下，当watermark通过end-of-window之后，再有之前的数据到达时，这些数据会被删除。
为了避免有些迟到的数据被删除，因此产生了allowedLateness的概念。
简单来讲，allowedLateness就是针对event time而言，对于watermark超过end-of-window之后，还允许有一段时间（也是以event time来衡量）来等待之前的数据到达，以便再次处理这些数据
默认情况下，如果不指定allowedLateness，其值是0，即对于watermark超过end-of-window之后，还有此window的数据到达时，这些数据被删除掉了。
注意：对于trigger是默认的EventTimeTrigger的情况下，allowedLateness会再次触发窗口的计算，而之前触发的数据，会buffer起来，直到watermark超过end-of-window + watermark设置的时间+allowedLateness（）的时间，窗口的数据及元数据信息才会被删除。再次计算就是DataFlow模型中的Accumulating（积累）的情况。
什么情况下数据会被丢弃或者说不会被计算？
两种情况：
a.未设置allowedLateness情况下，某条数据属于某个窗口，但是watermark超过了窗口的结束时间，则该条数据会被丢弃；
b.设置allowedLateness情况下，某条数据属于某个窗口，但是watermark超过了窗口的结束时间+延迟时间，则该条数据会被丢弃；
也就是说如果一个key下面的某条数据如果延迟到来太多，就会被丢弃，这个问题可以使用测输出流来解决最后还是迟到的数据；
因为其他key的数据会上报就会提高watermark，最后触发窗口计算。
代码测试：
package com.cuichunchi.watermark;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
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 java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.Iterator;
public class TestWaterMark {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.disableOperatorChaining();
DataStreamSource socketTextStream = env.socketTextStream("s201", 9099);
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
SingleOutputStreamOperator> tuple2Map = socketTextStream.map(new MapFunction>() {
@Override
public Tuple3 map(String value) throws Exception {
return new Tuple3<>(value.split(",")[0],1, Long.parseLong(value.split(",")[1]));
}
});
//提取时间戳
SingleOutputStreamOperator> tuple2WMDS = tuple2Map.assignTimestampsAndWatermarks(
//设置几秒watermark
WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(2))
.withTimestampAssigner(new SerializableTimestampAssigner>() {
@Override
public long extractTimestamp(Tuple3 element, long recordTimestamp) {
System.out.println("-------提取时间戳：" + dateFormat.format( element.f2 )+ ",默认处理时间戳：" + recordTimestamp);
return element.f2;
}
})
.withIdleness(Duration.ofMillis(3000))
);
//简单实时聚合
//TODO 通过源码 TimeWindow#getWindowStartWithOffset()来生成得watermark
//TODO org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows.assignWindows()
//TODO org.apache.flink.streaming.api.windowing.windows.TimeWindow.getWindowStartWithOffset()
//开窗聚合
OutputTag> diltyData = new OutputTag>("data"){};
SingleOutputStreamOperator> processDS =
tuple2WMDS.keyBy(value -> value.f0)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
// 容忍延迟 几秒的数据
.allowedLateness(Time.seconds(5))
//将延迟到的数据写入到侧输出流
.sideOutputLateData(diltyData)
.process(new ProcessWindowFunction, Tuple2, String,
TimeWindow>() {
@Override
public void process(String s, Context context, Iterable> elements,
Collector> out) throws Exception {
long start = context.window().getStart();
long end = context.window().getEnd();
long currentWatermark = context.currentWatermark();
System.out.println("======参数打印："+s+ ",elements:"+elements.toString()
+",当前开始window：["+dateFormat.format(start)+"]，当前结束window：["+dateFormat.format(end)+
"]，当前watermark:"+dateFormat.format(currentWatermark));
Iterator> it = elements.iterator();
int sum = 0;
while (it.hasNext()){
Tuple3 next = it.next();
sum += next.f1;
}
out.collect(new Tuple2<>(s,sum));
}
});
processDS.print("结果输出====》");
processDS.getSideOutput(diltyData).print("迟到数据测输出流====》");
env.execute();
}
}
watermark 设置的2秒，allowedLateness设置了5秒，window设置5秒。第一个窗口：[45，50）因为watermark 是2秒，所以推迟2秒 第一个窗口才触发，在52秒的时候才会触发，并且 允许延迟时间设置的5秒，所以第一个窗口在 50+2+5 = 57秒的时候，第一个窗口才会被清除。
测试结果：
-------提取时间戳：2020-04-10 11:32:46,默认处理时间戳：-9223372036854775808
-------提取时间戳：2020-04-10 11:32:47,默认处理时间戳：-9223372036854775808
-------提取时间戳：2020-04-10 11:32:52,默认处理时间戳：-9223372036854775808
======参数打印：01,elements:[(01,1,1586489566000), (01,1,1586489567000)],当前开始window：[2020-04-10 11:32:45]，当前结束window：[2020-04-10 11:32:50]，当前watermark:2020-04-10 11:32:49
结果输出====》> (01,2) （最后的结束窗口时间为50，那么在52秒的时候触发第一次窗口计算）
-------提取时间戳：2020-04-10 11:32:49,默认处理时间戳：-9223372036854775808
======参数打印：01,elements:[(01,1,1586489566000), (01,1,1586489567000), (01,1,1586489569000)],当前开始window：[2020-04-10 11:32:45]，当前结束window：[2020-04-10 11:32:50]，当前watermark:2020-04-10 11:32:49
结果输出====》> (01,3) （在允许迟到时间5秒范围内，50 + 2 + 5 = 57，如果超过57秒，就会停清除这个窗口的元数据，不会再触发计算，在范围内，每次到的数据都会触发一次计算）
-------提取时间戳：2020-04-10 11:32:57,默认处理时间戳：-9223372036854775808
======参数打印：01,elements:[(01,1,1586489572000)],当前开始window：[2020-04-10 11:32:50]，当前结束window：[2020-04-10 11:32:55]，当前watermark:2020-04-10 11:32:54
结果输出====》> (01,1) （57秒，触发的是第二个窗口，并且第一个窗口被清除元数据）
-------提取时间戳：2020-04-10 11:32:47,默认处理时间戳：-9223372036854775808
迟到数据测输出流====》> (01,1,1586489567000) （再次发送第一个窗口内的数据，因为第一个窗口已经被清除，所以最后在测输出流输出了）
