详解 Flink 的时间语义和 watermark

2024-06-08 20:12
文章标签 详解 时间 flink 语义 watermark

本文主要是介绍详解 Flink 的时间语义和 watermark，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、Flink 时间语义类型

在这里插入图片描述

  • Event Time：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink 通过时间戳分配器访问事件时间戳
  • Ingestion Time ：是数据进入 Flink 的时间
  • Processing Time：是每一个执行基于时间操作的算子的本地系统时间，与机器相关，默认的时间属性就是 Processing Time

二、EventTime 引入

Flink 默认是按照 ProcessingTime 来处理数据的

/**在 Flink 的流式处理中，绝大部分情况推荐使用 eventTime，一般只在 eventTime 无法使用时，才会被迫使用 ProcessingTime 或者 Ing estionTime 。使用 EventTime ，需要先引入 EventTime 的时间属性
*/
public class EventTimeTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//引入 EvenetTime//TimeCharacteristic 是一个枚举类，有 ProcessingTime、IngestionTime 和 EventTime 三个属性env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);}
}

三、Watermark

1. 数据乱序情况

在这里插入图片描述

  • 正常情况下，Flink 接收到的事件应该要是按照事件的产生时间 (EventTime) 的先后顺序排列的
  • 实际情况下，事件从产生到进入 source 再到触发 operator，其中间是有一个过程和时间的，而且由于网络、分布式等原因会造成 Flink 接收到的事件的先后顺序不是严格按照事件的 EventTime 顺序排列的，即所谓的乱序数据
  • 乱序数据的问题会造成窗口触发关闭的时间混乱，计算不准确
  • Flink 处理乱序数据的机制：Watermark + allowedLateness + sideOutputLateData

2. Watermark 介绍

  • Watermark 是一种使用延迟触发 window 执行来处理乱序数据的机制
  • 原理：当设置 Watermark = t 时 (即延迟时长为 t)，则 Flink 每一次都会获取已经到达的数据中的最大的 EventTime，然后判断 maxEventTime - t 是否等于某一个窗口的触发时间，如果相等则认为属于这个窗口的所有数据都已经到达，这个窗口被触发执行关闭，也可能存在数据丢失
  • 在数据有序的流中，相当于 Watermark = 0，即已经到达的数据中的最大的 EventTime 等于某一个窗口的触发时间，则这个窗口被触发执行关闭
  • 一般将 Watermark 设置为乱序数据流中最大的迟到时间差

3. Watermark 特点和行为

  • 水位线 (Watermark) 是作为一个特殊的数据插入到数据流中的一个标记
  • 水位线 (Watermark) 在 Flink 程序中是一个常量类，有一个时间戳属性，用来表示当前事件时间的进展
  • 水位线 (Watermark) 是基于数据的 EventTime 时间戳生成的
  • 水位线 (Watermark) 的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进

4. Watermark 在任务间的传递

任务并行度不为 1；Watermark 设置的位置越靠近 Source 端越好

在这里插入图片描述

  • 一个任务会接收上游多个并行任务的数据，也会向下游多个并行任务发送数据
  • 从上游多个并行任务接收 Watermark：使用 Partition WM 分别存储接收到的不同分区任务的 Watermark，并以其中最小的 Watermark 作为自己当前的事件时间
  • 向下游多个并行任务发送 Watermark：采取广播的分区策略，向下游的每一个任务都发送一份 Watermark，如果后续 Watermark 没有变更则不会重复发送

5. Watermark 引入

5.1 核心代码
/**方法签名：DataStream.assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks<T>)DataStream.assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks<T>)参数：1.AssignerWithPeriodicWatermarks：继承 TimestampAssigner 接口，周期性的生成 watermark，常用实现类为：BoundedOutOfOrdernessTimestampExtractor 和 AscendingTimestampExtractor2.AssignerWithPunctuatedWatermarks：继承 TimestampAssigner 接口，间断式地生成 watermark
*/
public class WatermarkTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//引入 EvenetTime       env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStream<String> dataStream = env.socketTextStream("localhost", 7777);DataStream<SensorReading> inputStream = dataStream.map(new MapFunction<SensorReading>() {@Overridepublic SensorReading map(String value) throws Exception {String[] fields = value.split(",");return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));}});//有序数据设置事件时间戳(毫秒数)和watermark//不需要传递watermark延迟时间，默认是当前事件时间戳 - 1ms 作为watermarkinputStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SensorReading>() {@Overridepublic long extractAscendingTimestamp(SensorReading element) {return element.getTimestamp() * 1000L;}});//乱序数据设置事件时间戳(毫秒数)和watermark//BoundedOutOfOrdernessTimestampExtractor 构造方法必须传入watermark延迟时间//生成的watermark时间戳 = 当前所有事件的最大时间戳 - 延迟时间inputStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {@Overridepublic long extractTimestamp(SensorReading element) {return element.getTimestamp() * 1000L;}});env.execute();}
}
5.2 AssignerWithPeriodicWatermarks

系统会周期性地生成 watermark 并插入到数据流中，默认周期是 200 毫秒

/**设置watermark生成周期：env.getConfig.setAutoWatermarkInterval(milliseconds);产生watermark的逻辑：每隔 0.2 秒钟，Flink 会调用 AssignerWithPeriodicWatermarks 的 getCurrentWatermark() 方法获取一个时间戳，如果大于之前水位的时间戳，新的 watermark 会被插入到流中。这个检查保证了水位线是单调递增的。如果方法返回的时间戳小于等于之前水位的时间戳，则不会产生新的 watermark自定义watermark周期生成器：实现 AssignerWithPeriodicWatermarks 接口，并重写 getCurrentWatermark 和 extractTimestamp 方法
*/
public class MyPeriodicAssigner implements AssignerWithPeriodicWatermarks<SensorReading> {private Long bound = 60 * 1000L;  // watermark延迟时间private Long maxTs = Long.MIN_VALUE;  // 当前最大时间戳@Nullable@Overridepublic Watermark getCurrentWatermark() {return new Watermark(maxTs - bound);}@Overridepublic long extractTimestamp(SensorReading element, long previousElementTimestamp) {maxTs = Math.max(maxTs, element.getTimestamp()); //获取当前最大的事件时间戳return element.getTimestamp();}
}
5.3 AssignerWithPunctuatedWatermarks

间断式地生成 watermark，可以根据需要对每条数据进行条件判断筛选来确定是否生成 watermark

public class MyPunctuatedAssigner implements AssignerWithPunctuatedWatermarks<SensorReading> {private Long bound = 60 * 1000L;  // 延迟时间@Nullable@Overridepublic Watermark checkAndGetNextWatermark(SensorReading lastElement, long extractedTimestamp) {if(lastElement.getId().equals("sensor_1")) {return new Watermark(extractedTimestamp - bound);} else {return null;}}@Overridepublic long extractTimestamp(SensorReading element, long previousElementTimestamp) {return element.getTimestamp();}
}

四、EventTime 的 window 操作

1. 滚动时间窗口操作

/**需求：统计 15 秒内的最小温度值，设置 2 秒的延迟
*/
public class TumblingEventTimeWindowTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);/*sensor_1,1547718199,35.8sensor_6,1547718201,15.4sensor_7,1547718202,6.7sensor_10,1547718205,38.1sensor_1,1547718207,36.3sensor_1,1547718209,32.8sensor_1,1547718212,37.1...*/DataStream<String> inputStream = env.socketTextStream("localhost", 7777);DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<SensorReading>() {@Overridepublic SensorReading map(String value) {String[] fields = value.split(",");return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));}}).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {@Overridepublic long extractTimestamp(SensorReading element) {return element.getTimestamp() * 1000L;}});//开窗聚合SingleOutputStreamOperator<SensorReading> minTempStream = dataStream.keyBy("id").timeWindow(Time.seconds(15)).minBy("temperature");minTempStream.print("minTemp");/**输出的结果分析：1.在接收到 sensor_1,1547718212,37.1 时，触发了一个窗口关闭，此时数据的 EventTime 为 1547718212，由于 watermark 延迟时间设置为 2，所以该窗口触发关闭的时间戳为 1547718212 - 2 = 1547718210，该窗口的范围为 [1547718195,1547718210)2.当前第一个窗口是 [1547718195,1547718210)，其起始点的确定规则为：2.1 滚动时间窗口使用的窗口分配器为 TumblingEventTimeWindows 类2.2 TumblingEventTimeWindows 的 assignWindows 方法中调用 getWindowStartWithOffset 方法获取起始点2.3 getWindowStartWithOffset(timestamp, offset, windowSize)：方法逻辑为 timestamp - (timestamp - offset + windowSize) % windowSize，默认 offset 为 0，所以最终得到的起始点应该是 windowSize 的整数倍，在本例中的起始点为 1547718199 - (1547718199-0+15)%15 = 15477181953.偏移量 offset：一般是用来处理不同时区的数据*/env.execute();}   
}

2. 迟到数据处理

/**需求：统计 15 秒内的最小温度值，设置 2 秒的延迟，并允许 1 分钟的迟到数据，1 分钟后的数据写入侧输出流
*/
public class TumblingEventTimeWindowDelayTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStream<String> inputStream = env.socketTextStream("localhost", 7777);DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<SensorReading>() {@Overridepublic SensorReading map(String value) {String[] fields = value.split(",");return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));}}).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {@Overridepublic long extractTimestamp(SensorReading element) {return element.getTimestamp() * 1000L;}});OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late"){};//开窗聚合SingleOutputStreamOperator<SensorReading> minTempStream = dataStream.keyBy("id").timeWindow(Time.seconds(15)).allowedLateness(Time.minutes(1));.sideOutputLateData(outputTag).minBy("temperature");minTempStream.print("minTemp");minTempStream.getSideOutput(outputTag).print("late");/**依次输入数据：sensor_1,1547718199,35.8sensor_1,1547718206,36.3sensor_1,1547718210,34.7sensor_1,1547718211,31sensor_1,1547718209,34.9sensor_1,1547718212,37.1sensor_1,1547718213,33sensor_1,1547718206,34.2sensor_1,1547718202,36...sensor_1,1547718272,34sensor_1,1547718203,30.6输出的结果分析：1.在接收到 sensor_1,1547718212,37.1 时，触发 [1547718195,1547718210) 窗口执行，此时输出数据 sensor_1,1547718209,34.9，此时 2 秒内的延迟数据能被处理  2.在接收到 sensor_1,1547718206,34.2 时，由于设置了允许 1 分钟迟到，所以 [1547718195,1547718210) 窗口仍然没有关闭，此时会更新数据为 sensor_1,1547718206,34.2，此时的系统时间戳为 1547718213 - 2 = 1547718211 - 1547718210 < 603.在接收到 sensor_1,1547718202,36 时，[1547718195,1547718210) 窗口仍然会更新输出一次数据 sensor_1,1547718206,34.24.在接收到 sensor_1,1547718272,34 时，属于 [1547718210,1547718225) 窗口的数据会输出 sensor_1,1547718211,31，此时的系统时间戳为 1547718272 - 2 = 1547718270，由于 1547718270 - 1547718210 >= 60，所以 [1547718195,1547718210) 窗口会真正的关闭5.在之后接收到 sensor_1,1547718203,30.6 时，会把数据输出到侧输出流中*/env.execute();}   
}

这篇关于详解 Flink 的时间语义和 watermark的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1043195。

相关文章

MySQL数据库双机热备的配置方法详解

MySQL数据库双机热备的配置方法详解

《MySQL数据库双机热备的配置方法详解》在企业级应用中,数据库的高可用性和数据的安全性是至关重要的,MySQL作为最流行的开源关系型数据库管理系统之一,提供了多种方式来实现高可用性,其中双机热备（M... 目录1. 环境准备1.1 安装mysql1.2 配置MySQL1.2.1 主服务器配置1.2.2 从
阅读更多...
Linux kill正在执行的后台任务 kill进程组使用详解

Linux kill正在执行的后台任务 kill进程组使用详解

《Linuxkill正在执行的后台任务kill进程组使用详解》文章介绍了两个脚本的功能和区别,以及执行这些脚本时遇到的进程管理问题,通过查看进程树、使用`kill`命令和`lsof`命令,分析了子... 目录零. 用到的命令一. 待执行的脚本二. 执行含子进程的脚本，并kill2.1 进程查看2.2 遇到的
阅读更多...
MyBatis常用XML语法详解

MyBatis常用XML语法详解

《MyBatis常用XML语法详解》文章介绍了MyBatis常用XML语法,包括结果映射、查询语句、插入语句、更新语句、删除语句、动态SQL标签以及ehcache.xml文件的使用,感兴趣的朋友跟随小... 目录1、定义结果映射2、查询语句3、插入语句4、更新语句5、删除语句6、动态 SQL 标签7、ehc
阅读更多...
详解SpringBoot+Ehcache使用示例

详解SpringBoot+Ehcache使用示例

《详解SpringBoot+Ehcache使用示例》本文介绍了SpringBoot中配置Ehcache、自定义get/set方式,并实际使用缓存的过程,文中通过示例代码介绍的非常详细,对大家的学习或者... 目录摘要概念内存与磁盘持久化存储：配置灵活性：编码示例引入依赖：配置ehcache.XML文件：配置
阅读更多...
从基础到高级详解Go语言中错误处理的实践指南

从基础到高级详解Go语言中错误处理的实践指南

《从基础到高级详解Go语言中错误处理的实践指南》Go语言采用了一种独特而明确的错误处理哲学,与其他主流编程语言形成鲜明对比,本文将为大家详细介绍Go语言中错误处理详细方法,希望对大家有所帮助... 目录1 Go 错误处理哲学与核心机制1.1 错误接口设计1.2 错误与异常的区别2 错误创建与检查2.1 基础
阅读更多...
k8s按需创建PV和使用PVC详解

k8s按需创建PV和使用PVC详解

《k8s按需创建PV和使用PVC详解》Kubernetes中,PV和PVC用于管理持久存储,StorageClass实现动态PV分配,PVC声明存储需求并绑定PV,通过kubectl验证状态,注意回收... 目录1.按需创建 PV（使用 StorageClass）创建 StorageClass2.创建 PV
阅读更多...
Python版本信息获取方法详解与实战

Python版本信息获取方法详解与实战

《Python版本信息获取方法详解与实战》在Python开发中,获取Python版本号是调试、兼容性检查和版本控制的重要基础操作,本文详细介绍了如何使用sys和platform模块获取Python的主... 目录1. python版本号获取基础2. 使用sys模块获取版本信息2.1 sys模块概述2.1.1
阅读更多...
一文详解Python如何开发游戏

一文详解Python如何开发游戏

《一文详解Python如何开发游戏》Python是一种非常流行的编程语言,也可以用来开发游戏模组,：本文主要介绍Python如何开发游戏的相关资料,文中通过代码介绍的非常详细,需要的朋友可以参考下... 目录一、python简介二、Python 开发 2D 游戏的优劣势优势缺点三、Python 开发 3D
阅读更多...
Redis 基本数据类型和使用详解

Redis 基本数据类型和使用详解

《Redis基本数据类型和使用详解》String是Redis最基本的数据类型,一个键对应一个值,它的功能十分强大,可以存储字符串、整数、浮点数等多种数据格式,本文给大家介绍Redis基本数据类型和... 目录一、Redis 入门介绍二、Redis 的五大基本数据类型2.1 String 类型2.2 Hash
阅读更多...
Java中的.close()举例详解

Java中的.close()举例详解

《Java中的.close()举例详解》.close()方法只适用于通过window.open()打开的弹出窗口,对于浏览器的主窗口,如果没有得到用户允许是不能关闭的,：本文主要介绍Java中的.... 目录当你遇到以下三种情况时，一定要记得使用 .close()：用法作用举例如何判断代码中的 input
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2