【大数据】详解 Flink 中的 WaterMark

2024-01-28 03:04
文章标签 数据 详解 flink watermark

本文主要是介绍【大数据】详解 Flink 中的 WaterMark，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

详解 Flink 中的 WaterMark

  • 1.基础概念
    • 1.1 流处理
    • 1.2 乱序
    • 1.3 窗口及其生命周期
    • 1.4 Keyed vs Non-Keyed
    • 1.5 Flink 中的时间
  • 2.Watermark
    • 2.1 案例一
    • 2.2 案例二
    • 2.3 如何设置最大乱序时间
    • 2.4 延迟数据重定向
  • 3.在 DDL 中的定义
    • 3.1 事件时间
    • 3.2 处理时间

1.基础概念

1.1 流处理

流处理，最本质的是在处理数据的时候，接受一条处理一条数据。

批处理，则是累积数据到一定程度在处理。这是他们本质的区别。

在设计上 Flink 认为数据是流式的，批处理只是流处理的特例。同时对数据分为有界数据和无界数据。

  • 有界数据对应批处理，API 对应 DateSet。
  • 无界数据对应流处理，API 对应 DataStream。

1.2 乱序

什么是乱序呢？

可以理解为数据到达的顺序和其实际产生时间的排序不一致。导致这的原因有很多，比如延迟，消息积压，重试等等。

我们知道，流处理从事件产生，到流经 source，再到 operator，中间是有一个过程和时间的。虽然大部分情况下，流到 operator 的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络、背压等原因，导致乱序的产生（out-of-order 或者说 late element）。

✅ 某数据源中的某些数据由于某种原因（如：网络原因，外部存储自身原因）会有 5 秒的延时，也就是在实际时间的第 1 秒产生的数据有可能在第 5 秒中产生的数据之后到来（比如到 Window 处理节点）。例如，有 1 ~ 10 个事件，乱序到达的序列是：2, 3, 4, 5, 1, 6, 3, 8, 9, 10, 7。

1.3 窗口及其生命周期

对于 Flink，如果来一条消息计算一条，这样是可以的，但是这样计算是非常频繁而且消耗资源，如果想做一些统计这是不可能的。所以对于 Spark 和 Flink 都产生了窗口计算。

比如，因为我们想看到过去一分钟或过去半小时的访问数据，这时候我们就需要窗口。

  • Window：Window 是处理无界流的关键，Window 将流拆分为一个个有限大小的 buckets，可以在每一个 buckets 中进行计算。
  • 当 Window 是时间窗口的时候，每个 Window 都会有一个开始时间（start_time）和结束时间（end_time）（左闭右开），这个时间是系统时间。

简而言之，只要属于此窗口的第一个元素到达，就会创建一个窗口，当时间（事件或处理时间）超过其结束时间戳加上用户指定的允许延迟时，窗口将被完全删除。

窗口有如下组件：

  • Window Assigner：用来决定某个元素被分配到哪个或哪些窗口中去。
  • Trigger：触发器。决定了一个窗口何时能够被计算或清除。触发策略可能类似于 “当窗口中的元素数量大于 4” 时，或 “当水位线通过窗口结束时”。
  • Evictor：驱逐器。Evictor 提供了在使用 WindowFunction 之前或者之后从窗口中删除元素的能力。

窗口还拥有函数，比如 ProcessWindowFunction，ReduceFunction，AggregateFunction 或 FoldFunction。该函数将包含要应用于窗口内容的计算，而触发器指定窗口被认为准备好应用该函数的条件。

1.4 Keyed vs Non-Keyed

在定义窗口之前，要指定的第一件事是流是否需要 Keyed，使用 keyBy(...) 将无界流分成逻辑的 keyed stream。如果未调用 keyBy(...)，则表示流不是 keyed stream。

  • 对于 Keyed 流，可以将传入事件的任何属性用作 key。拥有 keyed stream 将允许窗口计算由多个任务并行执行，因为每个逻辑 Keyed 流可以独立于其余任务进行处理。相同 Key 的所有元素将被发送到同一个任务。
  • 在 Non-Keyed 流的情况下，原始流将不会被分成多个逻辑流，并且所有窗口逻辑将由单个任务执行，即并行性为 1。

1.5 Flink 中的时间

Flink 在流处理程序支持不同的时间概念。分别为是 事件时间（Event Time）、处理时间（Processing Time）、提取时间（Ingestion Time）。

从时间序列角度来说，发生的先后顺序是：事件时间 ▶ 提取时间 ▶ 处理时间。

  • Event Time 是事件在现实世界中发生的时间，它通常由事件中的时间戳描述。
  • Ingestion Time 是数据进入 Apache Flink 流处理系统的时间，也就是 Flink 读取数据源时间。
  • Processing Time 是数据流入到具体某个算子 (消息被计算处理) 时候相应的系统时间。也就是 Flink 程序处理该事件时当前系统时间。

2.Watermark

Watermark 是 Apache Flink 为了处理 EventTime 窗口计算提出的一种机制，本质上也是一种时间戳。Watermark 是用于处理乱序事件或延迟数据的，这通常用 Watermark 机制结合 Window 来实现（Watermarks 用来触发 Window 窗口计算）。

2.1 案例一

public class BoundedOutOfOrdernessGenerator implements AssignerWithPeriodicWatermarks<MyEvent> {private final long maxOutOfOrderness = 3000; // 3.0 secondsprivate long currentMaxTimestamp;@Overridepublic long extractTimestamp(MyEvent element, long previousElementTimestamp) {long timestamp = element.getCreationTime();currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);return timestamp;}@Overridepublic Watermark getCurrentWatermark() {// return the watermark as current highest timestamp minus the out-of-orderness bound// 生成 watermarkreturn new Watermark(currentMaxTimestamp - maxOutOfOrderness);}
}

在这里插入图片描述
上图中是一个 10s 大小的窗口，10000～ 20000 为一个窗口。当 EventTime 为 23000 的数据到来，生成的 WaterMark 的时间戳为 20000，大于等于 window_end_time，会触发窗口计算。

2.2 案例二

public class TimeLagWatermarkGenerator implements AssignerWithPeriodicWatermarks<MyEvent> {private final long maxTimeLag = 3000; // 3 seconds@Overridepublic long extractTimestamp(MyEvent element, long previousElementTimestamp) {return element.getCreationTime();}@Overridepublic Watermark getCurrentWatermark() {// return the watermark as current time minus the maximum time lagreturn new Watermark(System.currentTimeMillis() - maxTimeLag);}
}

在这里插入图片描述
只是简单的用当前系统时间减去最大延迟时间生成 Watermark ，当 WaterMark 为 20000 时，大于等于窗口的结束时间，会触发 10000 ～ 20000 窗口计算。再当 EventTime 为 19500 的数据到来，它本应该是属于窗口 10000 ～ 20000 窗口的，但这个窗口已经触发计算了，所以此数据会被丢弃。

2.3 如何设置最大乱序时间

虽说水位线表明着早于它的事件不应该再出现，接收到水位线以前的的消息是不可避免的，这就是所谓的 迟到事件。实际上迟到事件是乱序事件的特例，和一般乱序事件不同的是它们的乱序程度超出了水位线的预计，导致窗口在它们到达之前已经关闭。

迟到事件出现时窗口已经关闭并产出了计算结果，因此处理的方法有 3 种：

  • 重新激活已经关闭的窗口并重新计算以修正结果。将迟到事件收集起来另外处理。将迟到事件视为错误消息并丢弃。Flink 默认的处理方式是直接丢弃，其他两种方式分别使用 Side Output 和 Allowed Lateness。
  • Side Output 机制 可以将迟到事件单独放入一个数据流分支，这会作为 Window 计算结果的副产品，以便用户获取并对其进行特殊处理。
  • Allowed Lateness 机制 允许用户设置一个允许的最大迟到时长。Flink 会在窗口关闭后一直保存窗口的状态直至超过允许迟到时长，这期间迟到的事件不会被丢弃，而是默认会触发窗口重新计算。因为保存窗口状态需要额外内存，并且如果窗口计算使用了 ProcessWindowFunction API 还可能使得每个迟到事件触发一次窗口的全量计算，代价比较大，所以允许迟到时长不宜设得太长，迟到事件也不宜过多，否则应该考虑降低水位线提高的速度或者调整算法。

这里总结机制为：

  • 窗口 Window 的作用是为了周期性的获取数据。
  • WaterMark 的作用是防止数据出现乱序（经常），事件时间内获取不到指定的全部数据，而做的一种保险方法。
  • allowLateNess 是将窗口关闭时间再延迟一段时间。
  • sideOutPut 是最后兜底操作，所有过期延迟数据，指定窗口已经彻底关闭了，就会把数据放到侧输出流。
public class TumblingEventWindowExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);env.setParallelism(1);
//        env.getConfig().setAutoWatermarkInterval(100);DataStream<String> socketStream = env.socketTextStream("localhost", 9999);DataStream<Tuple2<String, Long>> resultStream = socketStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(3)) {@Overridepublic long extractTimestamp(String element) {long eventTime = Long.parseLong(element.split(" ")[0]);System.out.println(eventTime);return eventTime;}}).map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> map(String value) throws Exception {return Tuple2.of(value.split(" ")[1], 1L);}}).keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(10))).allowedLateness(Time.seconds(2)) // 允许延迟处理2秒.reduce(new ReduceFunction<Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {return new Tuple2<>(value1.f0, value1.f1 + value2.f1);}});resultStream.print();env.execute();}
}

在这里插入图片描述
当 watermark 为 21000 时，触发了 [10000, 20000) 窗口计算，由于设置了 allowedLateness(Time.seconds(2))，即允许两秒延迟处理，watermark < window_end_time + lateTime 公式得到满足，因此随后 10000 和 12000 进入窗口时，依然能触发窗口计算；随后 watermark 增加到 22000，watermark < window_end_time + lateTime 不再满足，因此 11000 再次进入窗口时，窗口不再进行计算。

2.4 延迟数据重定向

流的返回值必须是 SingleOutputStreamOperator，其是 DataStream 的子类。通过 getSideOutput 方法获取延迟数据。可以将延迟数据重定向到其他流或者进行输出。

public class TumblingEventWindowExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);env.setParallelism(1);DataStream<String> socketStream = env.socketTextStream("localhost", 9999);//保存被丢弃的数据OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("late-data"){};//注意，由于getSideOutput方法是SingleOutputStreamOperator子类中的特有方法，所以这里的类型，不能使用它的父类dataStream。SingleOutputStreamOperator<Tuple2<String, Long>> resultStream = socketStream// Time.seconds(3)有序的情况修改为0.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(3)) {@Overridepublic long extractTimestamp(String element) {long eventTime = Long.parseLong(element.split(" ")[0]);System.out.println(eventTime);return eventTime;}}).map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> map(String value) throws Exception {return Tuple2.of(value.split(" ")[1], 1L);}}).keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(10))).sideOutputLateData(outputTag) // 收集延迟大于2s的数据.allowedLateness(Time.seconds(2)) //允许2s延迟.reduce(new ReduceFunction<Tuple2<String, Long>>() {@Overridepublic Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {return new Tuple2<>(value1.f0, value1.f1 + value2.f1);}});resultStream.print();//把迟到的数据暂时打印到控制台，实际中可以保存到其他存储介质中DataStream<Tuple2<String, Long>> sideOutput = resultStream.getSideOutput(outputTag);sideOutput.print();env.execute();}
}

3.在 DDL 中的定义

3.1 事件时间

事件时间属性是通过 CREATE TABLE DDL 语句中的 WATERMARK 语句定义的。水印语句在现有事件时间字段上定义 水印生成表达式，将事件时间字段标记为事件时间属性。

Flink SQL 支持在 TIMESTAMP 和 TIMESTAMP_LTZ 列上定义事件时间属性。如果源中的时间戳数据以 年-月-日-时-分-秒 表示，通常是不含时区信息的字符串值，例如 2020-04-15 20:13:40.564，建议将事件-时间属性定义为 TIMESTAMP 列。

CREATE TABLE user_actions (user_name STRING,data STRING,user_action_time TIMESTAMP(3),-- Declare the user_action_time column as an event-time attribute-- and use a 5-seconds-delayed watermark strategy.WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
) WITH (...
);SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);

如果数据源中的时间戳数据以纪元时间表示，通常是一个长值，例如 1618989564564，建议将事件时间属性定义为 TIMESTAMP_LTZ 列。

CREATE TABLE user_actions (user_name STRING,data STRING,ts BIGINT,time_ltz AS TO_TIMESTAMP_LTZ(ts, 3),-- Declare the time_ltz column as an event-time attribute-- and use a 5-seconds-delayed watermark strategy.WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND
) WITH (...
);SELECT TUMBLE_START(time_ltz, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(time_ltz, INTERVAL '10' MINUTE);

3.2 处理时间

处理时间能让表格程序根据本地机器的时间产生结果。这是最简单的时间概念，但会产生非确定性结果。处理时间不需要提取时间戳或生成水印。

在 CREATE TABLE DDL 语句中，使用系统 PROCTIME() 函数将处理时间属性定义为计算列。函数返回类型为 TIMESTAMP_LTZ。

CREATE TABLE user_actions (user_name STRING,data STRING,-- Declare an additional field as a processing-time attribute.user_action_time AS PROCTIME()
) WITH (...
);SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);

这篇关于【大数据】详解 Flink 中的 WaterMark的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

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

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

Spring Security 校验流程图 相关解释说明（认真看哦） AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证，则会调用 #attemptAuthentica
阅读更多...
大模型研发全揭秘：客服工单数据标注的完整攻略

大模型研发全揭秘：客服工单数据标注的完整攻略

在人工智能（AI）领域，数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者，掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中，工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注，不仅能够帮助提升客服自动化系统的智能化水平，还能优化客户服务流程，提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行
阅读更多...
基于MySQL Binlog的Elasticsearch数据同步实践

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展，我们的业务数据越来越多，单纯使用 MySQL 已经不能满足我们的数据查询需求，例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后，随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步，我们目前的解决方案是建立数据中间表。把需要检索的业务数据，统一放到一张M
阅读更多...
关于数据埋点，你需要了解这些基本知识

关于数据埋点，你需要了解这些基本知识

产品汪每天都在和数据打交道，你知道数据来自哪里吗？ 移动app端内的用户行为数据大多来自埋点，了解一些埋点知识，能和数据分析师、技术侃大山，参与到前期的数据采集，更重要是让最终的埋点数据能为我所用，否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式，可以区分为： 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理：自动程度高的，能解决通用统计，便于统一化管理，但个性化定
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...
异构存储（冷热数据分离）

异构存储（冷热数据分离）

异构存储主要解决不同的数据，存储在不同类型的硬盘中，达到最佳性能的问题。 异构存储Shell操作 （1）查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies （2）为指定路径（数据存储目录）设置指定的存储策略 hdfs storagepolicies -setStoragePo
阅读更多...
Hadoop集群数据均衡之磁盘间数据均衡

Hadoop集群数据均衡之磁盘间数据均衡

生产环境，由于硬盘空间不足，往往需要增加一块硬盘。刚加载的硬盘没有数据时，可以执行磁盘数据均衡命令。（Hadoop3.x新特性） plan后面带的节点的名字必须是已经存在的，并且是需要均衡的节点。 如果节点不存在，会报如下错误： 如果节点只有一个硬盘的话，不会创建均衡计划： （1）生成均衡计划 hdfs diskbalancer -plan hadoop102 （2）执行均衡计划 hd
阅读更多...
OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

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

1、简介 无感配网是指在设备联网过程中无需输入热点相关账号信息，即可快速实现设备配网，是一种兼顾高效性、可靠性和安全性的配网方式。 2、配网原理 2.1 通信原理 手机和智能设备之间的信息传递，利用特有的NAN协议实现。利用手机和智能设备之间的WiFi 感知订阅、发布能力，实现了数字管家应用和设备之间的发现。在完成设备间的认证和响应后，即可发送相关配网数据。同时还支持与常规Sof
阅读更多...
【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯，先赞后看哦~🎈🎈 🏆 作者简介：景天科技苑 🏆《头衔》：大厂架构师，华为云开发者社区专家博主，阿里云开发者社区专家博主，CSDN全栈领域优质创作者，掘金优秀博主，51CTO博客专家等。 🏆《博客》：Python全栈，前后端开发，小程序开发，人工智能，js逆向，App逆向，网络系统安全，数据分析，Django，fastapi
阅读更多...
烟火目标检测数据集 7800张 烟火检测 带标注 voc yolo

烟火目标检测数据集 7800张 烟火检测 带标注 voc yolo

一个包含7800张带标注图像的数据集，专门用于烟火目标检测，是一个非常有价值的资源，尤其对于那些致力于公共安全、事件管理和烟花表演监控等领域的人士而言。下面是对此数据集的一个详细介绍： 数据集名称：烟火目标检测数据集 数据集规模： 图片数量：7800张类别：主要包含烟火类目标，可能还包括其他相关类别，如烟火发射装置、背景等。格式：图像文件通常为JPEG或PNG格式；标注文件可能为X
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2