详解 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。

相关文章

Spring Security基于数据库验证流程详解

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明（认真看哦） AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证，则会调用 #attemptAuthentica
阅读更多...
服务器集群同步时间手记

服务器集群同步时间手记

1.时间服务器配置（必须root用户） （1）检查ntp是否安装 [root@node1 桌面]# rpm -qa|grep ntpntp-4.2.6p5-10.el6.centos.x86_64fontpackages-filesystem-1.41-1.1.el6.noarchntpdate-4.2.6p5-10.el6.centos.x86_64 （2）修改ntp配置文件 [r
阅读更多...
OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

1、简介 无感配网是指在设备联网过程中无需输入热点相关账号信息，即可快速实现设备配网，是一种兼顾高效性、可靠性和安全性的配网方式。 2、配网原理 2.1 通信原理 手机和智能设备之间的信息传递，利用特有的NAN协议实现。利用手机和智能设备之间的WiFi 感知订阅、发布能力，实现了数字管家应用和设备之间的发现。在完成设备间的认证和响应后，即可发送相关配网数据。同时还支持与常规Sof
阅读更多...
6.1.数据结构-c/c++堆详解下篇（堆排序，TopK问题）

6.1.数据结构-c/c++堆详解下篇（堆排序，TopK问题）

上篇：6.1.数据结构-c/c++模拟实现堆上篇（向下，上调整算法，建堆，增删数据）-CSDN博客 本章重点 1.使用堆来完成堆排序 2.使用堆解决TopK问题 目录 一.堆排序 1.1 思路 1.2 代码 1.3 简单测试 二.TopK问题 2.1 思路(求最小)： 2.2 C语言代码（手写堆） 2.3 C++代码（使用优先级队列 priority_queue）
阅读更多...
K8S（Kubernetes）开源的容器编排平台安装步骤详解

K8S（Kubernetes）开源的容器编排平台安装步骤详解

K8S（Kubernetes）是一个开源的容器编排平台，用于自动化部署、扩展和管理容器化应用程序。以下是K8S容器编排平台的安装步骤、使用方式及特点的概述： 安装步骤： 安装Docker：K8S需要基于Docker来运行容器化应用程序。首先要在所有节点上安装Docker引擎。 安装Kubernetes Master：在集群中选择一台主机作为Master节点，安装K8S的控制平面组件，如AP
阅读更多...
嵌入式Openharmony系统构建与启动详解

嵌入式Openharmony系统构建与启动详解

大家好，今天主要给大家分享一下，如何构建Openharmony子系统以及系统的启动过程分解。 第一：OpenHarmony系统构建      首先熟悉一下，构建系统是一种自动化处理工具的集合，通过将源代码文件进行一系列处理，最终生成和用户可以使用的目标文件。这里的目标文件包括静态链接库文件、动态链接库文件、可执行文件、脚本文件、配置文件等。      我们在编写hellowor
阅读更多...
LabVIEW FIFO详解

LabVIEW FIFO详解

在LabVIEW的FPGA开发中，FIFO（先入先出队列）是常用的数据传输机制。通过配置FIFO的属性，工程师可以在FPGA和主机之间，或不同FPGA VIs之间进行高效的数据传输。根据具体需求，FIFO有多种类型与实现方式，包括目标范围内FIFO（Target-Scoped）、DMA FIFO以及点对点流（Peer-to-Peer）。 FIFO类型 **目标范围FIFO（Target-Sc
阅读更多...
019、JOptionPane类的常用静态方法详解

019、JOptionPane类的常用静态方法详解

目录 JOptionPane类的常用静态方法详解 1. showInputDialog()方法 1.1基本用法 1.2带有默认值的输入框 1.3带有选项的输入对话框 1.4自定义图标的输入对话框 2. showConfirmDialog()方法 2.1基本用法 2.2自定义按钮和图标 2.3带有自定义组件的确认对话框 3. showMessageDialog()方法 3.1
阅读更多...
MiniGPT-3D, 首个高效的3D点云大语言模型，仅需一张RTX3090显卡，训练一天时间，已开源

MiniGPT-3D, 首个高效的3D点云大语言模型，仅需一张RTX3090显卡，训练一天时间，已开源

项目主页：https://tangyuan96.github.io/minigpt_3d_project_page/ 代码：https://github.com/TangYuan96/MiniGPT-3D 论文：https://arxiv.org/pdf/2405.01413 MiniGPT-3D在多个任务上取得了SoTA，被ACM MM2024接收，只拥有47.8M的可训练参数，在一张RTX
阅读更多...
脏页的标记方式详解

脏页的标记方式详解

脏页的标记方式 一、引言 在数据库系统中，脏页是指那些被修改过但还未写入磁盘的数据页。为了有效地管理这些脏页并确保数据的一致性，数据库需要对脏页进行标记。了解脏页的标记方式对于理解数据库的内部工作机制和优化性能至关重要。 二、脏页产生的过程 当数据库中的数据被修改时，这些修改首先会在内存中的缓冲池（Buffer Pool）中进行。例如，执行一条 UPDATE 语句修改了某一行数据，对应的缓
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2