写在前面
在流处理过程中，一般我们希望事件是按照我们期望的顺序产生的。但有时候，由于网络问题或者时间临界值问题造成时间漂移，从而导致事件乱序，本应该在当前窗口处理的数据却迟迟没有到来。
如果我们使用kafka作为source，并且存在多个分区，这种情况下乱序最为明显了，因为多个分区的数据无法保证顺序。
这样，当我们我们使用EventTime时，就无法保证事件的完整性，从而导致计算结果有误差。幸好，Flink为我们提供了一种能处理乱序的手段，就是：Watermark。
什么是Watermark
Watermark 可以理解为一种时间标记，表示在这个时间之前的所有事件都已经到达了流处理系统。
换句话说，Watermark 表示了事件流的进度，可以帮助系统确定事件时间窗口何时可以关闭以及何时可以触发计算结果。
总之：Watermark 就是为了处理乱序的。
只有在使用 EventTime 时，才需要 Watermark。因为使用processTime时，不存在乱序问题。
Watermark类型
Flink 提供了2种 Watermark 类型：Periodic Watermarks 和 Punctuated Watermarks。
Periodic Watermarks
按照固定时间间隔生成 watermark，周期默认是200ms。一般这种 Watermark 比较常用。
Flink 1.10 的写法
inputStream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {
private Long currentMaxTimestamp = 0L;
@Override
public Watermark getCurrentWatermark() {
// 最大允许的消息延迟是10秒
Long maxOutOfOrderNess = 10 * 1000;
return new Watermark(currentMaxTimestamp - maxOutOfOrderNess);
}
@Override
public long extractTimestamp(String line, long l) {
long timestamp = Long.parseLong(line.split('|')[1]);
currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
return timestamp;
}
});Flink 1.10 以后的写法
inputStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(10)) {
@Override
public long extractTimestamp(String element) {
return Long.parseLong(element.split(" ")[0]);
}
});上述的 extractTimestamp 用于从消息中提取事件时间，getCurrentWatermark 是用于生成新的 watermark。
Punctuated Watermarks
通过数据流中某些特殊标记事件来触发新水位线的生成。这种方式下窗口的触发与时间无关，而是决定于何时收到标记事件。
inputStream.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<InputModel>() {
@Override
public Watermark checkAndGetNextWatermark(InputModel lastElement, long extractedTimestamp) {
// 当 endTime 不为null时生成 watermark
return ObjectUtils.isNotEmpty(lastElement.getEndTime()) ? new Watermark(extractedTimestamp) : null;
}
@Override
public long extractTimestamp(InputModel element, long previousElementTimestamp) {
return element.getEndTime();
}
})窗口触发时机
我们以一个例子来说明窗口是何时触发的，完整代码如下：
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
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.timestamps.BoundedOutOfOrdernessTimestampExtractor;
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.util.Collector;
/**
* 基于事件时间的滚动窗口示例
*/
public class TumblingEventWindowDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.setParallelism(1);
DataStream<String> inputStream = env.socketTextStream("localhost", 10000);
SingleOutputStreamOperator<String> outputStream = inputStream
.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
@Override
public long extractTimestamp(String element) {
long eventTime = Long.parseLong(element.split(" ")[0]);
System.out.println("eventTime：" + eventTime);
return eventTime;
}
})
.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String value) throws Exception {
return Tuple2.of(value.split(" ")[1], 1L);
}
})
.keyBy(0)
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.process(new ProcessWindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
@Override
public void process(Tuple tuple, ProcessWindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>.Context ctx, Iterable<Tuple2<String, Long>> elements, Collector<String> out) throws Exception {
long sum = 0;
for (Tuple2<String, Long> item : elements) {
sum += item.f1;
}
JSONObject json = new JSONObject();
json.put("windowEnd", ctx.window().getEnd());
json.put("waterMark", ctx.currentWatermark());
json.put("key", tuple.getField(0));
json.put("sum", sum);
out.collect(JSON.toJSONString(json));
}
});
outputStream.print("out");
env.execute();
}
}启动 socket 监听
nc -l 10000启动 TumblingEventWindowDemo 程序
在 idea 中找到 TumblingEventWindowDemo 类，右键运行即可。
输入测试数据
casey@shengyudeMacBook-Pro workspace % nc -l 10000
10000 a
12000 a
20000 a
21000 a
22000 a
23000 a
13000 a
30000 a
33000 a观察控制台输出
eventTime：10000
eventTime：12000
eventTime：20000
out> {"windowEnd":20000,"waterMark":20000,"key":"a","sum":2}
eventTime：21000
eventTime：22000
eventTime：23000
eventTime：13000
eventTime：30000
out> {"windowEnd":30000,"waterMark":30000,"key":"a","sum":4}
eventTime：33000结果分析
代码中我们设置了10秒的滚动窗口，waterMark 最大乱序为0，意味着每10秒生成一次 waterMark。
① 第1次输入 10000,a
此时waterMark 为10000 < windowEnd（20000），所以没有打印任何输出。
② 第2次输入12000,a
此时waterMark 为12000 < windowEnd（20000），所以没有打印任何输出。
③ 第3次输入20000,a
此时waterMark 为20000 >= windowEnd（20000），触发窗口计算，数据范围为[10000,20000)：10000，12000
④ 第4次输入21000,a 此时waterMark为21000 < windowEnd（30000），所以没有打印任何输出。
⑤ 第5次输入22000,a
此时waterMark为22000 < windowEnd（30000），所以没有打印任何输出。
⑥ 第6次输入23000,a
此时waterMark为23000 < windowEnd（30000），所以没有打印任何输出。
⑦ 第7次输入13000,a
此时waterMark仍然为23000 < windowEnd（30000），所以没有打印任何输出。
⑧ 第8次输入30000,a
此时waterMark为30000 >= windowEnd（30000），触发窗口计算，数据范围为[20000,30000)：20000，21000，22000，23000
总结
从上面的分析可以得出，在最大乱序设置为0的情况下：
① 窗口触发严格按照窗口大小来触发，13000属于乱序数据，直接被抛弃。
② 窗口是前闭后开，比如，10000的数据落在[10000, 20000) 窗口，20000数据落在 [20000, 30000) 窗口。
③ 已经触发过的窗口不会被再次触发，比如 13000 不会再次触发窗口 [10000, 20000)
设置最大乱序时间
上面我们已经了解了窗口的触发时机，那么对于13000，14000 这种乱序数据，有时候我们也希望他们也能参与窗口计算，那该如何处理呢？这时候我们可以设置一个最大乱序时间，将上面代码稍作修改即可：
inputStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(3)) {
@Override
public long extractTimestamp(String element) {
long eventTime = Long.parseLong(element.split(" ")[0]);
System.out.println("eventTime：" + eventTime);
return eventTime;
}
})只需要设置下Time.seconds(3)，即可设置最大乱序为3秒。这时候我们再来测一下：
输入测试数据
casey@shengyudeMacBook-Pro workspace % nc -l 10000
10000 a
12000 a
20000 a
21000 a
22000 a
23000 a
13000 a
30000 a
33000 a观察控制台输出
eventTime：10000
eventTime：12000
eventTime：20000
eventTime：21000
eventTime：22000
eventTime：23000
out> {"windowEnd":20000,"waterMark":20000,"key":"a","sum":2}
eventTime：13000
eventTime：30000
eventTime：33000
out> {"windowEnd":30000,"waterMark":30000,"key":"a","sum":4}结果分析
① 第1次输入10000,a
此时waterMark 为7000 < windowEnd（20000），所以没有打印任何输出。
② 第2次输入12000,a
此时waterMark 为9000 < windowEnd（20000），所以没有打印任何输出。
③ 第3次输入20000,a
此时waterMark 为17000 < windowEnd（20000），所以没有打印任何输出。可以看到此时并未触发窗口计算。
④ 第4次输入21000,a 此时waterMark为18000 < windowEnd（20000），所以没有打印任何输出。
⑤ 第5次输入22000,a
此时waterMark为19000 < windowEnd（20000），所以没有打印任何输出。
⑥ 第6次输入23000,a
此时waterMark为20000 >= windowEnd（20000）。触发窗口计算，数据范围为[10000,20000)：10000，12000。可以看到此时才触发了20000的窗口，窗口延迟了三秒触发。
⑦ 第7次输入13000,a
此时waterMark为10000 < windowEnd（20000），由于此窗口已经关闭，所以没有打印任何输出。
⑧ 第8次输入30000,a
此时waterMark为27000 < windowEnd（30000）, 所以没有打印任何输出。
⑨ 第9次输入33000,a
此时waterMark为30000 >= windowEnd（30000）。触发窗口计算，数据范围为[20000,30000)：20000，21000，22000，23000
延迟数据处理
延迟数据，指的是窗口已经触发，则在此之后进入窗口的数据，不会对窗口进行再次触发计算。要想在数据进入已经被触发过的窗口后，还能继续触发窗口计算，则可以使用延迟数据处理机制。
只要延迟的数据满足：watermark < windowEnd + allowedLateness，机会对窗口进行再次触发计算。
我们对上述代码再做点修改：
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.allowedLateness(Time.seconds(2)) // 允许延迟处理2秒
.process(new ProcessWindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {...}输入测试数据
casey@shengyudeMacBook-Pro workspace % nc -l 10000
10000 a
12000 a
20000 a
21000 a
22000 a
23000 a
13000 a
30000 a
33000 a观察控制台输出
eventTime：10000
eventTime：12000
eventTime：20000
eventTime：21000
eventTime：22000
eventTime：23000
out> {"windowEnd":20000,"waterMark":20000,"key":"a","sum":2}
eventTime：13000
out> {"windowEnd":20000,"waterMark":20000,"key":"a","sum":3}
eventTime：30000
eventTime：33000
out> {"windowEnd":30000,"waterMark":30000,"key":"a","sum":4}结果分析
① 第1次输入10000,a
此时waterMark 为7000 < windowEnd（20000），所以没有打印任何输出。
② 第2次输入12000,a
此时waterMark 为9000 < windowEnd（20000），所以没有打印任何输出。
③ 第3次输入20000,a
此时waterMark 为17000 < windowEnd（20000），所以没有打印任何输出。可以看到此时并未触发窗口计算。
④ 第4次输入21000,a 此时waterMark为18000 < windowEnd（20000），所以没有打印任何输出。
⑤ 第5次输入22000,a
此时waterMark为19000 < windowEnd（20000），所以没有打印任何输出。
⑥ 第6次输入23000,a
此时waterMark为20000 >= windowEnd（20000）。触发窗口计算，数据范围为[10000,20000)：10000，12000。可以看到此时才触发了20000的窗口，窗口延迟了三秒触发。
⑦ 第7次输入13000,a
此时waterMark为20000 < windowEnd（20000）+ allowedLateness（2000），再次触发计算，会累加之前的结果，数据范围为：[10000,20000)：10000，12000，13000。
⑧ 第8次输入30000,a
此时waterMark为27000 < windowEnd（30000）, 所以没有打印任何输出。
⑨ 第9次输入33000,a
此时waterMark为30000 >= windowEnd（30000）。触发窗口计算，数据范围为[20000,30000)：20000，21000，22000，23000
侧输出流
从上述结果我们可以看到，由于设置了allowedLateness(Time.seconds(2))，所以延迟的数据会多次触发之前的窗口。导致同一维度的数据会有多条输出，需要下游去重，但有时候我们希望将延迟的数据重定向到其他地方做分析使用，不想再次触发窗口。这时候我们就要使用到 SideOutput 了。
SideOutput 允许我们将迟到的数据输出到另一条流中，我们可以再从另一条流中获取数据做相应处理。代码如下：
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
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.timestamps.BoundedOutOfOrdernessTimestampExtractor;
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.util.Collector;
import org.apache.flink.util.OutputTag;
public class TumblingEventWindowDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.setParallelism(1);
DataStream<String> socketStream = env.socketTextStream("localhost", 10000);
OutputTag<Tuple2<String, Long>> lateOutputTag = new OutputTag<Tuple2<String, Long>>("side-output") {};
SingleOutputStreamOperator<String> resultStream = socketStream
.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(3)) {
@Override
public long extractTimestamp(String element) {
long eventTime = Long.parseLong(element.split(" ")[0]);
System.out.println(eventTime);
return eventTime;
}
})
.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String value) throws Exception {
return Tuple2.of(value.split(" ")[1], 1L);
}
})
.keyBy(0)
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.sideOutputLateData(lateOutputTag)
.allowedLateness(Time.seconds(2)) // 允许延迟处理2秒
.process(new ProcessWindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
@Override
public void process(Tuple tuple, ProcessWindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>.Context ctx, Iterable<Tuple2<String, Long>> elements, Collector<String> out) throws Exception {
long sum = 0;
for (Tuple2<String, Long> item : elements) {
sum += item.f1;
}
JSONObject json = new JSONObject();
json.put("windowEnd", ctx.window().getEnd());
json.put("waterMark", ctx.currentWatermark());
json.put("key", tuple.getField(0));
json.put("sum", sum);
out.collect(JSON.toJSONString(json));
}
});
DataStream<Tuple2<String, Long>> sideOutput = resultStream.getSideOutput(lateOutputTag);
resultStream.print("main");
sideOutput.print("side");
env.execute();
}
}输入数据
casey@shengyudeMacBook-Pro workspace % nc -l 10000
10000 a
25000 a
11000 a控制台输出
10000
25000
main> {"windowEnd":20000,"waterMark":22000,"key":"a","sum":1}
11000
side> (a,1)当25000进入window时，watermark被更新到22000，触发[10000, 20000)窗口进行计算；
当延迟数据11000到达窗口时，由于不满足watermark < window_end_time + lateTime，窗口无法被再次计算。
但是11000会被收集，重定向到sideOutput流中进行打印输出。
往期推荐
原文作者：毛毛小妖的笔记；如有侵权，请联系管理员删除。
