【API篇】十、生成Flink水位线

文章目录

  • 1、水位线的生成原则
  • 2、有序流内置水位线
  • 3、乱序流内置水位线
  • 4、自定义周期性水位线生成器
  • 5、自定义断点式水位线生成器
  • 6、从数据源中发送水位线

1、水位线的生成原则

水位线出现，即代表这个时间之前的数据已经全部到齐，之后不会再出现之前的数据了。参考前面的乱序流，可以得出：

  • 想要保证数据绝对正确，就得加足够大的延迟，但实时性就没保障了
  • 想要实时性强，就得把延迟设置小，但此时迟到数据可能遗漏，准确性降低

水位线的定义，是对低延迟和结果准确性的一个权衡。Flink生成水位线的方法是.assignTimestampsAndWatermarks()，它主要用来为流中的数据分配时间戳，并生成水位线来指示事件时间

DataStream<Event> stream = env.addSource(xxx);DataStream withTimestampsAndWatermarks = stream.assignTimestampsAndWatermarks(WatermarkStrategy对象);

WatermarkStrategy是一个接口，包含了一个时间戳分配器TimestampAssigner和一个水位线生成WatermarkGenerator：

public interface WatermarkStrategy<T> extends TimestampAssignerSupplier<T>,WatermarkGeneratorSupplier<T>{// 负责从流中数据元素的某个字段中提取时间戳，并分配给元素。时间戳的分配是生成水位线的基础。@OverrideTimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context);// 主要负责按照既定的方式，基于时间戳生成水位线@OverrideWatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}

2、有序流内置水位线

有序流的时间戳全部单调递增，没有迟到数据，直接WatermarkStrategy.forMonotonousTimestamps()就可以拿到WatermarkStrategy对象

public class WatermarkMonoDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("node01", 9527).map(new WaterSensorMapFunction());// TODO 1.定义Watermark策略WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy// 1.1 指定watermark生成：升序的watermark，没有等待时间.<WaterSensor>forMonotonousTimestamps()// 1.2 指定 时间戳分配器，从数据中提取.withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {@Overridepublic long extractTimestamp(WaterSensor element, long recordTimestamp) {System.out.println("数据=" + element + ",recordTs=" + recordTimestamp);// 返回的时间戳，要毫秒，这里拿自定义对象的ts属性做为时间戳return element.getTs() * 1000L;}});// TODO 2. 指定 watermark策略SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);sensorDSwithWatermark.keyBy(sensor -> sensor.getId())// TODO 3.使用事件时间语义的窗口，别再用处理时间TumblingProcessTime.window(TumblingEventTimeWindows.of(Time.seconds(10))).process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {@Overridepublic void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {long startTs = context.window().getStart();long endTs = context.window().getEnd();String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");long count = elements.spliterator().estimateSize();out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());}}).print();env.execute();}
}

执行下，输入10时，逻辑时钟被推到了10s，到达区间，触发窗口，执行全窗口函数的process，输出当前窗口的数据：

在这里插入图片描述

3、乱序流内置水位线

调用WatermarkStrategy. forBoundedOutOfOrderness()，传入延迟时间：

public class WatermarkOutOfOrdernessDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("node01", 9527).map(new WaterSensorMapFunction());// TODO 1.定义Watermark策略WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy// 1.1 指定watermark生成：乱序的，等待3s.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))// 1.2 指定 时间戳分配器，从数据中提取.withTimestampAssigner((element, recordTimestamp) -> {// 返回的时间戳，要 毫秒System.out.println("数据=" + element + ",recordTs=" + recordTimestamp);return element.getTs() * 1000L;});// TODO 2. 指定 watermark策略SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);sensorDSwithWatermark.keyBy(sensor -> sensor.getId())// TODO 3.使用 事件时间语义 的窗口.window(TumblingEventTimeWindows.of(Time.seconds(10))).process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {@Overridepublic void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {long startTs = context.window().getStart();long endTs = context.window().getEnd();String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");long count = elements.spliterator().estimateSize();out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());}}).print();env.execute();}
}

执行：

在这里插入图片描述

简单分析下结果：

  • 第一条数据s1，1，1进来，创建窗口，水位线为1s-3s（延迟3s)
  • s1，10，10进来，水位线为10-3 =7s，还未到达10，窗口不触发（若是有序流，无等待下，此时窗口已被触发了）
  • 此时进来一条乱序数据，比如s1，6，6，6-3=3s，水位线保持上面的7不变，watermark不会推进，且6这条数据也会被统计在[0,10)的区间内
  • s1，11，11进来，11-3=8，也不会触发，但这条数据是属于[10,20)区间的那个桶的
  • s1，13，13进来，达到10，窗口触发

4、自定义周期性水位线生成器

上面只是定义了时间戳的提取逻辑，水位线的生成采用的默认内置策略。接下来自定义水位线生成器：周期性水位生成器。

周期性生成器是通过onEvent()观察判断输入的事件，而在onPeriodicEmit()里发射生成的水位线

// 自定义水位线的产生
public class CustomPeriodicWatermarkExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("node01", 9527).map(new WaterSensorMapFunction());// 定义Watermark策略WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy// 1.1 指定watermark生成器.<WaterSensor>forGenerator(context -> MyPeriodWatermarkGenerator<>(3000L))// 1.2 指定时间戳分配器，从数据中提取.withTimestampAssigner((element, recordTimestamp) -> {// 返回的时间戳，要 毫秒System.out.println("数据=" + element + ",recordTs=" + recordTimestamp);return element.getTs() * 1000L;});// TODO 2. 指定 watermark策略SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);sensorDSwithWatermark.keyBy(sensor -> sensor.getId())// TODO 3.使用 事件时间语义 的窗口.window(TumblingEventTimeWindows.of(Time.seconds(10))).process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {@Overridepublic void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {long startTs = context.window().getStart();long endTs = context.window().getEnd();String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");long count = elements.spliterator().estimateSize();out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());}}).print();env.execute();}}

模仿前面的内置生成器，定义自己的水位线生成器：

public class MyPeroidWatermarkGenerator implements WatermarkGenerator<Event> {private Long delayTime = 5000L; // 延迟时间private Long maxTs = -Long.MAX_VALUE + delayTime + 1L; // 观察到的最大时间戳//构造方法，传入延迟时间，构造水位线生成器对象public MyPeroidWatermarkGenerator(long delayTime){this.delayTime = delayTime;this.maxTs = Long.MIN_VALUE + this.delayTime + 1;}/*** 每条数据进来都调用一次，用来提取最大的事件事件*/@Overridepublic void onEvent(Event event，long eventTimestamp，WatermarkOutput output) {// 每来一条数据就调用一次maxTs = Math.max(event.timestamp，maxTs); // 更新最大时间戳System.out.println("调用了onEvent方法，获取目前为止最大的时间戳=" + maxTimestamp);}/*** 周期性调用，默认20ms*/@Overridepublic void onPeriodicEmit(WatermarkOutput output) {// 发射水位线，默认200ms调用一次output.emitWatermark(new Watermark(maxTs - delayTime - 1L));System,out,println("调用了onPeriodicEmit方法,生成watermark==" + (maxTimestamp - delayTs - 1) );}}

核心部分，指定水位线生成器的Lamdba表达式展开就是：

在这里插入图片描述

运行：

  • 数据没进来前，每200ms调用一次发射水位线的方法，此时的水位线是构造方法里Long.MIN_VALUE那个
  • 进来一条数据，调用onEvent，最大时间戳被更新，到周期后再发射水位线maxTs-delayTs-1
  • 继续周期性调用onPeriodicEmit方法

在这里插入图片描述

onPeriodicEmit()里调用output.emitWatermark()，就可以发出水位线了，这个方法由系统框架周期性地调用，默认200ms一次

修改默认的周期，比如改为400ms：

env.getConfig().setAutoWatermarkInterval(400L);

5、自定义断点式水位线生成器

断点式生成器会不停地检测onEvent()中的事件，发现带有水位线信息的当事件时，就立即发出水位线。改下代码，定义水位线生成器：

public class PointWatermarkGenerator implements WatermarkGenerator<Event> {private Long delayTime = 5000L; // 延迟时间private Long maxTs = -Long.MAX_VALUE + delayTime + 1L; // 观察到的最大时间戳//构造方法，传入延迟时间，构造水位线生成器对象public MyPeroidWatermarkGenerator(long delayTime){this.delayTime = delayTime;this.maxTs = Long.MIN_VALUE + this.delayTime + 1;}/*** 每条数据进来都调用一次，用来提取最大的事件事件*/@Overridepublic void onEvent(Event event，long eventTimestamp，WatermarkOutput output) {// 每来一条数据就调用一次maxTs = Math.max(event.timestamp，maxTs); // 更新最大时间戳// 发射水位线output.emitWatermark(new Watermark(maxTs - delayTime - 1L));System.out.println("调用了onEvent方法，获取目前为止最大的时间戳=" + maxTimestamp + ",生成watermark==" + (maxTimestamp - delayTs - 1));}/*** 周期性调用，默认20ms*/@Overridepublic void onPeriodicEmit(WatermarkOutput output) {}}

周期性代码改为：

//...// 定义Watermark策略WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy// 1.1 指定watermark生成器.<WaterSensor>forGenerator(context -> PointWatermarkGenerator<>(3000L))// 1.2 指定时间戳分配器，从数据中提取.withTimestampAssigner((element, recordTimestamp) -> {// 返回的时间戳，要 毫秒return element.getTs() * 1000L;});

运行：此时不再周期性的发射水位线

在这里插入图片描述

6、从数据源中发送水位线

在自定义的数据源中抽取事件时间，然后发送水位线：

env.fromSource(
kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource" 
)//注意fromSorce方法的第二个传参，之前用的WatermarkStrategy.noWatermark()

注意此时不用再assignTimestampsAndWatermarks了，在自定义数据源中生成水位线和在程序中使用assignTimestampsAndWatermarks方法生成水位线二者只能取其一

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/172675.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Kafka KRaft模式探索

Kafka KRaft模式探索

1.概述 Kafka是一种高吞吐量的分布式发布订阅消息系统&#xff0c;它可以处理消费者在网站中的所有动作流数据。其核心组件包含Producer、Broker、Consumer&#xff0c;以及依赖的Zookeeper集群。其中Zookeeper集群是Kafka用来负责集群元数据的管理、控制器的选举等。 2.内容…
阅读更多...
CAD2024最新中文版安装教程分享

CAD2024最新中文版安装教程分享

wx供重浩&#xff1a;创享日记 对话框发送&#xff1a;cad24 获取 AutoCAD是目前计算机辅助设计领域最流行的CAD软件&#xff0c;此软件功能强大、使用方便&#xff0c;在国内外广泛应用于机械、建筑、家居、纺织等诸多行业。CAD制图软件具有良好的用户界面&#xff0c;通过交互…
阅读更多...
【HarmonyOS】鸿蒙操作系统架构

【HarmonyOS】鸿蒙操作系统架构

HarmonyOS架构 一. 鸿蒙系统定位二. 架构整体遵从分层设计三. HarmonyOS具有的技术特性四. HarmonyOS有三大特征 其它相关推荐&#xff1a; 软考系统架构之案例篇(架构设计相关概念) 系统架构之微服务架构 系统架构设计之微内核架构 所属专栏&#xff1a;系统架构设计师 一. 鸿…
阅读更多...
人工智能之深度学习

人工智能之深度学习

1. 引言 时至今日&#xff0c;人们常用的计算机程序几乎都是软件开发人员从零编写的。 比如&#xff0c;现在开发人员要编写一个程序来管理网上商城。 经过思考&#xff0c;开发人员可能提出如下一个解决方案&#xff1a; 首先&#xff0c;用户通过Web浏览器&#xff08;或移动…
阅读更多...
如何将本地 PDF 文件进行翻译

如何将本地 PDF 文件进行翻译

在日常工作和学习中&#xff0c;我们经常会遇到需要翻译 PDF 文件的情况。比如&#xff0c;我们需要将一份英文的技术文档翻译成中文&#xff0c;或者将一份中文的法律文件翻译成英文。 传统上&#xff0c;我们可以使用专业翻译软件或服务来翻译 PDF 文件。但是&#xff0c;这…
阅读更多...
使用Gateway解决跨域问题时配置文件不生效的情况之一

使用Gateway解决跨域问题时配置文件不生效的情况之一

首先html文件只有一个发送ajax请求 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta http-equiv"X-UA-Compatible" content"IEedge"><meta name"viewport" content&q…
阅读更多...
【鸿蒙软件开发】ArkTS基础组件之Gauge(环形图表)、LoadingProgress(动态加载)

【鸿蒙软件开发】ArkTS基础组件之Gauge(环形图表)、LoadingProgress(动态加载)

文章目录 前言一、Gauge环形图表1.1 子组件1.2 接口参数介绍 1.2 属性1.3 示例代码二、LoadingProgress2.1 子组件2.2 接口2.3 属性2.4 示例代码 总结 前言 Gauge&#xff1a;数据量规图表组件&#xff0c;用于将数据展示为环形图表。 LoadingProgress&#xff1a;用于显示加载…
阅读更多...
公网远程访问macOS本地web服务器

公网远程访问macOS本地web服务器

#　公网访问macOS本地web服务器【内网穿透】 文章目录 1. 启动Apache服务器2. 公网访问本地web服务2.1 本地安装配置cpolar2.2 创建隧道2.3 测试访问公网地址3. 配置固定二级子域名3.1 保留一个二级子域名3.2 配置二级子域名4. 测试访问公网固定二级子域名 以macOS自带的Apache…
阅读更多...
Azure云工作站上做Machine Learning模型开发 - 全流程演示

Azure云工作站上做Machine Learning模型开发 - 全流程演示

目录 本文内容先决条件从“笔记本”开始设置用于原型制作的新环境&#xff08;可选&#xff09;创建笔记本开发训练脚本迭代检查结果 关注TechLead&#xff0c;分享AI全维度知识。作者拥有10年互联网服务架构、AI产品研发经验、团队管理经验&#xff0c;同济本复旦硕&#xff0…
阅读更多...
Android应用：实现网络加载商品数据【OKHttp、Glide、Gson】

Android应用：实现网络加载商品数据【OKHttp、Glide、Gson】

实现网络加载商品数据的功能&#xff1a; 1、在AndroidManifest.xml中声明网络权限&#xff1b; 2、在app/build.gradle中添加okhttp, glide, gson等必需的第3方库&#xff1b; 3、在MainActivity中通过OkHttpClient连接给定的Web服务&#xff0c;获取商品数据&#xff1b;对…
阅读更多...
Ubuntu安装VM TOOLS解决虚拟机无法和WINDOWS粘贴复制问题

Ubuntu安装VM TOOLS解决虚拟机无法和WINDOWS粘贴复制问题

1&#xff1a;首先使用VMware Workstation安装一个Ubuntu的系统。 2&#xff1a;现在已经不建议安装VM TOOLS。建议安装OPEN-VM-TOOLS。 3&#xff1a;进入系统使用下面的命令安装。 sudo apt install open-vm-tools 4&#xff1a;提示下面错误&#xff0c;Package open-vm…
阅读更多...
CVE-2022-32991靶场复现

CVE-2022-32991靶场复现

靶场环境&#xff1a; 题目提示了该CMS的welcome.php中存在SQL注入攻击。 CVE官方给出的提示&#xff1a; welcome.php页面存在SQL注入&#xff0c;并且这个参数是eid 打开靶场环境&#xff1a; 页面是一个登陆注册的界面 用户注册&#xff1a; 1 010.com 123456 123456 点击Re…
阅读更多...
第五天：前端页面展示不出来

第五天：前端页面展示不出来

0.问题展示 但是我这个 展示不来 1.前端页面 渲染 不出来问题 首先想到的是&#xff0c;前端传值的问题然后排查到前端 传值是正确的传值的话&#xff0c;开始debugger 页面没有&#xff0c;然后找他的父页面&#xff0c;父页面是有值的然后找父 方法内部debugger发现第三步…
阅读更多...
JavaWeb——IDEA操作：Project最终新建module

JavaWeb——IDEA操作：Project最终新建module

在project中创建新的module&#xff1a; 创建一个新的module很容易&#xff0c;但是它可能连接不上Tomcat&#xff0c;因此需要修改一些配置&#xff1a; 将以下地址修改为新module的地址
阅读更多...
设计模式（19）命令模式

设计模式（19）命令模式

一、介绍&#xff1a; 1、定义&#xff1a;命令模式&#xff08;Command Pattern&#xff09;是一种行为设计模式&#xff0c;它将请求封装为一个对象&#xff0c;从而使你可以使用不同的请求对客户端进行参数化。命令模式还支持请求的排队、记录日志、撤销操作等功能。 2、组…
阅读更多...
智能水厂运行与调控3D模拟仿真在线展示提高整个系统的协同效应

智能水厂运行与调控3D模拟仿真在线展示提高整个系统的协同效应

水厂在生活中的重要性不可忽视。它们提供清洁、安全的水源&#xff0c;满足人们饮用、洗浴、烹饪等基本需求&#xff0c;保障公共卫生&#xff0c;预防疾病传播;同时&#xff0c;水厂也促进经济发展&#xff0c;为工业生产和农业灌溉提供保障&#xff0c;吸引和支持企业的投资和…
阅读更多...
经常遇到的问题

经常遇到的问题

一个前端经常会遇到的问题 例如&#xff0c;我想要在一个项目里&#xff0c;监听所有的fetch请求&#xff0c;应该怎么办&#xff1f;又或者说&#xff0c;我想用别人封装好的方法&#xff0c;但是在它之前&#xff0c;需要经过一层处理、判断&#xff0c;然后再看情况是否调用…
阅读更多...
正点原子嵌入式linux驱动开发——Linux LCD驱动

正点原子嵌入式linux驱动开发——Linux LCD驱动

LCD是很常用的一个外设&#xff0c;通过LCD可以显示绚丽的图片、界面等&#xff0c;提交人机交互的效率。STM32MP1提供了一个LTDC接口用于连接RGB接口的液晶屏。本章就来学校一下如何在Linux下驱动LCD屏。 LCD和LTDC简介 LCD简介 这里在当时学习stm32裸机开发的时候就学过了…
阅读更多...
简化通知基础设施：开源的消息通知服务 | 开源专题 No.41

简化通知基础设施：开源的消息通知服务 | 开源专题 No.41

novuhq/novu Stars: 22.9k License: MIT Novu 是一个开源的通知基础设施项目&#xff0c;它提供了统一的 API 来通过多个渠道发送通知&#xff0c;包括应用内、推送、电子邮件、短信和聊天。主要功能有&#xff1a; 为所有消息提供商 (应用内、电子邮件、短信、推送和聊天) 提…
阅读更多...
【iPad已停用】解锁教程

【iPad已停用】解锁教程

iPad多次输错密码时&#xff0c;会自动锁定并停用&#xff0c;这时候你可以使用iTuens或Tenorshare进行解锁。 一、使用iTunes解锁 下载并安装iTunes 使用数据线将iPad连接上电脑 让iPad进入恢复模式&#xff0c;同时安装iPad电源键和Home键&#xff0c;直到Logo出现也不要松…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发