watermark 的缺点
由 flink1.12.x watermark 水印、水位线 可知下面缺点：
收到 数据5 时，由于 数据5的事件时间 09:21:26 - 延迟 5 秒 >= 窗口结束时间 09:21:20，会触发 窗口1 计算
由于 data3 迟到太多，仍然会被 丢弃
为了避免有些迟到的数据被删除，产生了 allowedLateness 的概念。
解决 allowedLateness
allowedLateness 就是针对 event time 而言，当 watermark 大于 窗口结束时间 后，先不关闭窗口，延迟等待设定的时间，如果有迟到的数据，每来一条数据，就执行一次相应函数
执行时机
watermark 大于 窗口结束时间，且 watermark 小于 allowedLateness 时间，每来一条数据，就执行一次相应函数
api调用方式
详见链接：
https://www.malaoshi.top/show_1IX3swfMv1j7.html
例子
package allowedLateness;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
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.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Iterator;
import java.util.Properties;
public class Std1 {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 自定义watermark
WatermarkStrategy<Tuple3<String, Double, Long>> watermarkStrategy = WatermarkStrategy.
<Tuple3<String, Double, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(2))
.withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Double, Long>>() {
@Override
public long extractTimestamp(Tuple3<String, Double, Long> element, long recordTimestamp) {
return element.f2;
}
});
// 处理
SingleOutputStreamOperator<Object> soso = env.addSource(buildKafka())
.flatMap(new FlatMapFunImpl())
.assignTimestampsAndWatermarks(watermarkStrategy)
.keyBy(value -> value.f0)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(1))
.process(new ProcessWindowFunImpl());
// soso.print("res:");
System.out.println(env.getExecutionPlan());
env.execute("myjob");
}
private static class ProcessWindowFunImpl extends ProcessWindowFunction<Tuple3<String,Double,Long>, Object, String, TimeWindow> {
@Override
public void process(String key, Context context, Iterable<Tuple3<String, Double, Long>> input, Collector<Object> out) throws Exception {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
TimeWindow window = context.window();
String startTime = sdf.format(window.getStart());
String endTime = sdf.format(window.getEnd());
String content = String.format("\tRichWindowFunImpl--开始计算，key：%s，窗口开始时间：%s，窗口结束时间：%s，watermark：%s",
key,
startTime,
endTime,
context.currentWatermark());
System.out.println(content);
Iterator<Tuple3<String,Double, Long>> iterator = input.iterator();
while(iterator.hasNext()){
Tuple3<String,Double, Long> temp = iterator.next();
System.out.println("\tRichWindowFunImpl--遍历--"+temp);
}
System.out.println("\tRichWindowFunImpl--计算结束，key："+key);
out.collect("1");// 这里为了演示，收集数据不重要，所以收集1
}
}
private static FlinkKafkaConsumer<String> buildKafka(){
Properties pro = new Properties();
pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop1:9092,hadoop2:9092,hadoop3:9092");
pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group");
pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
// 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
return fkc;
}
static class FlatMapFunImpl implements FlatMapFunction<String, Tuple3<String,Double,Long>> {
@Override
public void flatMap(String value, Collector<Tuple3<String,Double,Long>> out) throws Exception {
ObjectMapper om = new ObjectMapper();
JsonNode jsonNode = om.readTree(value);
JsonNode idJN = jsonNode.get("id");
JsonNode tempJN = jsonNode.get("temp");
JsonNode timeJN = jsonNode.get("time");
String id = idJN.asText();
Double temp = tempJN.asDouble();
long timeStamp = timeJN.asLong();
Tuple3<String, Double, Long> ret = Tuple3.of(id, temp, timeStamp);
System.out.println("接收数据:"+ret);
out.collect(ret);
}
}
}
测试数据
{"id":"1","temp":20,"time":1000}
{"id":"2","temp":21,"time":4000}
{"id":"1","temp":22,"time":2000}
{"id":"1","temp":30,"time":3000}
{"id":"2","temp":40,"time":6000}
{"id":"1","temp":21,"time":7000}
{"id":"1","temp":100,"time":2000}
{"id":"2","temp":100,"time":4000}
{"id":"1","temp":21,"time":8000}
{"id":"1","temp":1000,"time":2000}
{"id":"2","temp":1000,"time":4000}
执行结果：
接收数据:(1,20.0,1000)
接收数据:(2,21.0,4000)
接收数据:(1,22.0,2000)
接收数据:(1,30.0,3000)
接收数据:(2,40.0,6000)
接收数据:(1,21.0,7000)
RichWindowFunImpl--开始计算，窗口开始时间：1970-01-01 08:00:00.000，窗口结束时间：1970-01-01 08:00:05.000，watermark：4999
RichWindowFunImpl--遍历--(1,20.0,1000)
RichWindowFunImpl--遍历--(1,22.0,2000)
RichWindowFunImpl--遍历--(1,30.0,3000)
RichWindowFunImpl--计算结束
RichWindowFunImpl--开始计算，窗口开始时间：1970-01-01 08:00:00.000，窗口结束时间：1970-01-01 08:00:05.000，watermark：4999
RichWindowFunImpl--遍历--(2,21.0,4000)
RichWindowFunImpl--计算结束
接收数据:(1,100.0,2000)
RichWindowFunImpl--开始计算，窗口开始时间：1970-01-01 08:00:00.000，窗口结束时间：1970-01-01 08:00:05.000，watermark：4999
RichWindowFunImpl--遍历--(1,20.0,1000)
RichWindowFunImpl--遍历--(1,22.0,2000)
RichWindowFunImpl--遍历--(1,30.0,3000)
RichWindowFunImpl--遍历--(1,100.0,2000)
RichWindowFunImpl--计算结束
接收数据:(2,100.0,4000)
RichWindowFunImpl--开始计算，窗口开始时间：1970-01-01 08:00:00.000，窗口结束时间：1970-01-01 08:00:05.000，watermark：4999
RichWindowFunImpl--遍历--(2,21.0,4000)
RichWindowFunImpl--遍历--(2,100.0,4000)
RichWindowFunImpl--计算结束
接收数据:(1,21.0,8000)
接收数据:(1,1000.0,2000)
接收数据:(2,1000.0,4000)
参考：
https://blog.csdn.net/lmalds/article/details/55259718