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

flink 最后一个窗口一直没有新数据,窗口不关闭问题

flink 最后一个窗口一直没有新数据,窗口不关闭问题

  • 自定义实现 WatermarkStrategy接口

自定义实现 WatermarkStrategy接口

窗口类型:滚动窗口
代码:

    public static class WatermarkDemoFunction implements WatermarkStrategy<JSONObject>{private Tuple2<Long,Boolean> state = Tuple2.of(0L,true);@Overridepublic WatermarkGenerator<JSONObject> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {return new WatermarkGenerator<JSONObject>() {private long maxWatermark;@Overridepublic void onEvent(JSONObject waterSensor, long l, WatermarkOutput watermarkOutput) {maxWatermark = Math.max(maxWatermark,waterSensor.getLong("ts"));state.f0 = System.currentTimeMillis();System.out.println("maxWatermark is " + maxWatermark);state.f1 = false;}@Overridepublic void onPeriodicEmit(WatermarkOutput watermarkOutput) {//乱序时间long outOfTime = 3000L;if (maxWatermark - outOfTime <=0){} else {// 10s内没有数据则关闭当前窗口System.out.println("System.currentTimeMillis() - state.f0:" + (System.currentTimeMillis() - state.f0));System.out.println("state.f1:" + state.f1);if (System.currentTimeMillis() - state.f0 >= 9000L && !state.f1){watermarkOutput.emitWatermark(new Watermark(maxWatermark  + 6000L));state.f1 = true;System.out.println("触发窗口,maxWatermark  + 6000L:" + (maxWatermark  + 6000L));} else {System.out.println("正常发送水印");watermarkOutput.emitWatermark(new Watermark(maxWatermark - outOfTime));}}}};}}

代码部分逻辑说明
在这里插入图片描述若设置了自动生成watermark 参数,根据打印日志,设置对应的时间(多久没新数据写入,触发窗口计算)
env.getConfig().setAutoWatermarkInterval(5000);

使用自定义的watermark:
在这里插入图片描述
watermark 周期生成()的疑问:
1、默认200ms,会连续生成4次后,不会继续生成了
2、设置了周期生成间隔,env.getConfig().setAutoWatermarkInterval(1000L); 只会周期生成一次

参考:https://blog.csdn.net/lr131425/article/details/127422833

相关文章:

flink 最后一个窗口一直没有新数据,窗口不关闭问题

flink 最后一个窗口一直没有新数据&#xff0c;窗口不关闭问题 自定义实现 WatermarkStrategy接口 自定义实现 WatermarkStrategy接口 窗口类型&#xff1a;滚动窗口 代码&#xff1a; public static class WatermarkDemoFunction implements WatermarkStrategy<JSONObject…...

mybatis----小细节

1、起别名 在MyBatis中&#xff0c;<typeAliases>元素用于定义类型别名&#xff0c;它可以将Java类名映射为一个更简短的别名&#xff0c;这样在映射文件中可以直接使用别名而不需要完整的类名。 下面是一个示例&#xff1a; 在mybatis核心配置文件中配置typeAliases标…...

解密Oracle数据库引擎:揭开数据存储的神秘面纱

目录 1、介绍Oracle数据库引擎 1.1 什么是Oracle数据库引擎 1.2 Oracle数据库引擎的作用和功能 1.3 Oracle数据库引擎的历史和发展 2、Oracle数据库引擎的体系结构 2.1 Oracle数据库实例的组成部分 2.2 Oracle数据库引擎的层次结构 2.3 Oracle数据库引擎的关键组件 3、…...

「HDLBits题解」Karnaugh Map to Circuit

本专栏的目的是分享可以通过HDLBits仿真的Verilog代码 以提供参考 各位可同时参考我的代码和官方题解代码 或许会有所收益 相关资料&#xff1a;卡诺图化简法-CSDN博客 题目链接&#xff1a;Kmap1 - HDLBits module top_module(input a,input b,input c,output out );assig…...

由于找不到d3dcompiler_43.dll缺失,无法打开软件的解决方法分享

d3dcompiler43.dll是什么文件&#xff1f;为什么会出现丢失的情况&#xff1f;又该如何解决呢&#xff1f;本文将详细介绍d3dcompiler43.dll的作用和影响&#xff0c;并提供6个有效的解决方法。 一、d3dcompiler43.dll是什么文件&#xff1f; d3dcompiler43.dll是DirectX SDK…...

现阶段Python和Java哪个更吃香?

现阶段Python和Java哪个更吃香&#xff1f; 在开始前我有一些资料&#xff0c;是我根据网友给的问题精心整理了一份「Java的资料从专业入门到高级教程」&#xff0c; 点个关注在评论区回复“888”之后私信回复“888”&#xff0c;全部无偿共享给大家&#xff01;&#xff01;&…...

基于DNA的密码学和隐写术综述

摘要 本文全面调研了不同的脱氧核糖核酸(DNA)-基于密码学和隐写术技术。基于DNA的密码学是一个新兴领域,利用DNA分子的大规模并行性和巨大的存储容量来编码和解码信息。近年来,由于其相对传统密码学方法的潜在优势,如高存储容量、低错误率和对环境因素的抗性,该领域引起…...

【linux 多线程并发】多线程的控制,挂起线程暂停运行,直到唤醒线程,取消线程运行,可以设置合适的取消点属性避免不安全点被中止

线程运行控制 ​专栏内容&#xff1a; 参天引擎内核架构 本专栏一起来聊聊参天引擎内核架构&#xff0c;以及如何实现多机的数据库节点的多读多写&#xff0c;与传统主备&#xff0c;MPP的区别&#xff0c;技术难点的分析&#xff0c;数据元数据同步&#xff0c;多主节点的情况…...

java转义字符

//转义字符的使用 public class ChangeChar{//编写一个main方法public static void main(String[] args){// \t :一个制表位&#xff0c;实现对齐的功能System.out.println("北京\t天津\t上海");// \n :换行符&#xff0c;实现换行System.out.println("jack\nsm…...

UDP服务器和客户端的创建步骤

UDP服务器的创建步骤&#xff1a; 一、创建用户数据报套接字&#xff08;socket函数&#xff09;&#xff1a;通信域选择IPV4网络协议、套接字类型选择数据报式&#xff1b; int sockfd socket(AF_INET,SOCK_DGRAM,0); 二、填充服务器的网络信息结构体&#xff1a;1.定义网络…...

【问题记录】使用命令语句从kaggle中下载数据集

从Kaggle中下载Tusimple数据集 1.服务器环境中安装kaggle 使用命令&#xff1a;pip install kaggle 2.复制下载API 具体命令如下&#xff1a; kaggle datasets download -d manideep1108/tusimple3.配置kaggle.json文件 如果直接使用命令会报错&#xff1a; root:~# kagg…...

Centos 更换内核

文章目录 一、查看/更换系统内核1.1 查看当前运行环境的内核1.2 查看系统上所有可用内核1.3 切换内核方法一&#xff1a;通过启动菜单更换内核方法二&#xff1a;更换默认启动内核 二、安装内核2.1 使用ELRepo安装2.2 安装指定内核版本参考资料 一、查看/更换系统内核 1.1 查看…...

RestTemplate调用Http方法

场景&#xff1a;前端调用后端的接口完成设备参数的更新&#xff0c;后端在接口代码中需要调用设备端提供的接口来完成设备参数的更新。 RestTemplate 是用于同步client端访问 Restful 服务的一个核心类 默认使用 JDK 提供的包去建立HTTP连接 为每种 HTTP 请求都实现了相关的…...

软考笔记--操作系统-内存管理

任何程序和数据都必须占用内存空间后才能执行&#xff0c;因此&#xff0c;内存管理的好坏直接影响整个操作系统的性能。操作系统的任务之一是尽可能地方便用户使用和提高内存的利用率。同时&#xff0c;内存管理也是多道程序设计系统的关键支撑。内存管理的功能主要包括以下几…...

Spring Boot整合日期转换器(Converter)和拦截器(HandlerInterceptor)

添加日期转换器&#xff08;Converter&#xff09; public class DataConverter implements Converter<String, Date> {Overridepublic Date convert(String s) {try {SimpleDateFormat simpleDateFormat new SimpleDateFormat("yyyy-MM-dd");return simpleD…...

100天精通Python(实用脚本篇)——第112天:图像处理批量给图片添加水印,防盗必备建议收藏!!!(附上脚本代码)

文章目录 专栏导读1. 图片设置水印的意义?2. 需求3. 实现效果4. 添加全屏水印4.1 单张图片添加全屏水印4.2 批量给多张图片添加全屏水印5. 添加指定位置水印5.1 模块介绍5.2 批量添加脚本实现专栏导读 🔥🔥本文已收录于《100天精通Python从入门到就业》:本专栏专门针对零…...

SpringMVC(全局异常处理.动态接收Ajax请求)

1.全局异常处理 1 异常处理器 基于AOP 用户发起请求, SpringMVC接受请求, SpringMVC加载静态资源问题说明 请求过去了,但没有处理 规则说明:静态资源进入SpringMVC框架之后,没有找到要怎样处理静态资源的方法,所以他们就不解决,也就不显示 解决方法:SpringMVC基于Servlet处理…...

【FastAPI】P1 简单实现 a+b

目录 准备工作代码运行 说明&#xff1a;本文通过 FastAPI 实现返回两个参数 ab 的值&#xff1b; 准备工作 默认读者已准备完善 Python IDE工具以及包管理工具。 首先&#xff0c;需要安装 fastapi 和 uvicorn 库&#xff0c;如果没有请使用 pip 进行安装&#xff1a; pip…...

C#设计模式教程(4):单例模式

单例模式(Singleton Pattern)是一种常用的设计模式,用于确保一个类只有一个实例,并提供一个访问它的全局访问点。在C#中实现单例模式通常有几种不同的方法,这里介绍几种典型的实现方式。 C# 单例模式的各种实现方式 1. 懒汉式(线程不安全) public class Singleton {p…...

常用的git diff命令用法汇总和示例

文章目录 1. 查看工作目录和暂存区的差异2. 查看暂存区和最后一次提交的差异3. 查看两个提交之间的差异4. 查看特定文件的更改5. 查看特定文件在两个提交之间的差异6. 查看分支之间的差异7. 查看某次提交的更改8. 限制diff输出的格式9. 查看一定时间范围内的更改 Git的diff命令…...

大型活动交通拥堵治理的视觉算法应用

大型活动下智慧交通的视觉分析应用 一、背景与挑战 大型活动&#xff08;如演唱会、马拉松赛事、高考中考等&#xff09;期间&#xff0c;城市交通面临瞬时人流车流激增、传统摄像头模糊、交通拥堵识别滞后等问题。以演唱会为例&#xff0c;暖城商圈曾因观众集中离场导致周边…...

为什么需要建设工程项目管理?工程项目管理有哪些亮点功能?

在建筑行业&#xff0c;项目管理的重要性不言而喻。随着工程规模的扩大、技术复杂度的提升&#xff0c;传统的管理模式已经难以满足现代工程的需求。过去&#xff0c;许多企业依赖手工记录、口头沟通和分散的信息管理&#xff0c;导致效率低下、成本失控、风险频发。例如&#…...

基于数字孪生的水厂可视化平台建设:架构与实践

分享大纲&#xff1a; 1、数字孪生水厂可视化平台建设背景 2、数字孪生水厂可视化平台建设架构 3、数字孪生水厂可视化平台建设成效 近几年&#xff0c;数字孪生水厂的建设开展的如火如荼。作为提升水厂管理效率、优化资源的调度手段&#xff0c;基于数字孪生的水厂可视化平台的…...

高防服务器能够抵御哪些网络攻击呢?

高防服务器作为一种有着高度防御能力的服务器&#xff0c;可以帮助网站应对分布式拒绝服务攻击&#xff0c;有效识别和清理一些恶意的网络流量&#xff0c;为用户提供安全且稳定的网络环境&#xff0c;那么&#xff0c;高防服务器一般都可以抵御哪些网络攻击呢&#xff1f;下面…...

Linux --进程控制

本文从以下五个方面来初步认识进程控制&#xff1a; 目录 进程创建 进程终止 进程等待 进程替换 模拟实现一个微型shell 进程创建 在Linux系统中我们可以在一个进程使用系统调用fork()来创建子进程&#xff0c;创建出来的进程就是子进程&#xff0c;原来的进程为父进程。…...

网站指纹识别

网站指纹识别 网站的最基本组成&#xff1a;服务器&#xff08;操作系统&#xff09;、中间件&#xff08;web容器&#xff09;、脚本语言、数据厍 为什么要了解这些&#xff1f;举个例子&#xff1a;发现了一个文件读取漏洞&#xff0c;我们需要读/etc/passwd&#xff0c;如…...

基于Java+MySQL实现(GUI)客户管理系统

客户资料管理系统的设计与实现 第一章 需求分析 1.1 需求总体介绍 本项目为了方便维护客户信息为了方便维护客户信息&#xff0c;对客户进行统一管理&#xff0c;可以把所有客户信息录入系统&#xff0c;进行维护和统计功能。可通过文件的方式保存相关录入数据&#xff0c;对…...

【从零学习JVM|第三篇】类的生命周期(高频面试题)

前言&#xff1a; 在Java编程中&#xff0c;类的生命周期是指类从被加载到内存中开始&#xff0c;到被卸载出内存为止的整个过程。了解类的生命周期对于理解Java程序的运行机制以及性能优化非常重要。本文会深入探寻类的生命周期&#xff0c;让读者对此有深刻印象。 目录 ​…...

纯 Java 项目(非 SpringBoot)集成 Mybatis-Plus 和 Mybatis-Plus-Join

纯 Java 项目&#xff08;非 SpringBoot&#xff09;集成 Mybatis-Plus 和 Mybatis-Plus-Join 1、依赖1.1、依赖版本1.2、pom.xml 2、代码2.1、SqlSession 构造器2.2、MybatisPlus代码生成器2.3、获取 config.yml 配置2.3.1、config.yml2.3.2、项目配置类 2.4、ftl 模板2.4.1、…...

vue3 daterange正则踩坑

<el-form-item label"空置时间" prop"vacantTime"> <el-date-picker v-model"form.vacantTime" type"daterange" start-placeholder"开始日期" end-placeholder"结束日期" clearable :editable"fal…...