FLink迟到数据的处理之三

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

Flink迟到的数据更新窗口计算结果，窗口销毁后的迟到数据输出到测输出流

主程序：

//TODO 使用迟到的数据更新窗口的计算结果
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "hadoop106:9092");
SingleOutputStreamOperator<String> result = env.addSource(new FlinkKafkaConsumer<String>
("Tuple2", new SimpleStringSchema(), properties))
.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String value) throws Exception {
return Tuple2.of(value.split(" ")[0], Long.parseLong(value.split(" ")[1]) * 1000L);
}
}).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
@Override
public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
return element.f1;
}
})).keyBy(r -> r.f0)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(5))//允许迟到时间为5秒窗口闭合触发计算但是窗口不销毁 5秒内迟到的数据如果
// 到来的话更新窗口的计算结果允许迟到时间和最大延迟时间的区别是设置允许迟到时间可以提前看到窗口的聚合结果虽然不准确
// 但是后面迟到的数据到来的话会更新窗口的计算结果如果时间超过了允许迟到时间那么可以将迟到的数据发送到侧输出流中
.sideOutputLateData(new OutputTag<Tuple2<String, Long>>("lateStream") {
})
.process(new ProcessWindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
@Override
public void process(String s, Context context, Iterable<Tuple2<String, Long>> elements, Collector<String> out) throws Exception {
// out.collect("key为:"+s+"窗口的时间范围是："+new Timestamp(context.window().getStart())+
// "-"+new Timestamp(context.window().getEnd())+"元素的个数是："+elements.spliterator().
// getExactSizeIfKnown());
// 窗口闭合的时候，触发窗口的第一次计算
ValueState<Boolean> firstCalculate = context.windowState().getState(new
ValueStateDescriptor<Boolean>("firstCalculate", Types
.BOOLEAN));
if (firstCalculate.value() == null) {
out.collect("窗口第一次触发计算，窗口中的元素个数是：" + elements.spliterator().
getExactSizeIfKnown() + "窗口的时间范围是：" + new Timestamp(context.window().getStart()) + "-" +
new Timestamp(context.window().getEnd()));
firstCalculate.update(true);
} else {
out.collect("迟到的元素触发计算了,更新后的元素的个数是：" + elements.spliterator()
.getExactSizeIfKnown());
}
}
});
result.printToErr("主流输出");
result.getSideOutput(new OutputTag<Tuple2<String, Long>>("lateStream") {
}).print("测输出流");
env.execute();
}