一、Flink的多种时间概念介绍和应用场景
- 背景
- 前面我们使用了Window窗口函数，flink怎么知道哪个是字段是对应的时间呢？
- 由于网络问题，数据先产生，但是乱序延迟了，那属于哪个时间窗呢？
- Flink里面定义窗口，可以引用不同的时间概念
- Flink里面时间分类
- 事件时间EventTime（重点关注）
- 事件发生的时间
- 事件时间是每个单独事件在其产生进程上发生的时间，这个时间通常在记录进入 Flink 之前记录在对象中
- 在事件时间中，时间值 取决于数据产生记录的时间，而不是任何Flink机器上的
- 进入时间 IngestionTime
- 事件到进入Flink
- 处理时间ProcessingTime
- 事件被flink处理的时间
- 指正在执行相应操作的机器的系统时间
- 是最简单的时间概念，不需要流和机器之间的协调，它提供最佳性能和最低延迟
- 但是在分布式和异步环境中，处理时间有不确定性，存在延迟或乱序问题
- 事件时间EventTime（重点关注）
- 大家的疑惑
- 事件时间已经能够解决所有的问题了，那为何还要用处理时间呢？？？？
- 处理时间由于不用考虑事件的延迟与乱序，所以处理数据的速度高效
- 如果一些应用比较重视处理速度而非准确性，那么就可以使用处理时间，但结果具有不确定性
- 事件时间有延迟，但是能够保证处理的结果具有准确性，并且可以处理延迟甚至无序的数据
- 举个例子
- 小滴课堂-老王 做了一个电商平台买 "超短男装衣服"，如果要统计10分钟内成交额，你认为是哪个时间比较好？
- (EventTime) 下单支付时间是2022-11-11 01-01-01
- (IngestionTime ) 进入Flink时间2022-11-11 01-03-01（网络拥堵、延迟）
- (ProcessingTime)进入窗口时间2022-11-11 01-31-01（网络拥堵、延迟）
- 小滴课堂-老王 做了一个电商平台买 "超短男装衣服"，如果要统计10分钟内成交额，你认为是哪个时间比较好？
二、Flink乱序延迟时间处理-Watermark讲解《上》
- 背景
- 一般我们都是用EventTime事件时间进行处理统计数据
- 但数据由于网络问题延迟、乱序到达会导致窗口计算数据不准确
- 需求：比如时间窗是 [12:01:01,12:01:10 ) ，但是有数据延迟到达
- 当 12:01:10 秒数据到达的时候，不立刻触发窗口计算
- 而是等一定的时间，等迟到的数据来后再关闭窗口进行计算
- 生活中的例子
- 小滴课堂：每天10点后就是迟到，需要扣工资
- 老王上班 路途遥远(延迟) 经常迟到
- HR就规定迟到5分钟后就罚款100元（5分钟就是watermark）
- 迟到30分钟就是上午事假处理 (5~30分就是 allowLateness )
- 不请假都是要来的 (超过30分钟就是侧输出流，sideOutPut兜底)
- 超过5分钟就不用来了吗？还是要来的继续工作的，不然今天上午工资就没了
- 那如果迟到30分钟呢？ 也要来的，不然就容易产生更大的问题，缺勤开除。。。。
- Watermark 水位线介绍
- 由flink的某个operator操作生成后，就在整个程序中随event数据流转
- With Periodic Watermarks（周期生成，可以定义一个最大允许乱序的时间，用的很多）
- With Punctuated Watermarks（标点水位线，根据数据流中某些特殊标记事件来生成，相对少）
- 衡量数据是否乱序的时间，什么时候不用等早之前的数据
- 是一个全局时间戳，不是某一个key下的值
- 是一个特殊字段，单调递增的方式，主要是和数据本身的时间戳做比较
- 用来确定什么时候不再等待更早的数据了，可以触发窗口进行计算，忍耐是有限度的，给迟到的数据一些机会
- 注意
- Watermark 设置太小会影响数据准确性，设置太大会影响数据的实时性，更加会加重Flink作业的负担
- 需要经过测试，和业务相关联，得出一个较合适的值即可
- 由flink的某个operator操作生成后，就在整个程序中随event数据流转
- 窗口触发计算的时机
- watermark之前是按照窗口的关闭时间点计算的 [12:01:01,12:01:10 )
- watermark之后，触发计算的时机
- 窗口内有数据
- Watermaker >= Window EndTime窗口结束时间
- 触发计算后，其他窗口内数据再到达也被丢弃
- Watermaker = 当前计算窗口最大的事件时间 - 允许乱序延迟的时间
- 数据流中的事件是有序
- 数据流中的事件是无序
三、Flink乱序延迟时间处理-Watermark讲解《下》
案例
- window大小为10s，窗口是W1 [23:12:00~23:12:10) 、 W2[23:12:10~23:12:20)
- 下面是数据的event time
- 数据A 23:12:07
- 数据B 23:12:11
- 数据C 23:12:08
- 数据D 23:12:17
- 数据E 23:12:09
- 没加入watermark,由上到下进入flink
- 数据B到了之后，W1就进行了窗口计算，数据只有A
- 数据C 迟到了3秒，到了之后，由于W1已经计算了，所以就丢失了数据C
- 加入watermark, 允许5秒延迟乱序，由上到下进入flink
- 数据A到达
- watermark = 12:07 - 5 = 12:02 < 12:10 ,所以不触发W1计算, A属于W1
- 数据B到达
- watermark = max{ 12:11, 12:07} - 5 = 12:06 < 12:10 ,所以不触发W1计算, B属于W2
- 数据C到达
- watermark = max{12:08, 12:11, 12:07} - 5 = 12:06 < 12:10 ,所以不触发W1计算, C属于W1
- 数据D到达
- watermark = max{12:17, 12:08, 12:11, 12:07} - 5 = 12:12 > 23:12:10 , 触发W1计算, D属于W2
- 数据E到达
- watermark = max{12:09, 12:17, 12:08, 12:11, 12:07} - 5 = 12:12 > 23:12:10 , 之前已触发W1计算, 所以丢失了E数据，
- 数据A到达
- Watermaker 计算 = 当前计算窗口最大的事件时间 - 允许乱序延迟的时间
- 什么时候触发W1窗口计算
- Watermaker >= Window EndTime窗口结束时间
- 当前计算窗口最大的事件时间 - 允许乱序延迟的时间 >= Window EndTime窗口结束时间
四、Watermark+Window编码实战-分类统计电商订单成交额
这里需要的工具类下面拿
JDK8新版时间工具类（简洁版）——TimeUtil
编写代码
Flink18WatermarkWindowApp
点击查看完整内容
import net.xdclass.util.TimeUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
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.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 java.time.Duration;
import java.util.ArrayList;
import java.util.List;
public class Flink18WatermarkWindowApp {
/**
* source
* transformation
* sink
*
* @param args
*/
public static void main(String[] args) throws Exception {
//构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
//java,2022-11-11 09-10-10,15
DataStream<String> ds = env.socketTextStream("8.142.19.202", 8886);
SingleOutputStreamOperator<Tuple3<String, String, Integer>> flatMapDS = ds.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
@Override
public void flatMap(String value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
String[] arr = value.split(",");
out.collect(Tuple3.of(arr[0], arr[1], Integer.parseInt(arr[2])));
}
});
//指定watermark
SingleOutputStreamOperator<Tuple3<String, String, Integer>> watermarkDS = flatMapDS.assignTimestampsAndWatermarks(WatermarkStrategy
//指定允许乱序延迟的最大时间 3 秒
.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//指定POJO事件时间列，毫秒
.withTimestampAssigner((event, timestamp) -> TimeUtil.strToDate(event.f1).getTime()));
//分组 开窗
SingleOutputStreamOperator<String> sumDS = watermarkDS.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
@Override
public String getKey(Tuple3<String, String, Integer> value) throws Exception {
return value.f0;
}
})
//开窗
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
//聚合， 方便调试拿到窗口全部数据，全窗口函数
.apply(new WindowFunction<Tuple3<String, String, Integer>, String, String, TimeWindow>() {
@Override
public void apply(String key, TimeWindow window, Iterable<Tuple3<String, String, Integer>> input, Collector<String> out) throws Exception {
//准备list，存储窗口的事件时间
List<String> timeList = new ArrayList<>();
int total = 0;
for(Tuple3<String, String, Integer> order:input){
timeList.add(order.f1);
total = total+order.f2;
}
String outStr = String.format("分组key:%s,聚合值:%s,窗口开始结束:[%s~%s),窗口所有事件时间:%s", key,total, TimeUtil.format(window.getStart()),TimeUtil.format(window.getEnd()), timeList);
out.collect(outStr);
}
});
sumDS.print();
env.execute("watermark job");
}
}
五、Flink实战-分类统计电商订单成交额-SocketStream数据测试实战
- 测试数据
- 窗口 [23:12:00 ~ 23:12:10) | [23:12:10 ~ 23:12:20)
- 触发窗口计算条件
- 窗口内有数据
- watermark >= 窗口endtime
- 即 当前计算窗口最大的事件时间 - 允许乱序延迟的时间 >= Window EndTime窗口结束时间
控制台测试
测试数据
java,2022-11-11 23:12:07,10 java,2022-11-11 23:12:11,10 java,2022-11-11 23:12:08,10 mysql,2022-11-11 23:12:13,10 java,2022-11-11 23:12:13,10 java,2022-11-11 23:12:17,10 java,2022-11-11 23:12:09,10 java,2022-11-11 23:12:20,10 java,2022-11-11 23:12:22,10 java,2022-11-11 23:12:23,10
六、Flink 二次兜底延迟数据处理 allowedLateness 更新数据实战
- 背景
- 超过了watermark的等待后，还有延迟数据到达怎么办？
- watermark先输出，然后配置allowedLateness 再延长时间，然后到了后更新之前的窗口数据
关键代码
//允许1分钟延迟
.allowedLateness(Time.minutes(1)) 控制台测试
测试数据
java,2022-11-11 23:12:07,10 java,2022-11-11 23:12:11,10 java,2022-11-11 23:12:08,10 java,2022-11-11 23:12:13,10 java,2022-11-11 23:12:23,10 #延迟1分钟内，所以会输出 java,2022-11-11 23:12:09,10 java,2022-11-11 23:12:02,10 java,2022-11-11 23:14:30,10 #延迟超过1分钟，不会输出 java,2022-11-11 23:12:03,10
七、Flink 最后的兜底延迟数据处理 测输出流实战
背景
- 超过了watermark的等待后，还有延迟数据到达怎么办？
- watermark先输出，然后配置allowedLateness 再延长时间，然后到了后更新之前的窗口数据
- 数据超过了allowedLateness 后，就丢失了吗？用侧输出流 SideOutput
关键代码
//最后的兜底数据
OutputTag<Tuple3<String, String, Integer>> lateData = new OutputTag<Tuple3<String, String, Integer>>("lateDataOrder"){}; //最后的兜底容忍
.sideOutputLateData(lateData) //最后兜底处理，更新之前的数据
sumDS.getSideOutput(lateData).print("late data order"); 关键代码位置图
完整代码放出
Flink18WatermarkWindowApp
点击查看完整内容
import net.xdclass.util.TimeUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
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.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.time.Duration;
import java.util.ArrayList;
import java.util.List;
public class Flink18WatermarkWindowApp {
/**
* source
* transformation
* sink
*
* @param args
*/
public static void main(String[] args) throws Exception {
//构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
//java,2022-11-11 09-10-10,15
DataStream<String> ds = env.socketTextStream("8.142.19.202", 8886);
SingleOutputStreamOperator<Tuple3<String, String, Integer>> flatMapDS = ds.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
@Override
public void flatMap(String value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
String[] arr = value.split(",");
out.collect(Tuple3.of(arr[0], arr[1], Integer.parseInt(arr[2])));
}
});
//指定watermark
SingleOutputStreamOperator<Tuple3<String, String, Integer>> watermarkDS = flatMapDS.assignTimestampsAndWatermarks(WatermarkStrategy
//指定允许乱序延迟的最大时间 3 秒
.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//指定POJO事件时间列，毫秒
.withTimestampAssigner((event, timestamp) -> TimeUtil.strToDate(event.f1).getTime()));
//最后的兜底数据
OutputTag<Tuple3<String, String, Integer>> lateData = new OutputTag<Tuple3<String, String, Integer>>("lateDataOrder"){};
//分组 开窗
SingleOutputStreamOperator<String> sumDS = watermarkDS.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
@Override
public String getKey(Tuple3<String, String, Integer> value) throws Exception {
return value.f0;
}
})
//开窗
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
//允许1分钟延迟
.allowedLateness(Time.minutes(1))
//最后的兜底容忍
.sideOutputLateData(lateData)
//聚合， 方便调试拿到窗口全部数据，全窗口函数
.apply(new WindowFunction<Tuple3<String, String, Integer>, String, String, TimeWindow>() {
@Override
public void apply(String key, TimeWindow window, Iterable<Tuple3<String, String, Integer>> input, Collector<String> out) throws Exception {
//准备list，存储窗口的事件时间
List<String> timeList = new ArrayList<>();
int total = 0;
for(Tuple3<String, String, Integer> order:input){
timeList.add(order.f1);
total = total+order.f2;
}
String outStr = String.format("分组key:%s,聚合值:%s,窗口开始结束:[%s~%s),窗口所有事件时间:%s", key,total, TimeUtil.format(window.getStart()),TimeUtil.format(window.getEnd()), timeList);
out.collect(outStr);
}
});
sumDS.print();
//最后兜底处理，更新之前的数据
sumDS.getSideOutput(lateData).print("late data order");
env.execute("watermark job");
}
}
- 测试数据
- 窗口 [23:12:00 ~ 23:12:10) | [23:12:10 ~ 23: 12:20)
- 触发窗口计算条件
- 窗口内有数据
- watermark >= 窗口endtime
- 即 当前计算窗口最大的事件时间 - 允许乱序延迟的时间 >= Window EndTime窗口结束时间
java,2022-11-11 23:12:07,10 java,2022-11-11 23:12:11,10 java,2022-11-11 23:12:08,10 java,2022-11-11 23:12:13,10 java,2022-11-11 23:12:23,10 #延迟1分钟内，所以会输出 java,2022-11-11 23:12:09,10 java,2022-11-11 23:12:02,10 java,2022-11-11 23:14:30,10 #延迟超过1分钟，不会输出，配置了sideOutPut，会在兜底输出 java,2022-11-11 23:12:03,10 java,2022-11-11 23:12:04,10
控制台效果图
八、【面试题】Flink乱序延迟时间处理-多层保证措施介绍和归纳
- 面试题：如何保证在需要的窗口内获得指定的数据？数据有乱序延迟
- flink采用watermark 、allowedLateness() 、sideOutputLateData()三个机制来保证获取数据
- watermark的作用是防止数据出现延迟乱序，允许等待一会再触发窗口计算，提前输出
- allowLateness，是将窗口关闭时间再延迟一段时间.设置后就像window变大了
- 那么为什么不直接把window设置大一点呢？或者把watermark加大点?
- watermark先输出数据，allowLateness会局部修复数据并主动更新窗口的数据输出
- 这期间的迟到数据不会被丢弃，而是会触发窗口重新计算
- sideOutPut是最后兜底操作，超过allowLateness后，窗口已经彻底关闭了，就会把数据放到侧输出流
- 测输出流 OutputTag tag = new OutputTag(){}, 由于泛型查除问题，需要重写方法，加花括号
- 应用场景：实时监控平台
- 可以用watermark及时输出数据
- allowLateness 做短期的更新迟到数据
- sideOutPut做兜底更新保证数据准确性
- 总结Flink的机制
- 第一层 窗口window 的作用是从DataStream数据流里指定范围获取数据。
- 第二层 watermark的作用是防止数据出现乱序延迟，允许窗口等待延迟数据达到，再触发计算
- 第三层 allowLateness 会让窗口关闭时间再延迟一段时间, 如果还有数据达到，会局部修复数据并主动更新窗口的数据输出
- 第四层 sideOutPut侧输出流是最后兜底操作，在窗口已经彻底关闭后，所有过期延迟数据放到侧输出流，可以单独获取，存储到某个地方再批量更新之前的聚合的数据
- 注意
- Flink 默认的处理方式直接丢弃迟到的数据
- sideOutPut还可以进行分流功能
- DataStream没有getSideOutput方法，SingleOutputStreamOperator才有，
- 版本弃用API
新接口，`WatermarkStrategy`，`TimestampAssigner` 和 `WatermarkGenerator` 因为其对时间戳和 watermark 等重点的抽象和分离很清晰，并且还统一了周期性和标记形式的 watermark 生成方式 新接口之前是用AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks ,现在可以弃用了
本文作者为DBC，转载请注明。
