一步一个脚印,一天一道面试题。
感觉我现在很难把水印描述的很好,但,完成比完美更重要。后续我再补充。各位如果有什么建议或补充也欢迎留言。
在实时处理任务时,由于网络延迟,人工异常,各种问题,数据往往会出现乱序,不按照我们的预期到达处理框架。
WaterMark 水印,就是为了一定程度的解决数据,延迟乱序问题的。
使用 WaterMark 一般有以下几个步骤:
- 定义时间特性
(Flink 1.12 已废弃,默认使用 事件时间)
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
- 设置 Watermark 策略,赋值事件时间
// 分配时间戳和水位线DataStream<Tuple2<Long, Integer>> withTimestampsAndWatermarks = parsedStream.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<Long, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((element, recordTimestamp) -> element.f0));
话不多说,直接给个 Watermark 水印样例代码。
public class SimpleWatermarkExample {public static void main(String[] args) throws Exception {// 设置流执行环境final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 从 socket 文本流接收数据DataStream<String> input = env.addSource(new SocketTextStreamFunction("localhost", 9999, "\n", -1));// 解析输入的数据DataStream<Tuple2<Long, Integer>> parsedStream = input.map(new MapFunction<String, Tuple2<Long, Integer>>() {@Overridepublic Tuple2<Long, Integer> map(String value) throws Exception {String[] parts = value.split(",");return new Tuple2<>(Long.parseLong(parts[0]), Integer.parseInt(parts[1]));}});// 分配时间戳和水位线DataStream<Tuple2<Long, Integer>> withTimestampsAndWatermarks = parsedStream.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<Long, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((element, recordTimestamp) -> element.f0));// 使用窗口函数统计每10秒内的最大值DataStream<String> maxValues = withTimestampsAndWatermarks.windowAll(TumblingEventTimeWindows.of(Time.seconds(10))).apply(new WindowFunction<Tuple2<Long, Integer>, String, TimeWindow>() {@Overridepublic void apply(TimeWindow window, Iterable<Tuple2<Long, Integer>> values, Collector<String> out) throws Exception {int maxValue = Integer.MIN_VALUE;for (Tuple2<Long, Integer> value : values) {maxValue = Math.max(maxValue, value.f1);}out.collect("Window: " + window + " Max Value: " + maxValue);}});// 打印结果maxValues.print();// 执行程序env.execute("Simple Flink Watermark Example");}
}