上接：flink1.12.x WaterMark 迟到数据丢失和收集（OutputTag、allowLateness）
案例
基于 flink1.12.x WaterMark 案例：滚动窗口 改进
flink部分
关键代码
/*
----------------------关键----------------------
创建 OutputTag 对象，收集 由于迟到太多丢弃的数据
*/
OutputTag<Tuple3<String, Double, Long>> outputTag = new OutputTag("late_data", Types.TUPLE(Types.STRING,Types.DOUBLE,Types.LONG));
WindowedStream<Tuple3<String, Double, Long>, String, TimeWindow> sideOutputWS = window
.allowedLateness(Time.seconds(maxOutOfOrderness)) // 指定允许迟到的时间，一般为最大超时时间
.sideOutputLateData(outputTag); // 传入侧流 outputTag，丢弃的数据放入到这里
// 指定聚合操作，求最大值
SingleOutputStreamOperator<Tuple3<String, Double, Long>> maxSOSO = sideOutputWS
.max(1);// 根据索引是1的字段，取出最大值
/*
----------------------关键----------------------
从 OutputTag对象中，获取数据
*/
DataStream<Tuple3<String, Double, Long>> lateDataDS = maxSOSO.getSideOutput(outputTag);
maxSOSO.print("result:");
// 这里直接打印
lateDataDS.print("later data:");
完整代码
package watermarker;
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.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
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.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.Properties;
public class SlidingWindow收集迟到太多的数据 {
//最大超时时间
static long maxOutOfOrderness = 5;
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 定义kafka相关配置项
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);
DataStreamSource<String> dss = env.addSource(fkc);
// map解析json数据
SingleOutputStreamOperator<Tuple3<String, Double, Long>> mapSOSO = dss.map(new MyMap());
// 设置watermark最大迟到时间
WatermarkStrategy<Tuple3<String, Double, Long>> maxDelayWS = WatermarkStrategy
.<Tuple3<String, Double, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(maxOutOfOrderness));
//指定watermark事件时间字段
WatermarkStrategy<Tuple3<String, Double, Long>> timestampWS = maxDelayWS
.withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Double, Long>>(){
@Override
public long extractTimestamp(Tuple3<String, Double, Long> element, long recordTimestamp) {
return element.f2;
}
});
// 设置 watermark
SingleOutputStreamOperator<Tuple3<String, Double, Long>> watermarkSOSO = mapSOSO
.assignTimestampsAndWatermarks(timestampWS);
// keyBy分组
KeyedStream<Tuple3<String, Double, Long>, String> keyByKS = watermarkSOSO
.keyBy(new KeySelector<Tuple3<String, Double, Long>, String>() {
@Override
public String getKey(Tuple3<String, Double, Long> value) throws Exception {
return value.f0;
}
});
// 设置滚动窗口
WindowedStream<Tuple3<String, Double, Long>, String, TimeWindow> window = keyByKS
.window(TumblingEventTimeWindows.of(Time.seconds(10)));
/*
----------------------关键----------------------
创建 OutputTag 对象，收集 由于迟到太多丢弃的数据
*/
OutputTag<Tuple3<String, Double, Long>> outputTag = new OutputTag("late_data", Types.TUPLE(Types.STRING,Types.DOUBLE,Types.LONG));
WindowedStream<Tuple3<String, Double, Long>, String, TimeWindow> sideOutputWS = window
.allowedLateness(Time.seconds(maxOutOfOrderness)) // 指定允许迟到的时间，一般为最大超时时间
.sideOutputLateData(outputTag); // 传入侧流 outputTag，丢弃的数据放入到这里
// 指定聚合操作，求最大值
SingleOutputStreamOperator<Tuple3<String, Double, Long>> maxSOSO = sideOutputWS
.max(1);// 根据索引是1的字段，取出最大值
/*
----------------------关键----------------------
从 OutputTag对象中，获取数据
*/
DataStream<Tuple3<String, Double, Long>> lateDataDS = maxSOSO.getSideOutput(outputTag);
maxSOSO.print("result:");
// 这里直接打印
lateDataDS.print("later data:");
env.execute();
}
public static class MyMap implements MapFunction<String, Tuple3<String,Double,Long>>{
@Override
public Tuple3<String,Double,Long> map(String data) throws Exception {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date sysDate = new Date();
String sysTime = sdf.format(sysDate);
ObjectMapper om = new ObjectMapper();
JsonNode jsonNode = om.readTree(data);
JsonNode idJN = jsonNode.get("id");
JsonNode tempJN = jsonNode.get("temp");
JsonNode timeJN = jsonNode.get("time");
String id = null;
Double temp = null;
long timeStamp = 0;
if(idJN!=null){
id = idJN.asText();
}
if(tempJN!=null){
temp = tempJN.asDouble();
}
String timeStr = null;
if(timeJN!=null){
timeStr = timeJN.asText();
Date date = sdf.parse(timeStr);
timeStamp = date.getTime();
}
System.out.println("receive:"+Tuple3.of(id,temp,timeStr)+",receive time:"+sysTime);
return Tuple3.of(id,temp,timeStamp);
}
}
}
kafka 生产者
其他配置见 flink1.12.x WaterMark 案例：kafka生产者
主类
package top.malaoshi.kafka;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.Random;
public class Producer循环发送迟到太多的数据 {
final static String TOPIC = "kafka2flink";
public static void main(String[] args) throws InterruptedException {
// 1. 创建 kafka 生产者的配置对象
Properties properties = new Properties();
// 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
// key,value 序列化（必须）：key.serializer，value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
// StringSerializer.class.getName() 等于 org.apache.kafka.common.serialization.StringSerializer
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer") ;
// 3. 创建 kafka 生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer(properties);
Random random = new Random();
double temp = 1;
for(int i=0;i<50;i++) {
// 随机生成 20 以下的数字，作为迟到时间
int delay = random.nextInt(100);
String event1 = EventBuilder.buildJson("1", temp+i, delay);
// 4. 调用 send 方法,发送消息
kafkaProducer.send(new ProducerRecord(TOPIC , event1));
System.out.println("send:" + event1);
Thread.sleep(500);
}
// 5. 关闭资源
kafkaProducer.close();
System.out.println("end...");
}
}
测试
启动zk
登录 hadoop1，执行下面命令：
/program/bin/zk.sh start
启动kafka
登录 hadoop1，执行下面命令：
/program/bin/kafka.sh start
启动 flink
在 idea 中运行 flink 程序
启动生产者、指定主题
执行上面的 kafka 生产者
执行结果
在 idea 控制台中打印结果如下：
收集迟到太久的数据