当前位置: 首页 > news >正文

Flink实时开发添加水印的案例分析

在Flink中,处理时间序列数据时,通常需要考虑事件时间和水印(watermarks)的处理。以下是修改前后的代码对比分析:

修改前的代码:

val systemDS = unitDS.map(dp => {dp.setDeviceCode(DeviceCodeEnum.fromPidToSystem(dp.getDeviceCode))dp
}).keyBy(_.getDeviceCode)
.window(TumblingEventTimeWindows.of(Time.seconds(60)))
.process(new MySystemWinF)
  1. unitDS 经过一个 map 操作,将每个元素的 deviceCode 转换为系统设备码。
  2. 使用 keyBy(_.getDeviceCode) 对转换后的设备码进行分组。
  3. 定义了一个基于事件时间的滚动窗口,窗口大小为60秒。
  4. 使用 process 操作应用自定义的窗口函数 HPageSystemWinF 来处理每个窗口中的数据。

注意:修改前的代码没有显示地处理水印(watermarks),这可能导致在处理乱序数据或延迟数据时出现问题。

修改后的代码:

val systemDS = unitDS.map(dp => {dp.setDeviceCode(DeviceCodeEnum.fromPidToSystem(dp.getDeviceCode))dp
}).keyBy(_.getDeviceCode)
.assignTimestampsAndWatermarks(WatermarkStrategy.<boundedOutOfOrdernessDaysPower>forBoundedOutOfOrderness(Duration.ofSeconds(5)) // 假设这里应该是.forBoundedOutOfOrderness而不是.forBoundedOutOfOrdernessDaysPower.withIdleness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner[DaysPower] {override def extractTimestamp(element: DaysPower, recordTimestamp: Long): Long = {Math.max(element.getEventTime, recordTimestamp)}})
).keyBy(_.getDeviceCode)
.window(TumblingEventTimeWindows.of(Time.seconds(60)))
.process(new MySystemWinF)
  1. 与修改前相同的部分:mapkeyBy, 和 window 操作。
  2. 添加了 assignTimestampsAndWatermarks 方法来处理事件时间和水印:
    • 使用 WatermarkStrategy.forBoundedOutOfOrderness 允许一定程度的乱序数据(这里是5秒)。
    • .withIdleness(Duration.ofSeconds(5)) 设置了空闲超时时间为5秒,用于处理不活跃的键。
    • 使用 withTimestampAssigner 自定义了时间戳分配器,确保使用的事件时间是元素中的 eventTime 和记录的 recordTimestamp 中的较大值。

不同点和适用场景:

  • 事件时间和水印处理:修改后的代码显式地处理了事件时间和水印,这对于处理乱序数据、延迟数据以及确保正确的时间窗口计算是非常重要的。如果您的数据流中存在乱序或延迟数据,或者您希望更严格地保证处理时间窗口的正确性,那么应该使用修改后的代码。
  • 空闲超时:通过设置空闲超时,可以处理那些长时间不活跃的键,避免因为某些键长时间没有新数据而导致整个程序挂起。
  • 延迟数据处理:如果数据有可能晚到,但仍然需要被纳入正确的窗口进行计算,水印可以帮助界定数据的“迟到”界限。
    精确的时间窗口分析:对于需要基于事件实际发生时间而非数据处理时间进行分析的场景,如实时监控、金融交易分析等,事件时间模型是必须的。

相关文章:

Flink实时开发添加水印的案例分析

在Flink中&#xff0c;处理时间序列数据时&#xff0c;通常需要考虑事件时间和水印&#xff08;watermarks&#xff09;的处理。以下是修改前后的代码对比分析&#xff1a; 修改前的代码&#xff1a; val systemDS unitDS.map(dp > {dp.setDeviceCode(DeviceCodeEnum.fro…...

收银系统源码-线上商城diy装修

线下线上一体化收银系统越来越受门店重视&#xff0c;尤其是连锁多门店&#xff0c;想通过线下线上相互带动&#xff0c;相互引流&#xff0c;提升门店营业额。商城商城如何装修呢&#xff1f; 1.收银系统开发语言 核心开发语言: PHP、HTML5、Dart后台接口: PHP7.3后合管理网…...

Linux中nohup(no hang up)不挂起,用于在系统后台不挂断地运行命令,即使退出终端也不会影响程序的运行。

nohup的英文全称是 no hang up&#xff0c;即“不挂起”。这个命令在Linux或Unix系统中非常有用&#xff0c;主要用于在系统后台不挂断地运行命令&#xff0c;即使退出终端也不会影响程序的运行。默认情况下&#xff08;非重定向时&#xff09;&#xff0c;nohup会将输出写入一…...

【.NET全栈】ASP.NET开发Web应用——站点导航技术

文章目录 前言一、站点地图1、定义站点地图文件2、使用SiteMapPath控件3、SiteMap类4、URL地址映射 二、TreeView控件1、使用TreeView控件2、以编程的方式添加节点3、使用TreeView控件导航4、绑定到XML文件5、按需加载节点6、带复选框的TreeView控件 三、Menu控件1、使用Menu控…...

docker 容器内部UI映射host

方法有很多&#xff0c; 目前我总计一个我自己尝试成功的方法&#xff0c;通过xpra。 Xpra可以看作是screen或tmux的图形版本&#xff0c;支持远程X11应用程序的显示和交互。 在远程服务器上&#xff0c;安装Xpra&#xff1a; sudo apt-get install xpra启动Xpra服务器会话&…...

数仓面试题——DWS层新增维度字段需求

前言 在数据仓库开发中&#xff0c;数据仓库的设计和维护一直是一个备受关注的话题。随着业务需求的不断变化&#xff0c;数据仓库的结构也需要随之调整。 面试过程中&#xff0c;多次被提问&#xff1a;当DWS构建好后&#xff0c;突然来了一个新的需求&#xff0c;需要添加某个…...

Qt实现MDI应用程序

本文记录Qt实现MDI应用程序的相关操作实现 目录 1.MDM模式下窗口的显示两种模式 1.1TabbedView 页签化显示 1.2 SubWindowView 子窗体显示 堆叠cascadeSubWindows 平铺tileSubWindows 2.MDM模式实现记录 2.1. 窗体继承自QMainWindow 2.2.增加组件MdiArea 2.3.定义统一…...

逆向案例二十六——webpack自执行函数是完整的,但我们只需要加载器,某职业技术学校登陆密码逆向

网址&#xff1a;统一身份认证平台 找到登陆包&#xff0c;搜索找到加密位置。 找到加密位置&#xff0c;打上断点 分析&#xff0c;E就是加密结果 进入n.i函数&#xff0c;就是t.i,看一下这个函数&#xff0c;传一个值&#xff0c;然后不变的返回&#xff0c;所以没什么意义 …...

容器安全最佳实践和工具

容器安全最佳实践和工具 什么是容器安全 容器安全是指保护容器化应用程序和基础设施免受潜在威胁和攻击的措施和策略。容器化技术&#xff08;如Docker、Kubernetes&#xff09;使得应用程序能够在隔离的环境中运行&#xff0c;这既提供了灵活性&#xff0c;也引入了新的安全…...

牛客周赛 Round 51

目录 A.小红的同余 B.小红的三倍数 C.小红充电 D.小红的gcd E.小红走矩阵 F.小红的数组 这次周赛题目比较简单&#xff0c;算法题也基本上是板子题&#xff0c;出得很好(&#xff5e;&#xffe3;▽&#xffe3;)&#xff5e; A.小红的同余 思路&#xff1a;签到题&am…...

【Linux】详解加锁实现线程互斥

一、多线程不加线程互斥可能会引发的问题 下面是一个抢标逻辑。抢票为什么会抢到负数&#xff1a;假设当票数为1时&#xff0c;此时四个进程的判断条件tickets都大于0&#xff0c;都会进入抢票操作&#xff0c;第一个进程抢完票以后tickets0并写回内存&#xff0c;第二个进程再…...

Java学习高级四

JDK8开始&#xff0c;接口新增了三种形式的方法 接口的多继承 内部类 成员内部类 静态内部类 局部内部类 匿名内部类 import javax.swing.*; import java.awt.event.ActionEvent;public class Test {public static void main(String[] args) {// 扩展 内部类在开发中的真实使用…...

mmc-utils 的 MMC 测试工具

MMC 工具介绍 有一个名为 mmc-utils 的 MMC 测试工具&#xff0c;由 Ulf Hansson 维护&#xff0c;您可以在以下公共 git 存储库中找到它&#xff1a; mmc/mmc-utils.git - Unnamed repository; edit this file description to name the repository. 功能 mmc-utils 工具可以…...

使用Python Turtle绘制圣诞树和装饰

简介(❤ ω ❤) 在这篇文章中&#xff0c;我们将探索如何使用Python的Turtle模块来绘制一个充满节日气氛的圣诞树&#xff0c;以及一些可爱的装饰品。Turtle是一个受Logo语言启发的图形库&#xff0c;非常适合初学者学习编程和创建图形。 码农不是吗喽&#xff08;大学生版&…...

非常好的新版网盘系统,是一款PHP网盘与外链分享程序,支持文件预览

这是一款PHP网盘与外链分享程序&#xff0c;支持所有格式文件的上传&#xff0c; 可以生成文件外链、图片外链、音乐视频外链&#xff0c;生成外链同时自动生成相应的UBB代码和HTML代码&#xff0c; 还可支持文本、图片、音乐、视频在线预览&#xff0c;这不仅仅是一个网盘&a…...

针对【module_or_function】的单元测试,全面覆盖可能的【edge_cases】

针对【module_or_function】的单元测试&#xff0c;全面覆盖可能的【edge_cases】 编写单元测试是为了验证代码模块或函数的正确性和鲁棒性。对于module_or_function&#xff0c;首先需要确定这个模块或函数的具体功能和预期输入范围。一个好的单元测试应该包括以下几个步骤&a…...

OTA测试!

OTA测试&#xff0c;全称“Over-The-Air Testing”&#xff0c;是一种无线通信设备的性能测试方法&#xff0c;主要用于评估设备在无线传输环境中的性能表现。以下是关于OTA测试的详细介绍&#xff1a; 一、定义与目的 OTA测试着重进行整机辐射性能方面的测试&#xff0c;以评…...

[H最短路] lc2959. 关闭分部的可行集合数目(Floyd最短路+二进制枚举+模板题)

文章目录 1. 题目来源2. 题目解析 1. 题目来源 链接&#xff1a;2959. 关闭分部的可行集合数目 2. 题目解析 看了看题好像还没啥思路&#xff0c;结果一看数据范围&#xff0c;好家伙…n 最大就 10 啊&#xff0c;那不直接闭眼直接 Floyd枚举所有情况即可吗&#xff1f;&…...

pyinstaller用法详解3

本文使用创作助手。 大家好&#xff0c;时隔多日&#xff0c;我又更新了pyinstaller的用法详解&#xff01; 当然&#xff0c;这一次要比之前更详细&#xff0c;十分详细。 谢谢大家的支持&#xff0c;我们现在开始&#xff01; 一、快速开始使用pyinstaller 我之前的文章…...

养猫新手不会挑智能猫砂盆?2024最新挑选干货分享!

不得不说智能猫砂盆真的帮了我很大的忙&#xff0c;四年以来我陆陆续续养了很多的猫咪&#xff0c;但是因为需要上班&#xff0c;所以有时候也对铲屎的工作有些力不从心&#xff0c;后面听了朋友的建议&#xff0c;去入手了智能猫砂盆&#xff0c;不得不说买智能猫砂盆也非常的…...

【大模型RAG】拍照搜题技术架构速览:三层管道、两级检索、兜底大模型

摘要 拍照搜题系统采用“三层管道&#xff08;多模态 OCR → 语义检索 → 答案渲染&#xff09;、两级检索&#xff08;倒排 BM25 向量 HNSW&#xff09;并以大语言模型兜底”的整体框架&#xff1a; 多模态 OCR 层 将题目图片经过超分、去噪、倾斜校正后&#xff0c;分别用…...

观成科技:隐蔽隧道工具Ligolo-ng加密流量分析

1.工具介绍 Ligolo-ng是一款由go编写的高效隧道工具&#xff0c;该工具基于TUN接口实现其功能&#xff0c;利用反向TCP/TLS连接建立一条隐蔽的通信信道&#xff0c;支持使用Let’s Encrypt自动生成证书。Ligolo-ng的通信隐蔽性体现在其支持多种连接方式&#xff0c;适应复杂网…...

Swift 协议扩展精进之路:解决 CoreData 托管实体子类的类型不匹配问题(下)

概述 在 Swift 开发语言中&#xff0c;各位秃头小码农们可以充分利用语法本身所带来的便利去劈荆斩棘。我们还可以恣意利用泛型、协议关联类型和协议扩展来进一步简化和优化我们复杂的代码需求。 不过&#xff0c;在涉及到多个子类派生于基类进行多态模拟的场景下&#xff0c;…...

CMake基础:构建流程详解

目录 1.CMake构建过程的基本流程 2.CMake构建的具体步骤 2.1.创建构建目录 2.2.使用 CMake 生成构建文件 2.3.编译和构建 2.4.清理构建文件 2.5.重新配置和构建 3.跨平台构建示例 4.工具链与交叉编译 5.CMake构建后的项目结构解析 5.1.CMake构建后的目录结构 5.2.构…...

使用Spring AI和MCP协议构建图片搜索服务

目录 使用Spring AI和MCP协议构建图片搜索服务 引言 技术栈概览 项目架构设计 架构图 服务端开发 1. 创建Spring Boot项目 2. 实现图片搜索工具 3. 配置传输模式 Stdio模式&#xff08;本地调用&#xff09; SSE模式&#xff08;远程调用&#xff09; 4. 注册工具提…...

【Elasticsearch】Elasticsearch 在大数据生态圈的地位 实践经验

Elasticsearch 在大数据生态圈的地位 & 实践经验 1.Elasticsearch 的优势1.1 Elasticsearch 解决的核心问题1.1.1 传统方案的短板1.1.2 Elasticsearch 的解决方案 1.2 与大数据组件的对比优势1.3 关键优势技术支撑1.4 Elasticsearch 的竞品1.4.1 全文搜索领域1.4.2 日志分析…...

MySQL的pymysql操作

本章是MySQL的最后一章&#xff0c;MySQL到此完结&#xff0c;下一站Hadoop&#xff01;&#xff01;&#xff01; 这章很简单&#xff0c;完整代码在最后&#xff0c;详细讲解之前python课程里面也有&#xff0c;感兴趣的可以往前找一下 一、查询操作 我们需要打开pycharm …...

如何配置一个sql server使得其它用户可以通过excel odbc获取数据

要让其他用户通过 Excel 使用 ODBC 连接到 SQL Server 获取数据&#xff0c;你需要完成以下配置步骤&#xff1a; ✅ 一、在 SQL Server 端配置&#xff08;服务器设置&#xff09; 1. 启用 TCP/IP 协议 打开 “SQL Server 配置管理器”。导航到&#xff1a;SQL Server 网络配…...

Visual Studio Code 扩展

Visual Studio Code 扩展 change-case 大小写转换EmmyLua for VSCode 调试插件Bookmarks 书签 change-case 大小写转换 https://marketplace.visualstudio.com/items?itemNamewmaurer.change-case 选中单词后&#xff0c;命令 changeCase.commands 可预览转换效果 EmmyLua…...

OpenGL-什么是软OpenGL/软渲染/软光栅?

‌软OpenGL&#xff08;Software OpenGL&#xff09;‌或者软渲染指完全通过CPU模拟实现的OpenGL渲染方式&#xff08;包括几何处理、光栅化、着色等&#xff09;&#xff0c;不依赖GPU硬件加速。这种模式通常性能较低&#xff0c;但兼容性极强&#xff0c;常用于不支持硬件加速…...