目录

事件时间和窗口

什么是水位线

1. 有序流中的水位线

2. 乱序流中的水位线

3. 水位线的特性

如何生成水位线

1. 生成水位线的总体原则

2. 水位线生成策略（Watermark Strategies）

3. Flink 内置水位线生成器

（1）有序流

（2）乱序流

4. 自定义水位线策略

（1）周期性水位线生成器（Periodic Generator）

（2）断点式水位线生成器（Punctuated Generator）

5. 在自定义数据源中发送水位线

水位线的传递

水位线的总结

事件时间和窗口

什么是水位线

1. 有序流中的水位线

2. 乱序流中的水位线

public SingleOutputStreamOperator<T> assignTimestampsAndWatermarks(
WatermarkStrategy<T> watermarkStrategy)

DataStream<Event> stream = env.addSource(new ClickSource());
DataStream<Event> withTimestampsAndWatermarks =
stream.assignTimestampsAndWatermarks(<watermark strategy>);

public interface WatermarkStrategy<T>
extends TimestampAssignerSupplier<T>,
WatermarkGeneratorSupplier<T>{
@Override
TimestampAssigner<T>
createTimestampAssigner(TimestampAssignerSupplier.Context context);
@Override
WatermarkGenerator<T>
createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}

3. Flink 内置水位线生成器

（1）有序流

stream.assignTimestampsAndWatermarks(
WatermarkStrategy.<Event>forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssigner<Event>()
{
@Override
public long extractTimestamp(Event element, long recordTimestamp)
{
return element.timestamp;
}
})
);

（2）乱序流

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import java.time.Duration;
public class WatermarkTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env
.addSource(new ClickSource())
// 插入水位线的逻辑
.assignTimestampsAndWatermarks(
// 针对乱序流插入水位线，延迟时间设置为 5sWatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner(new
SerializableTimestampAssigner<Event>() {
// 抽取时间戳的逻辑
@Override
public long extractTimestamp(Event element, long
recordTimestamp) {
return element.timestamp;
}
})
)
.print();
env.execute();
}
}

WatermarkStrategy.forMonotonousTimestamps()

WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(0))

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
// 自定义水位线的产生
public class CustomWatermarkTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);env.addSource(new ClickSource()).assignTimestampsAndWatermarks(new CustomWatermarkStrategy()).print();env.execute();}public static class CustomWatermarkStrategy implements
WatermarkStrategy<Event> {@Overridepublic TimestampAssigner<Event>
createTimestampAssigner(TimestampAssignerSupplier.Context context) {return new SerializableTimestampAssigner<Event>() {@Overridepublic long extractTimestamp(Event element, long recordTimestamp)
{return element.timestamp; // 告诉程序数据源里的时间戳是哪一个字段}};}@Overridepublic WatermarkGenerator<Event>
createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {return new CustomPeriodicGenerator();}}public static class CustomPeriodicGenerator implements
WatermarkGenerator<Event> {private Long delayTime = 5000L; // 延迟时间private Long maxTs = Long.MIN_VALUE + delayTime + 1L; // 观察到的最大时间戳@Overridepublic void onEvent(Event event, long eventTimestamp, WatermarkOutput
output) {// 每来一条数据就调用一次maxTs = Math.max(event.timestamp, maxTs); // 更新最大时间戳}@Overridepublic void onPeriodicEmit(WatermarkOutput output) {// 发射水位线，默认 200ms 调用一次output.emitWatermark(new Watermark(maxTs - delayTime - 1L));}}
}

public class CustomPunctuatedGenerator implements WatermarkGenerator<Event> {@Overridepublic void onEvent(Event r, long eventTimestamp, WatermarkOutput output) {
// 只有在遇到特定的 itemId 时，才发出水位线if (r.user.equals("Mary")) {output.emitWatermark(new Watermark(r.timestamp - 1));}}@Overridepublic void onPeriodicEmit(WatermarkOutput output) {// 不需要做任何事情，因为我们在 onEvent 方法中发射了水位线}
}

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import java.util.Calendar;
import java.util.Random;
public class EmitWatermarkInSourceFunction {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);env.addSource(new ClickSourceWithWatermark()).print();env.execute();}// 泛型是数据源中的类型public static class ClickSourceWithWatermark implements SourceFunction<Event>
{private boolean running = true;@Overridepublic void run(SourceContext<Event> sourceContext) throws Exception {Random random = new Random();String[] userArr = {"Mary", "Bob", "Alice"};String[] urlArr = {"./home", "./cart", "./prod?id=1"};while (running) {long currTs = Calendar.getInstance().getTimeInMillis(); // 毫秒时
间戳String username = userArr[random.nextInt(userArr.length)];String url = urlArr[random.nextInt(urlArr.length)];Event event = new Event(username, url, currTs);// 使用 collectWithTimestamp 方法将数据发送出去，并指明数据中的时间戳的字段sourceContext.collectWithTimestamp(event, event.timestamp);// 发送水位线sourceContext.emitWatermark(new Watermark(event.timestamp - 1L));Thread.sleep(1000L);}}@Overridepublic void cancel() {running = false;}}
}

Flink中水位线Watermark相关推荐

1. 关于Flink中time与watermark的理解

   关于Flink中time与watermark的理解 空间和时间只是我们人类思考的模式,并不是我们赖以生存的条件. --爱因斯坦 时间属性是流处理中最重要的一个方面,是流处理系统的基石之一:流计算是一种 ...

2. Flink 合流水位线图解

   1.合流发送最小的水位线 2 2.4覆盖后,发送最小的3 3.7覆盖后,发送最小的3 4.6覆盖后,发送最小的4

3. 4.1.19 Flink-流处理框架-Flink中的时间语义和watermark水位线

   目录 1.Flink中的时间语义 1.1 EventTime 的代码设置 2.Watermark水位线 2.1 watermark的基本概念 2.2 watermark的特点和传递 2.3 Water ...

4. 一文理解Flink 水位线(Flink Watermark）

   文章目录 Flink 中的时间语义 `处理时间` `事件时间` 水位线(Watermark) `事件时间和窗口` `什么是水位线` 有序流中的水位线 乱序流中的水位线 `水位线的特性` `如何生成水位 ...

5. Flink 中的时间和窗口

   时间和窗口 一.时间语义 1. Flink 中的时间语义 1.1 处理时间(Processing Time) 1.2 事件时间(Event Time) 1.3 两种时间语义的对比 二.水位线(Wate ...

6. Flink-时间和窗口(水位线、窗口、迟到数据的处理等)

   文章目录 时间和窗口 时间 水位线(Watermark) 时间和窗口 水位线 有序和无序流的插入 水位线生成策略(Watermark Strategies) 水位线的传递 窗口(Window) 窗口 ...

7. 如何在 Apache Flink 中使用 Python API？

   本文根据 Apache Flink 系列直播课程整理而成,由 Apache Flink PMC,阿里巴巴高级技术专家 孙金城 分享.重点为大家介绍 Flink Python API 的现状及未来规划, ...

8. python flink_如何在 Apache Flink 中使用 Python API？

   原标题:如何在 Apache Flink 中使用 Python API? 导读:本文重点为大家介绍 Flink Python API 的现状及未来规划,主要内容包括:Apache Flink Pyth ...

9. Flink Pre-defined Timestamp Extractors / Watermark Emitters（预定义的时间戳提取/水位线发射器）...

   https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/event_timestamp_extractors.html 根据官网 ...

10. Flink Watermark相关概念（窗口、水位线、迟到事件）

    Flink 为实时计算提供了三种时间,即事件时间(event time).摄入时间(ingestion time)和处理时间(processing time).在进行 window 计算时,使用摄入时 ...

最新文章

1. show在php,show.php
2. iOS 设置View阴影
3. m个足球放入n个篮子中或者放苹果问题
4. GeoServer 数据导出
5. 员工工号怎么编码_华为员工感慨：工号就留在这了，感谢公司给我自己写墓志铭的机会...
6. react中axios的二次封装
7. stackoverflow图片大小修改
8. AppScan安装教程
9. 常见网络问题——架设篇
10. 短视频源码应该优化的六个方面
11. Qt Moc及信号-槽源代码解析
12. 网站SEO优化之图片优化方法
13. postman使用方法
14. netlink怎么读_跑长途的时候如何不寂寞，车机互联G-Netlink全解说明
15. CSS基础（8）- 盒模型应用
16. IT行业分析之嵌入式应用
17. 股神巴菲特忠告中国股民及巴菲特忠告中国股民把控风险
18. 软件测试工具之——selenium-IDE下载安装与该附件组件已无法安装，因为它似乎已损坏问题解决
19. 远程桌面连接的开启，和借助内外网软件，实现外网PC手机远程控制内网电脑，详细图文教程
20. 第三天 本来是吃吃玩玩愉快的一天， Airpods耳机充电仓丢了 ，淦 ➡_➡！但学习还得继续！！

热门文章

1. 安装Ubuntu后必须要做的几件事(一)--基础应用篇
2. thrift0.8.0支持win7的方法
3. mac电脑双开微信的方法
4. [转]PT与PX区别
5. 视频插帧论文阅读（一）——FLAVR
6. Word文字的三种隐藏方法
7. 计算机的优势和劣势_计算机专业毕业生考研还是就业应该怎么选？
8. python中函数调用
9. iphone双重认证关闭不了怎么办_iPhone双重认证怎么关闭？苹果手机关闭双重认证的两种方法[多图]...
10. Python chain函数的用法