最近朋友面试某猪的时候，被问到一个问题答得面试官不太满意，问的是前司数据延迟问题是怎么解决的，我稍作整理。

一、什么是延迟数据

大数据处理过程中 Join 的场景太多太多了，几乎所有公司的 APP 都会涉及到两条流数据之间的维度拼接，将表变宽等场景，避免不了进行多流 Join 操作。同时join场景中受网络或物理设备等因素影响也有可能，以致出现不同的流式数据到达计算引擎的时间不一定，那这些数据称为延迟数据。即延迟数据是指系统的事件时间戳在经过延迟元素时间戳之后的时间到达的数据。所以延迟数据可以说是一种特殊的乱序数据，它没有被watermark和Window机制处理，因为是在窗口关闭后才到达的数据。一般这种情况有三种处理办法：

• 重新激活已经关闭的窗口并重新计算以修正结果。

• 将迟到数据收集起来另外处理。

• 将迟到数据直接丢弃。

Flink默认采用第三种方法，将迟到数据直接丢弃。接下来我们演示一个延迟数据案例，之后使用Flink提供的waterMark、allowedLateness机制、sideOutput机制解决延迟数据的场景。

二、Flink延迟数据场景

这里是模拟将两个Socket流使用事件时间进行5秒滚动窗口Left Outer join后，延迟数据被丢失的场景，延迟数据没有被正常输出。

1. 代码

object MyCoGroupJoin {def main(args: Array[String]): Unit = {// 创建环境变量val env = StreamExecutionEnvironment.getExecutionEnvironment// 指定事件时间env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)// 创建Socket源数据流，内容格式 "时间戳 单词"val s1 = env.socketTextStream("hadoop52", 9999)// 设置事件时间戳字段.assignAscendingTimestamps(_.split(" ")(0).toLong).map(line => {val strs = line.split(" ")(strs(0).toLong, strs(1))})// 创建Socket源数据流，内容格式 "时间戳 单词"val s2 = env.socketTextStream("hadoop52", 8888)// 设置事件时间戳字段.assignAscendingTimestamps(_.split(" ")(0).toLong).map(line => {val strs = line.split(" ")(strs(0).toLong, strs(1))})//将两个数据流进行合并统计，这里是将两数据流利用窗口进行单词拼串处理s1.coGroup(s2).where(_._2).equalTo(_._2).window(TumblingEventTimeWindows.of(Time.seconds(5))) //滚动窗口，窗口大小5秒.apply(new CoGroupFunction[(Long, String), (Long, String), String] {override def coGroup(first: lang.Iterable[(Long, String)], second: lang.Iterable[(Long, String)], out: Collector[String]): Unit = {// 对两数据流中的数据进行循环遍历，并拼串下发first.forEach(r1 => {second.forEach(r2 => {println(s"${r1}::${r2}")val str = r1._2 + r2._2out.collect(str)})})}}).print()env.execute("cogroupjoin")}}

2. Socket数据源1

[atguigu@hadoop52 ~]$ nc -l 9999

1001000 hello

1005000 java

1003000 hello [这条被丢弃了]

1010000 xixi

3. Socket数据源2

[atguigu@hadoop52 ~]$ nc -l 8888

1002000 hello

1005000 java

1001000 hello [这条被丢弃了]

1010000 xixi

4. 程序执行控制台输出结果

(1001000,hello)::(1002000,hello)

4> hellohello

(1005000,java)::(1005000,java)

2> javajava

我们发现在Socket数据源输入 "1005000 java" 后，会统计1005000时间戳之前的数据，而在1005000时间戳之后输入的hello就没有被统计输出。当输入 "1010000 xixi" 后，触发了第2个窗口，只输出了java，还是没有后输入的hello统计结果，这也更明确了1005000时间戳之后输入的hello被丢弃了。数据很重要，不想丢弃怎么办，我们可以使用Flink提供的水印位(waterMark)解决。

三、解决方案之waterMark

watermark是flink为了处理event time窗口计算提出的一种机制，本质上就是一个时间戳，代表着比这个时间早的事件已经全部进入到相应的窗口，后续不会再有比这个时间小的事件出现，基于这个前提我们才有可能将event time窗口视为完整并触发窗口的计算。

1. 程序代码

object MyCoGroupJoin {def main(args: Array[String]): Unit = {// 创建环境变量val env = StreamExecutionEnvironment.getExecutionEnvironment// 指定事件时间env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)// 创建Socket源数据流，内容格式 "时间戳 单词"val s1 = env.socketTextStream("hadoop52", 9999)// 设置事件时间戳字段// .assignAscendingTimestamps(_.split(" ")(0).toLong)// 这里可以指定周期性产生WaterMark 或 间歇性产生WaterMark，分别使用AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks来实现// 这里使用周期性产生WaterMark,延长2秒.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(2)) {override def extractTimestamp(element: String): Long = {val strs = element.split(" ")strs(0).toLong}}).map(line => {val strs = line.split(" ")(strs(0).toLong, strs(1))})// 创建Socket源数据流，内容格式 "时间戳 单词"val s2 = env.socketTextStream("hadoop52", 8888)// 设置事件时间戳字段// .assignAscendingTimestamps(_.split(" ")(0).toLong)// 这里可以指定周期性产生WaterMark 或 间歇性产生WaterMark，分别使用AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks来实现// 这里使用周期性产生WaterMark,延长2秒.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(2)) {override def extractTimestamp(element: String): Long = {val strs = element.split(" ")strs(0).toLong}}).map(line => {val strs = line.split(" ")(strs(0).toLong, strs(1))})//将两个数据流进行合并统计，这里是将两数据流利用窗口进行单词拼串处理s1.coGroup(s2).where(_._2).equalTo(_._2).window(TumblingEventTimeWindows.of(Time.seconds(5))) //滚动窗口，窗口大小5秒.apply(new CoGroupFunction[(Long, String), (Long, String), String] {override def coGroup(first: lang.Iterable[(Long, String)], second: lang.Iterable[(Long, String)], out: Collector[String]): Unit = {// 对两数据流中的数据进行循环遍历，并拼串下发first.forEach(r1 => {second.forEach(r2 => {println(s"${r1}::${r2}")val str = r1._2 + r2._2out.collect(str)})})}}).print()env.execute("cogroupjoin")}}

2. Socket数据源1

[atguigu@hadoop52 ~]$ nc -l 9999

1001000 hello

1005000 java

1003000 hello

1007000 java

3. Socket数据源2

[atguigu@hadoop52 ~]$ nc -l 8888

1002000 hello

1005000 java

1001000 hello

1007000 java

4. 程序执行控制台输出结果

(1001000,hello)::(1002000,hello)

4> hellohello

(1001000,hello)::(1001000,hello)

4> hellohello

(1003000,hello)::(1002000,hello)

4> hellohello

(1003000,hello)::(1001000,hello)

4> hellohello

当我们使用WaterMark后，我们可以发现在两个Socket终端输入"1005000 java"时，控制台并没有立刻统计输出信息。而是在两个Socket终端输入 "1007000 java"

后，控制台才将统计结果输出出来且在时间戳"1005000"之后输入的hello也同时给统计出来了，上面的问题可以解决了，但是 "1007000 java" 之后我们再输入 hello ，你会发现还是存在问题，没有输出又给丢弃了。继续测试如下。

5. Socket数据源1

[atguigu@hadoop52 ~]$ nc -l 9999

1001000 hello

1005000 java

1003000 hello

1007000 java

1003000 hello

1012000 spark

6. Socket数据源2

[atguigu@hadoop52 ~]$ nc -l 8888

1002000 hello

1005000 java

1001000 hello

1007000 java

1004000 hello

1012000 spark

7. 程序执行控制台输出结果

(1001000,hello)::(1002000,hello)

4> hellohello

(1001000,hello)::(1001000,hello)

4> hellohello

(1003000,hello)::(1002000,hello)

4> hellohello

(1003000,hello)::(1001000,hello)

4> hellohello

(1005000,java)::(1005000,java)

2> javajava

(1005000,java)::(1007000,java)

2> javajava

(1007000,java)::(1005000,java)

2> javajava

(1007000,java)::(1007000,java)

2> javajava

所以waterMark只能在一定程度上解决这种问题。我们再来看看allowedLateness机制。

四、解决方案之allowedLateness机制

默认情况下，当watermark通过end-of-window之后，再有之前的数据到达时，这些数据会被丢弃。为了避免有些迟到的数据被丢弃，因此产生了allowedLateness机制。简单来讲，allowedLateness就是针对event time而言，对于watermark超过end-of-window之后还允许有一段时间（也是以event time来衡量）来等待之后的数据到达，以便再次处理这些数据。

1. 程序代码

object MyCoGroupJoin {def main(args: Array[String]): Unit = {// 创建环境变量val env = StreamExecutionEnvironment.getExecutionEnvironment// 指定事件时间env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)// 创建Socket源数据流，内容格式 "时间戳 单词"val s1 = env.socketTextStream("hadoop52", 9999)// 设置事件时间戳字段// .assignAscendingTimestamps(_.split(" ")(0).toLong)// 这里可以指定周期性产生WaterMark 或 间歇性产生WaterMark，分别使用AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks来实现// 这里使用周期性产生WaterMark,延长2秒.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(2)) {override def extractTimestamp(element: String): Long = {val strs = element.split(" ")strs(0).toLong}}).map(line => {val strs = line.split(" ")(strs(0).toLong, strs(1))})// 创建Socket源数据流，内容格式 "时间戳 单词"val s2 = env.socketTextStream("hadoop52", 8888)// 设置事件时间戳字段// .assignAscendingTimestamps(_.split(" ")(0).toLong)// 这里可以指定周期性产生WaterMark 或 间歇性产生WaterMark，分别使用AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks来实现// 这里使用周期性产生WaterMark,延长2秒.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(2)) {override def extractTimestamp(element: String): Long = {val strs = element.split(" ")strs(0).toLong}}).map(line => {val strs = line.split(" ")(strs(0).toLong, strs(1))})//将两个数据流进行合并统计，这里是将两数据流利用窗口进行单词拼串处理s1.coGroup(s2).where(_._2).equalTo(_._2).window(TumblingEventTimeWindows.of(Time.seconds(5))) //滚动窗口，窗口大小5秒// 允许数据迟到2秒，窗口触发后2秒内过来的数据还可以重新被计算.allowedLateness(Time.seconds(2)).apply(new CoGroupFunction[(Long, String), (Long, String), String] {override def coGroup(first: lang.Iterable[(Long, String)], second: lang.Iterable[(Long, String)], out: Collector[String]): Unit = {// 对两数据流中的数据进行循环遍历，并拼串下发first.forEach(r1 => {second.forEach(r2 => {println(s"${r1}::${r2}")val str = r1._2 + r2._2out.collect(str)})})}}).print()env.execute("cogroupjoin")}}

2. Socket数据源1

[atguigu@hadoop52 ~]$ nc -l 9999

1001000 hello

1007000 java

3. Socket数据源2

[atguigu@hadoop52 ~]$ nc -l 8888

1002000 hello

1007000 java

1003000 hello

4. 程序执行控制台输出结果

(1001000,hello)::(1002000,hello)

4> hellohello

(1001000,hello)::(1002000,hello)

4> hellohello

(1001000,hello)::(1003000,hello)

4> hellohello

到这里估计有朋友又有疑问了，allowedLateness机制解决数据延迟设置的时间段，那之后再来的延迟数据呢，还是被丢弃了并没有彻底解决问题。别慌，针对allowedLateness机制之后来的延迟数据Flink还提供了另一种方案就是sideOutput机制。

五、解决方案之sideOutput机制

Side Output简单来说就是在程序执行过程中,将主流stream流中的不同的业务类型或者不同条件的数据分别输出到不同的地方。如果我们想对没能及时在Flink窗口计算的延迟数据专门处理，也就是窗口已经计算了，但后面才来的数据专门处理，我们可以使用旁路输出到侧流中去处理。

1. 程序代码

object MyCoGroupJoin {def main(args: Array[String]): Unit = {// 创建环境变量val env = StreamExecutionEnvironment.getExecutionEnvironment// 指定事件时间env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)// 创建Socket源数据流，内容格式 "时间戳 单词"val s1 = env.socketTextStream("hadoop52", 9999)// 这里可以指定周期性产生WaterMark 或 间歇性产生WaterMark，分别使用AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks来实现// 这里使用周期性产生WaterMark,延长2秒
.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(2)) {override def extractTimestamp(element: String): Long = {val strs = element.split(" ")strs(0).toLong}}).map(line => {val strs = line.split(" ")(strs(0).toLong, strs(1))})// 定义一个侧输出流val lateData: OutputTag[(Long, String)] = new OutputTag[(Long, String)]("late")val s2 = s1.timeWindowAll(Time.seconds(5))//允许迟到2秒数据.allowedLateness(Time.seconds(2))//迟到长于2秒的数据将被保存到lateData侧数据流中.sideOutputLateData(lateData).process(new ProcessAllWindowFunction[(Long, String), (Long, String), TimeWindow] {override def process(context: Context, elements: Iterable[(Long, String)], out: Collector[(Long, String)]): Unit = {elements.foreach(ele => {out.collect(ele)})}})s2.print("主流")s2.getSideOutput(lateData).print("侧流")env.execute("cogroupjoin")}}

2. Socket数据源

[atguigu@hadoop52 ~]$ nc -l 9999

1001000 hello

1005000 java

1007000 python

1002000 hello

1009000 java

1001000 xixi

1002000 haha

3. 程序执行控制台输出结果

主流:10> (1001000,hello)

主流:11> (1001000,hello)

主流:12> (1002000,hello)

侧流:1> (1001000,xixi)

侧流:2> (1002000,haha)

通过上面测试可以发现晚于allowedLateness机制的延迟数据，Flink没有丢弃而是输出到了侧输出流中等待处理了，这样延迟数据就完美解决了。

六、总结

sideOutput机制可以将迟到事件单独放入一个数据流分支，这会作为 window 计算结果的副产品，以便用户获取并对其进行特殊处理。

allowedLateness机制允许用户设置一个允许的最大迟到时长。Flink 会在窗口关闭后一直保存窗口的状态直至超过允许迟到时长，这期间的迟到事件不会被丢弃，而是默认会触发窗口重新计算。