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

0基础学习PyFlink——时间滑动窗口(Sliding Time Windows)

在《0基础学习PyFlink——时间滚动窗口(Tumbling Time Windows)》我们介绍了不会有重复数据的时间滚动窗口。本节我们将介绍存在重复计算数据的时间滑动窗口。
关于滑动窗口,可以先看下《0基础学习PyFlink——个数滑动窗口(Sliding Count Windows)》。下图就是个数滑动窗口示意图。
在这里插入图片描述
我们看到个数滑动窗口也会因为窗口内数据不够而不被触发。但是时间滑动窗口则可以解决这个问题,我们只要把窗口改成时间类型即可。
相应的代码我们参考《0基础学习PyFlink——时间滚动窗口(Tumbling Time Windows)》,只要把TumblingProcessingTimeWindows改成SlidingProcessingTimeWindows,并增加一个偏移参数(Time.milliseconds(1))即可。这意味着我们将运行一个时间长度为2毫秒,每次递进1毫秒的窗口。

完整代码

from typing import Iterable
import time
from pyflink.common import Types, Time
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, WindowFunction
from pyflink.datastream.window import  TimeWindow, SlidingProcessingTimeWindowsclass SumWindowFunction(WindowFunction[tuple, tuple, str, TimeWindow]):def apply(self, key: str, window: TimeWindow, inputs: Iterable[tuple]):print(*inputs, window)return [(key,  len([e for e in inputs]))]word_count_data = [("A",2),("A",1),("A",4),("A",3),("A",6),("A",5),("A",7),("A",8),("A",9),("A",10),("A",11),("A",12),("A",13),("A",14),("A",15),("A",16),("A",17),("A",18),("A",19),("A",20)]def word_count():env = StreamExecutionEnvironment.get_execution_environment()env.set_runtime_mode(RuntimeExecutionMode.STREAMING)# write all the data to one fileenv.set_parallelism(1)source_type_info = Types.TUPLE([Types.STRING(), Types.INT()])# define the source# mappgingsource = env.from_collection(word_count_data, source_type_info)# source.print()# keyingkeyed=source.key_by(lambda i: i[0]) # reducingreduced=keyed.window(SlidingProcessingTimeWindows.of(Time.milliseconds(2), Time.milliseconds(1))) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))# # define the sinkreduced.print()# submit for executionenv.execute()if __name__ == '__main__':word_count()

运行结果

运行两次上述代码,我们发现每次都不同,而且有重叠计算的元素。

(‘A’, 2) (‘A’, 1) (‘A’, 4) TimeWindow(start=1698773292650, end=1698773292652)
(‘A’, 2) (‘A’, 1) (‘A’, 4) (‘A’, 3) (‘A’, 6) (‘A’, 5) (‘A’, 7) (‘A’, 8) (‘A’, 9) (‘A’, 10) (‘A’, 11) TimeWindow(start=1698773292651, end=1698773292653)
(A,3)
(A,11)
(‘A’, 3) (‘A’, 6) (‘A’, 5) (‘A’, 7) (‘A’, 8) (‘A’, 9) (‘A’, 10) (‘A’, 11) (‘A’, 12) (‘A’, 13) (‘A’, 14) (‘A’, 15) (‘A’, 16) (‘A’, 17) (‘A’, 18) (‘A’, 19) (‘A’, 20) TimeWindow(start=1698773292652, end=1698773292654)
(A,17)

在这里插入图片描述

(‘A’, 2) (‘A’, 1) (‘A’, 4) TimeWindow(start=1698773319933, end=1698773319935)
(‘A’, 2) (‘A’, 1) (‘A’, 4) (‘A’, 3) (‘A’, 6) (‘A’, 5) (‘A’, 7) (‘A’, 8) (‘A’, 9) (‘A’, 10) (‘A’, 11) (‘A’, 12) TimeWindow(start=1698773319934, end=1698773319936)
(A,3)
(A,12)
(‘A’, 3) (‘A’, 6) (‘A’, 5) (‘A’, 7) (‘A’, 8) (‘A’, 9) (‘A’, 10) (‘A’, 11) (‘A’, 12) (‘A’, 13) (‘A’, 14) (‘A’, 15) (‘A’, 16) (‘A’, 17) (‘A’, 18) (‘A’, 19) (‘A’, 20) TimeWindow(start=1698773319935, end=1698773319937)
(A,17)

在这里插入图片描述

参考资料

  • https://nightlies.apache.org/flink/flink-docs-master/api/python/reference/pyflink.datastream/api/pyflink.datastream.window.SlidingProcessingTimeWindows.html#pyflink.datastream.window.SlidingProcessingTimeWindows

相关文章:

0基础学习PyFlink——时间滑动窗口(Sliding Time Windows)

在《0基础学习PyFlink——时间滚动窗口(Tumbling Time Windows)》我们介绍了不会有重复数据的时间滚动窗口。本节我们将介绍存在重复计算数据的时间滑动窗口。 关于滑动窗口,可以先看下《0基础学习PyFlink——个数滑动窗口(Sliding Count Windows&#x…...

API安全之《大话:API的前世今生》

写在前面:本文结合API使用的业界现状,系统性地阐述API的基本概念、发展历史、表现形式等基础内容,主要包含以下内容: 1.什么是API 2.API的发展历史 3.现代API常用消息格式 4.top N 互联网企业API 使用现状 当前的世界是一个信…...

H5或者Vue实现二维码识别

前言 1、扫码识别库采用开源的zxing/library 2、支持js&#xff0c;Vue&#xff0c;lit等实现 原文章地址和代码仓库地址 1、在界面创建video标签用来显示摄像头内容 <!-- 视区 --><!-- lit写法 --> <video ${ref(this.videoRef)} class"xy-scan-wrap…...

stm32整理(三)ADC

1 ADC简介 1.1 ADC 简介 12 位 ADC 是逐次趋近型模数转换器。它具有多达 19 个复用通道&#xff0c;可测量来自 16 个外部 源、两个内部源和 VBAT 通道的信号。这些通道的 A/D 转换可在单次、连续、扫描或不连续 采样模式下进行。ADC 的结果存储在一个左对齐或右对齐的 16 位…...

Redis-持久化+主从架构

文章目录 Redis的持久化RDB模式异步持久化的实现AOF模式总结 Redis的主从架构1.端口以及文件调试测试2.主从配置3.数据同步原理&#xff08;第一次同步为全局同步&#xff09;4.增量同步5.主从配置优化6.问:master主机怎么判断从机slave是不是第一次同步数据&#xff1f; Redis…...

STM32H750之FreeRTOS学习--------(四)中断管理

四、FreeRTOS中断管理 中断的概念不再过多叙述&#xff0c;学习过逻辑的都知道 中断的执行过程 中断请求 外设产生中断请求&#xff08;GPIO外部中断、定时器中断等&#xff09;响应中断 CPU停止执行当前程序&#xff0c;转而去执行中断处理程序&#xff08;ISR&#xff09;…...

Macroscope安全漏洞检测工具简介

学习目标&#xff1a; 本介绍旨在帮助感兴趣者尽快了解 Macroscope&#xff0c;这是一款用于安全测试自动化和漏洞管理的企业工具。 全覆盖应用程序安全测试&#xff1a; 如下图所示&#xff0c;如果使用多种互补工具&#xff08;SAST/DAST/SCA 等&#xff09;来检测应用程序…...

【Linux】Nignx的入门使用负载均衡动静分离(前后端项目部署)---超详细

一&#xff0c;Nignx入门 1.1 Nignx是什么 Nginx是一个高性能的开源Web服务器和反向代理服务器。它使用事件驱动的异步框架&#xff0c;可同时处理大量请求&#xff0c;支持负载均衡、反向代理、HTTP缓存等常见Web服务场景。Nginx可以作为一个前端的Web服务器&#xff0c;也可…...

【入门Flink】- 04Flink部署模式和运行模式【偏概念】

部署模式 在一些应用场景中&#xff0c;对于集群资源分配和占用的方式&#xff0c;可能会有特定的需求。Flink为各种场景提供了不同的部署模式&#xff0c;主要有以下三种&#xff1a;会话模式&#xff08;Session Mode&#xff09;、单作业模式&#xff08;Per-Job Mode&…...

react面试要点

# React面试知识点 ## React是什么&#xff1f;谈一谈你对react的理解 1 React是一个网页UI库 2 react的特点是 声明式 组件化 通用性 3 react优点&#xff1a; 简单&#xff0c;低耦合高内聚&#xff0c;由于虚拟dom概念&#xff0c;可以做到一次学习到处使用。 …...

在Google Kubernetes集群创建分布式Jenkins(一)

因为项目需要&#xff0c;在GKE的集群上需要创建一个CICD的环境&#xff0c;记录一下安装部署一个分布式Jenkins集群的过程。 分布式Jenkins由一个主服务器和多个Agent组成&#xff0c;Agent可以执行主服务器分派的任务。如下图所示&#xff1a; 如上图&#xff0c;Jenkins Ag…...

【Python全栈_公开课学习记录】

一、初识python (一).Python起源 Python创始人为吉多范罗苏姆&#xff08;荷兰&#xff09;&#xff0c;Python崇尚优美、清晰、简明的编辑风格。Python语言结构清晰简单、数据库丰富、运行成熟稳定&#xff0c;科学计算统计分析领先。目前广泛应用于云计算、Web开发、科学运算…...

uniapp循环列表单选框实现单选

目录 图片源码参考最后 图片 源码 参考 大佬 最后 感觉文章好的话记得点个心心和关注和收藏&#xff0c;有错的地方麻烦指正一下&#xff0c;如果需要转载,请标明出处&#xff0c;多谢&#xff01;&#xff01;&#xff01;...

【强化学习】14 —— A3C(Asynchronous Advantage Actor Critic)

A3C算法&#xff08; Asynchronous Methods for Deep Reinforcement Learning&#xff09;于2016年被谷歌DeepMind团队提出。A3C是一种非常有效的深度强化学习算法&#xff0c;在围棋、星际争霸等复杂任务上已经取得了很好的效果。接下来&#xff0c;我们先从A3C的名称入手&…...

Google单元测试sample分析(四)

GoogleTest单元测试可用实现在每个测试用例结束后监控其内存使用情况&#xff0c; 可以通过GoogleTest提供的事件侦听器EmptyTestEventListener 来实现&#xff0c;下面通过官方提供的sample例子&#xff0c;路径在samples文件夹下的sample10_unittest.cpp // Copyright 2009…...

网络套接字编程(二)

网络套接字编程(二) 文章目录 网络套接字编程(二)简易TCP网络程序服务端创建套接字服务端绑定IP地址和端口号服务端监听服务端运行服务端网络服务服务端启动客户端创建套接字客户端的绑定和监听问题客户端建立连接并通信客户端启动程序测试单执行流服务器的弊端 多进程版TCP网络…...

LLaMA-Adapter源码解析

LLaMA-Adapter源码解析 伪代码 def transformer_block_with_llama_adapter(x, gating_factor, soft_prompt):residual xy zero_init_attention(soft_prompt, x) # llama-adapter: prepend prefixx self_attention(x)x x gating_factor * y # llama-adapter: apply zero_init…...

JavaScript设计模式之发布-订阅模式

发布者和订阅者完全解耦&#xff08;通过消息队列进行通信&#xff09; 适用场景&#xff1a;功能模块间进行通信&#xff0c;如Vue的事件总线。 ES6实现方式&#xff1a; class eventManager {constructor() {this.eventList {};}on(eventName, callback) {if (this.eventL…...

mysql---索引

概要 索引&#xff1a;排序的列表&#xff0c;列表当中存储的是索引的值和包含这个值的数据所在的行的物理地址 作用&#xff1a;加快查找速度 注&#xff1a;索引要在创建表时尽量创建完全&#xff0c;后期添加影响变动大。 索引也需要占用磁盘空间&#xff0c;innodb表数据…...

微信小程序——简易复制文本

在微信小程序中&#xff0c;可以使用wx.setClipboardData()方法来实现复制文本内容的功能。以下是一个示例代码&#xff1a; // 点击按钮触发复制事件 copyText: function() {var that this;wx.setClipboardData({data: 要复制的文本内容,success: function(res) {wx.showToa…...

深度学习在微纳光子学中的应用

深度学习在微纳光子学中的主要应用方向 深度学习与微纳光子学的结合主要集中在以下几个方向&#xff1a; 逆向设计 通过神经网络快速预测微纳结构的光学响应&#xff0c;替代传统耗时的数值模拟方法。例如设计超表面、光子晶体等结构。 特征提取与优化 从复杂的光学数据中自…...

Vue记事本应用实现教程

文章目录 1. 项目介绍2. 开发环境准备3. 设计应用界面4. 创建Vue实例和数据模型5. 实现记事本功能5.1 添加新记事项5.2 删除记事项5.3 清空所有记事 6. 添加样式7. 功能扩展&#xff1a;显示创建时间8. 功能扩展&#xff1a;记事项搜索9. 完整代码10. Vue知识点解析10.1 数据绑…...

Linux 文件类型,目录与路径,文件与目录管理

文件类型 后面的字符表示文件类型标志 普通文件&#xff1a;-&#xff08;纯文本文件&#xff0c;二进制文件&#xff0c;数据格式文件&#xff09; 如文本文件、图片、程序文件等。 目录文件&#xff1a;d&#xff08;directory&#xff09; 用来存放其他文件或子目录。 设备…...

day52 ResNet18 CBAM

在深度学习的旅程中&#xff0c;我们不断探索如何提升模型的性能。今天&#xff0c;我将分享我在 ResNet18 模型中插入 CBAM&#xff08;Convolutional Block Attention Module&#xff09;模块&#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.构…...

从零开始打造 OpenSTLinux 6.6 Yocto 系统(基于STM32CubeMX)(九)

设备树移植 和uboot设备树修改的内容同步到kernel将设备树stm32mp157d-stm32mp157daa1-mx.dts复制到内核源码目录下 源码修改及编译 修改arch/arm/boot/dts/st/Makefile&#xff0c;新增设备树编译 stm32mp157f-ev1-m4-examples.dtb \stm32mp157d-stm32mp157daa1-mx.dtb修改…...

Module Federation 和 Native Federation 的比较

前言 Module Federation 是 Webpack 5 引入的微前端架构方案&#xff0c;允许不同独立构建的应用在运行时动态共享模块。 Native Federation 是 Angular 官方基于 Module Federation 理念实现的专为 Angular 优化的微前端方案。 概念解析 Module Federation (模块联邦) Modul…...

零基础设计模式——行为型模式 - 责任链模式

第四部分&#xff1a;行为型模式 - 责任链模式 (Chain of Responsibility Pattern) 欢迎来到行为型模式的学习&#xff01;行为型模式关注对象之间的职责分配、算法封装和对象间的交互。我们将学习的第一个行为型模式是责任链模式。 核心思想&#xff1a;使多个对象都有机会处…...

Java 二维码

Java 二维码 **技术&#xff1a;**谷歌 ZXing 实现 首先添加依赖 <!-- 二维码依赖 --><dependency><groupId>com.google.zxing</groupId><artifactId>core</artifactId><version>3.5.1</version></dependency><de…...

在Ubuntu24上采用Wine打开SourceInsight

1. 安装wine sudo apt install wine 2. 安装32位库支持,SourceInsight是32位程序 sudo dpkg --add-architecture i386 sudo apt update sudo apt install wine32:i386 3. 验证安装 wine --version 4. 安装必要的字体和库(解决显示问题) sudo apt install fonts-wqy…...