watermark缺点
由 flink1.12.x watermark 水印、水位线 可知下面缺点：
收到 数据5 时，由于 数据5的事件时间 09:21:26 - 延迟 5 秒 >= 窗口结束时间 09:21:20，会触发 窗口1 计算
由于 data3 迟到太多，仍然会被 丢弃
allowedLateness、OutputTag
由下面文章可知：
allowedLateness 之后，即：收到 事件时间是 8000 的数据，8000ms - 2s(watermark) - 3s(allowedLateness) > 5s（窗口结束时间），那么 窗口1关闭，再接收到 窗口1的数据，就放入到 OutputTag 侧流中
API 说明
创建 OuputTag
OutputTag<Tuple3<String, Double, Long>> outputTag = new OutputTag("late_data", Types.TUPLE(Types.STRING,Types.DOUBLE,Types.LONG));
api 调用
详见 链接
获取侧输出流
DataStream<Tuple3<String, Double, Long>> lateDataDS = maxSOSO.getSideOutput(outputTag);
例子
package allowerdLateness_outputtag;
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.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
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.flink.util.OutputTag;
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 Std3 {
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;
}
});
/*
----------------------关键----------------------
创建 OutputTag 对象，收集 由于迟到太多丢弃的数据
*/
OutputTag<Tuple3<String, Double, Long>> outputTag = new OutputTag("late_data", Types.TUPLE(Types.STRING,Types.DOUBLE,Types.LONG));
// 处理
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))
.sideOutputLateData(outputTag)
.process(new ProcessWindowFunImpl());
DataStream<Tuple3<String, Double, Long>> sideOutput = soso.getSideOutput(outputTag);
sideOutput.print("sideOutput::");
// 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}
执行结果
参考：
https://blog.csdn.net/lmalds/article/details/55259718