当前位置：网站首页>Flink EventTime和Watermarks案例分析
Flink EventTime和Watermarks案例分析
2022-08-05 05:14:00 【bigdata1024】
目录
4.2：allowedLateness 指定允许数据延迟的时间
4.3：sideOutputLateData 收集迟到的数据
针对数据乱序需求的案例分析，需要使用eventtime和watermark来解决
解释：
watermarks的生成方式有两种
1：With Periodic Watermarks：周期性的触发watermark的生成和发送
2：With Punctuated Watermarks：基于某些事件触发watermark的生成和发送
第一种方式比较常用，所以在这里我们使用第一种方式进行分析。
参考官网文档中With Periodic Watermarks的使用方法
代码中的extractTimestamp方法是从数据本身中提取EventTime
getCurrentWatermar方法是获取当前水位线，利用currentMaxTimestamp - maxOutOfOrderness
这里的maxOutOfOrderness表示是允许数据的最大乱序时间
所以在这里我们使用的话也实现接口AssignerWithPeriodicWatermarks。
1：实现watermark相关代码
1.1：程序说明
从socket模拟接收数据，然后使用map进行处理，后面再调用assignTimestampsAndWatermarks方法抽取timestamp并生成watermark。最后再调用window打印信息来验证window被触发的时机。
1.2：代码如下
完整代码如下：
package xuwei.tech.streaming.streamApiDemo;
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.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
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 javax.annotation.Nullable;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
/**
*
* Watermark 案例
*
* Created by xuwei.tech.
*/
public class StreamingWindowWatermark {
public static void main(String[] args) throws Exception {
//定义socket的端口号
int port = 9000;
//获取运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//设置使用eventtime，默认是使用processtime
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//设置并行度为1,默认并行度是当前机器的cpu数量
env.setParallelism(1);
//连接socket获取输入的数据
DataStream<String> text = env.socketTextStream("hadoop100", port, "\n");
//解析输入的数据
DataStream<Tuple2<String, Long>> inputMap = text.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String value) throws Exception {
String[] arr = value.split(",");
return new Tuple2<>(arr[0], Long.parseLong(arr[1]));
}
});
//抽取timestamp和生成watermark
DataStream<Tuple2<String, Long>> waterMarkStream = inputMap.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {
Long currentMaxTimestamp = 0L;
final Long maxOutOfOrderness = 10000L;// 最大允许的乱序时间是10s
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
/**
* 定义生成watermark的逻辑
* 默认100ms被调用一次
*/
@Nullable
@Override
public Watermark getCurrentWatermark() {
return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
}
//定义如何提取timestamp
@Override
public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
long timestamp = element.f1;
currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
System.out.println("key:"+element.f0+",eventtime:["+element.f1+"|"+sdf.format(element.f1)+"],currentMaxTimestamp:["+currentMaxTimestamp+"|"+
sdf.format(currentMaxTimestamp)+"],watermark:["+getCurrentWatermark().getTimestamp()+"|"+sdf.format(getCurrentWatermark().getTimestamp())+"]");
return timestamp;
}
});
//分组，聚合
DataStream<String> window = waterMarkStream.keyBy(0)
.window(TumblingEventTimeWindows.of(Time.seconds(3)))//按照消息的EventTime分配窗口，和调用TimeWindow效果一样
.apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
/**
* 对window内的数据进行排序，保证数据的顺序
* @param tuple
* @param window
* @param input
* @param out
* @throws Exception
*/
@Override
public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
String key = tuple.toString();
List<Long> arrarList = new ArrayList<Long>();
Iterator<Tuple2<String, Long>> it = input.iterator();
while (it.hasNext()) {
Tuple2<String, Long> next = it.next();
arrarList.add(next.f1);
}
Collections.sort(arrarList);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
String result = key + "," + arrarList.size() + "," + sdf.format(arrarList.get(0)) + "," + sdf.format(arrarList.get(arrarList.size() - 1))
+ "," + sdf.format(window.getStart()) + "," + sdf.format(window.getEnd());
out.collect(result);
}
});
//测试-把结果打印到控制台即可
window.print();
//注意：因为flink是懒加载的，所以必须调用execute方法，上面的代码才会执行
env.execute("eventtime-watermark");
}
}
1.3：程序详解
- 接收socket数据
- 将每行数据按照逗号分隔，每行数据调用map转换成tuple<String,Long>类型。其中tuple中的第一个元素代表具体的数据，第二行代表数据的eventtime
- 抽取timestamp，生成watermar，允许的最大乱序时间是10s，并打印（key,eventtime,currentMaxTimestamp,watermark）等信息
- 分组聚合，window窗口大小为3秒，输出（key，窗口内元素个数，窗口内最早元素的时间，窗口内最晚元素的时间，窗口自身开始时间，窗口自身结束时间）
2：通过数据跟踪watermark的时间
在这里重点查看watermark和timestamp的时间，通过数据的输出来确定window的触发时机
首先我们开启socker，输入第一条数据
输出的内容如下：
为了查看方便，我们把输入内容汇总到表格中
Key | Event Time | CurrentMaxTimeStamp | WaterMark |
0001 | 1538359882000 | 1538359882000 | 1538359872000 |
2018-10-01 10:11:22.000 | 2018-10-01 10:11:22.000 | 2018-10-01 10:11:12.000 |
此时，wartermark的时间，已经落后于currentMaxTimestamp10秒了。我们继续输入
此时，输入内容如下：
我们再次汇总，如下表：
Key | Event Time | CurrentMaxTimeStamp | WaterMark |
0001 | 1538359882000 | 1538359882000 | 1538359872000 |
2018-10-01 10:11:22.000 | 2018-10-01 10:11:22.000 | 2018-10-01 10:11:12.000 | |
0001 | 1538359886000 | 1538359886000 | 1538359876000 |
2018-10-01 10:11:26.000 | 2018-10-01 10:11:26.000 | 2018-10-01 10:11:16.000 |
继续输入：
输出内容如下：
汇总如下：
Key | Event Time | CurrentMaxTimeStamp | WaterMark |
0001 | 1538359882000 | 1538359882000 | 1538359872000 |
2018-10-01 10:11:22.000 | 2018-10-01 10:11:22.000 | 2018-10-01 10:11:12.000 | |
0001 | 1538359886000 | 1538359886000 | 1538359876000 |
2018-10-01 10:11:26.000 | 2018-10-01 10:11:26.000 | 2018-10-01 10:11:16.000 | |
0001 | 1538359892000 | 1538359892000 | 1538359882000 |
2018-10-01 10:11:32.000 | 2018-10-01 10:11:32.000 | 2018-10-01 10:11:22.000 |
到这里，window仍然没有被触发，此时watermark的时间已经等于了第一条数据的Event Time了。那么window到底什么时候被触发呢？
我们再次输入：
输出内容如下：
汇总如下：
Key | Event Time | CurrentMaxTimeStamp | WaterMark |
0001 | 1538359882000 | 1538359882000 | 1538359872000 |
2018-10-01 10:11:22.000 | 2018-10-01 10:11:22.000 | 2018-10-01 10:11:12.000 | |
0001 | 1538359886000 | 1538359886000 | 1538359876000 |
2018-10-01 10:11:26.000 | 2018-10-01 10:11:26.000 | 2018-10-01 10:11:16.000 | |
0001 | 1538359892000 | 1538359892000 | 1538359882000 |
2018-10-01 10:11:32.000 | 2018-10-01 10:11:32.000 | 2018-10-01 10:11:22.000 | |
0001 | 1538359893000 | 1538359893000 | 1538359883000 |
2018-10-01 10:11:33.000 | 2018-10-01 10:11:33.000 | 2018-10-01 10:11:23.000 |
window仍然没有触发，此时，我们的数据已经发到2018-10-01 10:11:33.000了，根据eventtime来算，最早的数据已经过去了11秒了，window还没有开始计算，那到底什么时候会触发window呢？
我们再次增加1秒，输入：
输出：
汇总如下：
Key | Event Time | CurrentMaxTimeStamp | WaterMark | window_start_time | window_end_time |
0001 | 1538359882000 | 1538359882000 | 1538359872000 |
|
|
2018-10-01 10:11:22.000 | 2018-10-01 10:11:22.000 | 2018-10-01 10:11:12.000 |
|
| |
0001 | 1538359886000 | 1538359886000 | 1538359876000 |
|
|
2018-10-01 10:11:26.000 | 2018-10-01 10:11:26.000 | 2018-10-01 10:11:16.000 |
|
| |
0001 | 1538359892000 | 1538359892000 | 1538359882000 |
|
|
2018-10-01 10:11:32.000 | 2018-10-01 10:11:32.000 | 2018-10-01 10:11:22.000 |
|
| |
0001 | 1538359893000 | 1538359893000 | 1538359883000 |
|
|
2018-10-01 10:11:33.000 | 2018-10-01 10:11:33.000 | 2018-10-01 10:11:23.000 |
|
| |
0001 | 1538359894000 | 1538359894000 | 1538359884000 |
|
|
2018-10-01 10:11:34.000 | 2018-10-01 10:11:34.000 | 2018-10-01 10:11:24.000 | [10:11:21.000 | 10:11:24.000) |
到这里，我们做一个说明：
window的触发机制，是先按照自然时间将window划分，如果window大小是3秒，那么1分钟内会把window划分为如下的形式【左闭右开】:
[00:00:00,00:00:03)
[00:00:03,00:00:06)
[00:00:06,00:00:09)
[00:00:09,00:00:12)
[00:00:12,00:00:15)
[00:00:15,00:00:18)
[00:00:18,00:00:21)
[00:00:21,00:00:24)
[00:00:24,00:00:27)
[00:00:27,00:00:30)
[00:00:30,00:00:33)
[00:00:33,00:00:36)
[00:00:36,00:00:39)
[00:00:39,00:00:42)
[00:00:42,00:00:45)
[00:00:45,00:00:48)
[00:00:48,00:00:51)
[00:00:51,00:00:54)
[00:00:54,00:00:57)
[00:00:57,00:01:00)
...window的设定无关数据本身，而是系统定义好了的。
输入的数据中，根据自身的Event Time，将数据划分到不同的window中，如果window中有数据，则当watermark时间>=Event Time时，就符合了window触发的条件了，最终决定window触发，还是由数据本身的Event Time所属的window中的window_end_time决定。
上面的测试中，最后一条数据到达后，其水位线已经升至10:11:24秒，正好是最早的一条记录所在window的window_end_time，所以window就被触发了。
为了验证window的触发机制，我们继续输入数据:
输出：
汇总如下：
Key | Event Time | CurrentMaxTimeStamp | WaterMark | window_start_time | window_end_time |
0001 | 1538359882000 | 1538359882000 | 1538359872000 |
|
|
2018-10-01 10:11:22.000 | 2018-10-01 10:11:22.000 | 2018-10-01 10:11:12.000 |
|
| |
0001 | 1538359886000 | 1538359886000 | 1538359876000 |
|
|
2018-10-01 10:11:26.000 | 2018-10-01 10:11:26.000 | 2018-10-01 10:11:16.000 |
|
| |
0001 | 1538359892000 | 1538359892000 | 1538359882000 |
|
|
2018-10-01 10:11:32.000 | 2018-10-01 10:11:32.000 | 2018-10-01 10:11:22.000 |
|
| |
0001 | 1538359893000 | 1538359893000 | 1538359883000 |
|
|
2018-10-01 10:11:33.000 | 2018-10-01 10:11:33.000 | 2018-10-01 10:11:23.000 |
|
| |
0001 | 1538359894000 | 1538359894000 | 1538359884000 |
|
|
2018-10-01 10:11:34.000 | 2018-10-01 10:11:34.000 | 2018-10-01 10:11:24.000 | [10:11:21.000 | 10:11:24.000) | |
0001 | 1538359896000 | 1538359896000 | 1538359886000 |
|
|
2018-10-01 10:11:36.000 | 2018-10-01 10:11:36.000 | 2018-10-01 10:11:26.000 |
|
|
此时，watermark时间虽然已经达到了第二条数据的时间，但是由于其没有达到第二条数据所在window的结束时间，所以window并没有被触发。那么，第二条数据所在的window时间是:
[00:00:24,00:00:27)也就是说，我们必须输入一个10:11:27秒的数据，第二条数据所在的window才会被触发。我们继续输入：
输出：
汇总如下：
Key | Event Time | CurrentMaxTimeStamp | WaterMark | window_start_time | window_end_time |
0001 | 1538359882000 | 1538359882000 | 1538359872000 |
|
|
2018-10-01 10:11:22.000 | 2018-10-01 10:11:22.000 | 2018-10-01 10:11:12.000 |
|
| |
0001 | 1538359886000 | 1538359886000 | 1538359876000 |
|
|
2018-10-01 10:11:26.000 | 2018-10-01 10:11:26.000 | 2018-10-01 10:11:16.000 |
|
| |
0001 | 1538359892000 | 1538359892000 | 1538359882000 |
|
|
2018-10-01 10:11:32.000 | 2018-10-01 10:11:32.000 | 2018-10-01 10:11:22.000 |
|
| |
0001 | 1538359893000 | 1538359893000 | 1538359883000 |
|
|
2018-10-01 10:11:33.000 | 2018-10-01 10:11:33.000 | 2018-10-01 10:11:23.000 |
|
| |
0001 | 1538359894000 | 1538359894000 | 1538359884000 |
|
|
2018-10-01 10:11:34.000 | 2018-10-01 10:11:34.000 | 2018-10-01 10:11:24.000 | [10:11:21.000 | 10:11:24.000) | |
0001 | 1538359896000 | 1538359896000 | 1538359886000 |
|
|
2018-10-01 10:11:36.000 | 2018-10-01 10:11:36.000 | 2018-10-01 10:11:26.000 |
|
| |
0001 | 1538359897000 | 1538359897000 | 1538359887000 |
|
|
2018-10-01 10:11:37.000 | 2018-10-01 10:11:37.000 | 2018-10-01 10:11:27.000 | [10:11:24.000 | 10:11:27.000) |
此时，我们已经看到，window的触发要符合以下几个条件：
1、watermark时间 >= window_end_time
2、在[window_start_time,window_end_time)区间中有数据存在，注意是左闭右开的区间同时满足了以上2个条件，window才会触发。
3：watermark+window处理乱序数据
我们上面的测试，数据都是按照时间顺序递增的，现在，我们输入一些乱序的（late）数据，看看watermark结合window机制，是如何处理乱序的。
输入两行数据：
输出：
汇总如下：
Key | Event Time | CurrentMaxTimeStamp | WaterMark | window_start_time | window_end_time |
0001 | 1538359882000 | 1538359882000 | 1538359872000 |
|
|
2018-10-01 10:11:22.000 | 2018-10-01 10:11:22.000 | 2018-10-01 10:11:12.000 |
|
| |
0001 | 1538359886000 | 1538359886000 | 1538359876000 |
|
|
2018-10-01 10:11:26.000 | 2018-10-01 10:11:26.000 | 2018-10-01 10:11:16.000 |
|
| |
0001 | 1538359892000 | 1538359892000 | 1538359882000 |
|
|
2018-10-01 10:11:32.000 | 2018-10-01 10:11:32.000 | 2018-10-01 10:11:22.000 |
|
| |
0001 | 1538359893000 | 1538359893000 | 1538359883000 |
|
|
2018-10-01 10:11:33.000 | 2018-10-01 10:11:33.000 | 2018-10-01 10:11:23.000 |
|
| |
0001 | 1538359894000 | 1538359894000 | 1538359884000 |
|
|
2018-10-01 10:11:34.000 | 2018-10-01 10:11:34.000 | 2018-10-01 10:11:24.000 | [10:11:21.000 | 10:11:24.000) | |
0001 | 1538359896000 | 1538359896000 | 1538359886000 |
|
|
2018-10-01 10:11:36.000 | 2018-10-01 10:11:36.000 | 2018-10-01 10:11:26.000 |
|
| |
0001 | 1538359897000 | 1538359897000 | 1538359887000 |
|
|
2018-10-01 10:11:37.000 | 2018-10-01 10:11:37.000 | 2018-10-01 10:11:27.000 | [10:11:24.000 | 10:11:27.000) | |
0001 | 1538359899000 | 1538359899000 | 1538359889000 |
|
|
2018-10-01 10:11:39.000 | 2018-10-01 10:11:39.000 | 2018-10-01 10:11:29.000 |
|
| |
0001 | 1538359891000 | 1538359899000 | 1538359889000 |
|
|
2018-10-01 10:11:31.000 | 2018-10-01 10:11:39.000 | 2018-10-01 10:11:29.000 |
|
|
可以看到，虽然我们输入了一个10:11:31的数据，但是currentMaxTimestamp和watermark都没变。此时，按照我们上面提到的公式：
1、watermark时间 >= window_end_time
2、在[window_start_time,window_end_time)中有数据存在watermark时间（10:11:29） < window_end_time（10:11:33），因此不能触发window。
那如果我们再次输入一条10:11:43的数据，此时watermark时间会升高到10:11:33，这时的window一定就会触发了，我们试一试：
输入：
输出：
汇总如下：
Key | Event Time | CurrentMaxTimeStamp | WaterMark | window_start_time | window_end_time |
0001 | 1538359882000 | 1538359882000 | 1538359872000 |
|
|
2018-10-01 10:11:22.000 | 2018-10-01 10:11:22.000 | 2018-10-01 10:11:12.000 |
|
| |
0001 | 1538359886000 | 1538359886000 | 1538359876000 |
|
|
2018-10-01 10:11:26.000 | 2018-10-01 10:11:26.000 | 2018-10-01 10:11:16.000 |
|
| |
0001 | 1538359892000 | 1538359892000 | 1538359882000 |
|
|
2018-10-01 10:11:32.000 | 2018-10-01 10:11:32.000 | 2018-10-01 10:11:22.000 |
|
| |
0001 | 1538359893000 | 1538359893000 | 1538359883000 |
|
|
2018-10-01 10:11:33.000 | 2018-10-01 10:11:33.000 | 2018-10-01 10:11:23.000 |
|
| |
0001 | 1538359894000 | 1538359894000 | 1538359884000 |
|
|
2018-10-01 10:11:34.000 | 2018-10-01 10:11:34.000 | 2018-10-01 10:11:24.000 | [10:11:21.000 | 10:11:24.000) | |
0001 | 1538359896000 | 1538359896000 | 1538359886000 |
|
|
2018-10-01 10:11:36.000 | 2018-10-01 10:11:36.000 | 2018-10-01 10:11:26.000 |
|
| |
0001 | 1538359897000 | 1538359897000 | 1538359887000 |
|
|
2018-10-01 10:11:37.000 | 2018-10-01 10:11:37.000 | 2018-10-01 10:11:27.000 | [10:11:24.000 | 10:11:27.000) | |
0001 | 1538359899000 | 1538359899000 | 1538359889000 |
|
|
2018-10-01 10:11:39.000 | 2018-10-01 10:11:39.000 | 2018-10-01 10:11:29.000 |
|
| |
0001 | 1538359891000 | 1538359899000 | 1538359889000 |
|
|
2018-10-01 10:11:31.000 | 2018-10-01 10:11:39.000 | 2018-10-01 10:11:29.000 |
|
| |
0001 | 1538359903000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:43.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:30.000 | 10:11:33.000) |
这里，我么看到，窗口中有2个数据，10:11:31和10:11:32，但是没有10:11:33的数据，原因是窗口是一个前闭后开的区间，10:11:33的数据是属于[10:11:33,10:11:36)的窗口的。
Flink应该如何设置最大乱序时间 ?
这个要结合自己的业务以及数据情况去设置。如果maxOutOfOrderness设置的太小，而自身数据发送时由于网络等原因导致乱序或者late太多，那么最终的结果就是会有很多单条的数据在window中被触发，数据的正确性影响太大
对于严重乱序的数据，需要严格统计数据最大延迟时间，才能保证计算的数据准确，延时设置太小会影响数据准确性，延时设置太大不仅影响数据的实时性，更加会加重Flink作业的负担，不是对eventTime要求特别严格的数据，尽量不要采用eventTime方式来处理，会有丢数据的风险。
上边的结果，已经表明，对于out-of-order的数据，Flink可以通过watermark机制结合window的操作，来处理一定范围内的乱序数据。那么对于“迟到(late element)”太多的数据，Flink是怎么处理的呢？
4：late element(延迟数据)的处理
延迟数据的三种处理方案
4.1：丢弃(默认)
我们输入一个乱序很多的（其实只要Event Time < watermark时间）数据来测试下：
输入：【输入两条内容】
输出：
汇总如下：
Key | Event Time | CurrentMaxTimeStamp | WaterMark | window_start_time | window_end_time |
0001 | 1538359890000 | 1538359890000 | 1538359880000 |
|
|
2018-10-01 10:11:30.000 | 2018-10-01 10:11:30.000 | 2018-10-01 10:11:20.000 |
|
| |
0001 | 1538359903000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:43.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) |
注意：此时watermark是2018-10-01 10:11:33.000
下面我们再输入几个eventtime小于watermark的时间
输入：【输入了三行内容】
输出：
注意：此时并没有触发window。因为输入的数据所在的窗口已经执行过了，flink默认对这些迟到的数据的处理方案就是丢弃。
4.2：allowedLateness 指定允许数据延迟的时间
在某些情况下，我们希望对迟到的数据再提供一个宽容的时间。
Flink提供了allowedLateness方法可以实现对迟到的数据设置一个延迟时间，在指定延迟时间内到达的数据还是可以触发window执行的。
修改代码：
下面我们来验证一下：
输入：【输入两行内容】
输出：
正常触发window，没什么问题。
汇总：
Key | Event Time | CurrentMaxTimeStamp | WaterMark | window_start_time | window_end_time |
0001 | 1538359890000 | 1538359890000 | 1538359880000 |
|
|
2018-10-01 10:11:30.000 | 2018-10-01 10:11:30.000 | 2018-10-01 10:11:20.000 |
|
| |
0001 | 1538359903000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:43.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) |
此时watermark是2018-10-01 10:11:33.000
那么现在我们输入几条eventtime<watermark的数据验证一下效果
输入：【输入三行内容】
输出：
在这里看到每条数据都触发了window执行。
汇总：
Key | Event Time | CurrentMaxTimeStamp | WaterMark | window_start_time | window_end_time |
0001 | 1538359890000 | 1538359890000 | 1538359880000 |
|
|
2018-10-01 10:11:30.000 | 2018-10-01 10:11:30.000 | 2018-10-01 10:11:20.000 |
|
| |
0001 | 1538359903000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:43.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) | |
0001 | 1538359890000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:30.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) | |
0001 | 1538359891000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:31.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) | |
0001 | 1538359892000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:32.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) |
我们再输入一条数据，把water调整到10:11:34
输入：
输出：
汇总如下：
Key | Event Time | CurrentMaxTimeStamp | WaterMark | window_start_time | window_end_time |
0001 | 1538359890000 | 1538359890000 | 1538359880000 |
|
|
2018-10-01 10:11:30.000 | 2018-10-01 10:11:30.000 | 2018-10-01 10:11:20.000 |
|
| |
0001 | 1538359903000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:43.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) | |
0001 | 1538359890000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:30.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) | |
0001 | 1538359891000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:31.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) | |
0001 | 1538359892000 | 1538359903000 | 1538359893000 |
|
|
2018-10-01 10:11:32.000 | 2018-10-01 10:11:43.000 | 2018-10-01 10:11:33.000 | [10:11:33.000 | 10:11:33.000) | |
0001 | 1538359904000 | 1538359904000 | 1538359894000 |
|
|
2018-10-01 10:11:44.000 | 2018-10-01 10:11:44.000 | 2018-10-01 10:11:34.000 |
|
|
此时，把water上升到了10:11:34，我们再输入几条eventtime<watermark的数据验证一下效果
输入：
输出：
发现输入的三行数据都触发了window的执行。
我们再输入一条数据，把water调整到10:11:35
输入：
输出：
此时，watermark上升到了10:11:35
我们再输入几条eventtime<watermark的数据验证一下效果
输入：
输出：
发现这几条数据都没有触发window。
分析：
当watemark等于10:11:33的时候，正好是window_end_time，所以会触发[10:11:30~10:11:33) 的window执行。
当窗口执行过后，我们输入[10:11:30~10:11:33) window内的数据会发现window是可以被触发的。
当watemark提升到10:11:34的时候，我们输入[10:11:30~10:11:33)window内的数据会发现window也是可以被触发的。
当watemark提升到10:11:35的时候，我们输入[10:11:30~10:11:33)window内的数据会发现window不会被触发了。
由于我们在前面设置了allowedLateness(Time.seconds(2))，可以允许延迟在2s内的数据继续触发window执行。
所以当watermark是10:11:34的时候可以触发window，但是10:11:35的时候就不行了。
总结：
对于此窗口而言，允许2秒的迟到数据，即第一次触发是在watermark >=window_end_time时
第二次（或多次）触发的条件是watermark < window_end_time + allowedLateness时间内，这个窗口有late数据到达时。
解释：
当watermark等于10:11:34的时候，我们输入eventtime为10:11:30、10:11:31、10:11:32的数据的时候，是可以触发的，因为这些数据的window_end_time都是10:11:33，也就是10:11:34<10:11:33+2 为true。
但是当watermark等于10:11:35的时候，我们再输入eventtime为10:11:30、10:11:31、10:11:32的数据的时候，这些数据的window_end_time都是10:11:33，此时，10:11:35<10:11:33+2 为false了。所以最终这些数据迟到的时间太久了，就不会再触发window执行了。
4.3：sideOutputLateData 收集迟到的数据
通过sideOutputLateData 可以把迟到的数据统一收集，统计存储，方便后期排查问题。
需要先调整代码：
我们来输入一些数据验证一下
输入：
输出：
此时，window被触发执行了，此时watermark是10:11:33
下面我们再输入几个eventtime小于watermark的数据测试一下
输入：
输出：
此时，针对这几条迟到的数据，都通过sideOutputLateData保存到了outputTag中。
更多资源和问题请加群沟通学习：
获取更多大数据资料，视频以及技术交流请加群：
边栏推荐
- Convert the paper official seal in the form of a photo into an electronic official seal (no need to download ps)
- day10-字符串作业
- Xiaobai, you big bulls are lightly abused
- SQL(二) —— join窗口函数视图
- redis persistence
- BFC（Block Formatting Context）
- Machine Learning (2) - Machine Learning Fundamentals
- [Go through 7] Notes from the first section of the fully connected neural network video
- day9-字符串作业
- Flink Table API 和 SQL之概述
猜你喜欢
DOM及其应用
![[Student Graduation Project] Design and Implementation of the Website Based on the Web Student Information Management System (13 pages)](/img/86/9c9a2541f2b7089ae47e9832fffdb3.png)
[Student Graduation Project] Design and Implementation of the Website Based on the Web Student Information Management System (13 pages)
Basic properties of binary tree + oj problem analysis
【过一下7】全连接神经网络视频第一节的笔记
DOM and its applications
软件设计 实验四 桥接模式实验
Difference between for..in and for..of
flink中文文档-目录v1.4
![coppercam入门手册[6]](/img/d3/a7d44aa19acfb18c5a8cacdc8176e9.png)
coppercam入门手册[6]
Flink Table API 和 SQL之概述
随机推荐
第二讲 Linear Model 线性模型
Pycharm中使用pip安装第三方库安装失败:“Non-zero exit code (2)“的解决方法
【记一下1】2022年6月29日 哥和弟 双重痛苦
【NFT开发】设计师无技术基础保姆级开发NFT教程在Opensea上全套开发一个NFT项目+构建Web3网站
物理层的接口有哪几个方面的特性？各包含些什么内容？
Pandas（五）—— 分类数据、读取数据库
Returned object not currently part of this pool
【Over 16】Looking back at July
【Reading】Long-term update
Flink Distributed Cache 分布式缓存
LeetCode: 1403. Minimum subsequence in non-increasing order [greedy]
第四讲 back propagation 反向传播
[Go through 10] sklearn usage record
[Decoding tools] Some online tools for Bitcoin
DOM及其应用
el-table,el-table-column,selection，获取多选选中的数据
el-pagination左右箭头替换成文字上一页和下一页
Matplotlib（二）—— 子图
Do you use tomatoes to supervise your peers?Add my study room, come on together
RDD和DataFrame和Dataset