Flink解决延迟数据题目 - ToB企服应用市场:ToB评测及商务社交产业平台

使用方式:
先定义OutputTag对象(注意，必须new一个匿名内部类形式的OutputTag对象的实例)
然后调用sideOutputLateData方法
// side output OutputTag对象必须是匿名内部类的形式创建出来, 本质上得到的是OutputTag对象的一个匿名子类
OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("side output"){};
WindowedStream<Tuple2<String, Long>, Tuple, TimeWindow> sideOutputLateData =
allowedLateness.sideOutputLateData(outputTag);

复制代码

用法:
DataStream<Tuple2<String, Long>> sideOutput = result.getSideOutput(outputTag);
// 对得到的保存超级迟到数据的DataStream进行处理
sideOutput.print("late>>>");

复制代码

package com.bigdata.day05;
import com.bigdata.day04.OrderInfo;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
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.Random;
import java.util.UUID;
class MyOrderSource2 implements SourceFunction<OrderInfo> {
@Override
public void run(SourceContext<OrderInfo> ctx) throws Exception {
Random random = new Random();
while(true){
OrderInfo orderInfo = new OrderInfo();
orderInfo.setOrderId(UUID.randomUUID().toString().replace("-",""));
// 在这个地方可以模拟迟到数据
long orderTime = System.currentTimeMillis() - 1000*random.nextInt(100);
orderInfo.setOrdertime(orderTime);
int money = random.nextInt(10);
System.out.println("订单产生的时间："+ DateFormatUtils.format(orderTime,"yyyy-MM-dd HH:mm:ss")+"，金额："+money);
orderInfo.setMoney(money);
orderInfo.setUserId(random.nextInt(2));
ctx.collect(orderInfo);
Thread.sleep(500);
}
}
@Override
public void cancel() {
}
}
public class Demo01 {
public static void main(String[] args) throws Exception {
//1. env-准备环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
env.setParallelism(1);
// 每隔五秒统计每个用户的前面5秒的订单的总金额
//2. source-加载数据
DataStreamSource<OrderInfo> streamSource = env.addSource(new MyOrderSource2());
//-2.告诉Flink最大允许的延迟时间/乱序时间为多少
SingleOutputStreamOperator<OrderInfo> orderDSWithWatermark = streamSource.assignTimestampsAndWatermarks(
WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//-3.告诉Flink哪一列是事件时间
.withTimestampAssigner((order, time) -> order.getOrdertime())
);
OutputTag<OrderInfo> outputTag = new OutputTag<OrderInfo>("side output"){};
//3. transformation-数据处理转换
SingleOutputStreamOperator<String> result = orderDSWithWatermark.keyBy(orderInfo -> orderInfo.getUserId()).
window(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(4))
.sideOutputLateData(outputTag)
.apply(new WindowFunction<OrderInfo, String, Integer, TimeWindow>() {
@Override
public void apply(Integer key, // 代表分组key值五旬老太守国门
TimeWindow window, // 代表窗口对象
Iterable<OrderInfo> input, // 分组过之后的数据 [1,1,1,1,1]
Collector<String> out // 用于输出的对象
) throws Exception {
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
long start = window.getStart();
long end = window.getEnd();
int sum = 0;
// 专门存放迟到的订单时间
for (OrderInfo orderInfo : input) {
sum += orderInfo.getMoney();
}
out.collect(key + ",窗口开始：" + dateFormat.format(new Date(start)) + ",结束时间:" + dateFormat.format(new Date(end)) + "," + sum);
//out.collect(key+",窗口开始："+start+",结束时间:"+end+","+sum);
}
});
result.print("流中的数据，包含迟到的数据：");
result.getSideOutput(outputTag).print("严重迟到的数据：");
//4. sink-数据输出
//5. execute-执行
env.execute();
}
}

复制代码