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

0基础学习PyFlink——个数滑动窗口(Sliding Count Windows)

大纲

  • 滑动(Sliding)和滚动(Tumbling)的区别
  • 样例
    • 窗口为2,滑动距离为1
    • 窗口为3,滑动距离为1
    • 窗口为3,滑动距离为2
    • 窗口为3,滑动距离为3
  • 完整代码
  • 参考资料

在 《0基础学习PyFlink——个数滚动窗口(Tumbling Count Windows)》一文中,我们介绍了滚动窗口。本节我们要介绍滑动窗口。

滑动(Sliding)和滚动(Tumbling)的区别

正如其名,“滑动”是指这个窗口沿着一定的方向,按着一定的速度“滑行”。
在这里插入图片描述
而滚动窗口,则是一个个“衔接着”,而不是像上面那样交错着。
在这里插入图片描述
它们的相同之处就是:只有窗口内的事件数量到达窗口要求的数值时,这些窗口才会触发计算。

样例

我们只要对《0基础学习PyFlink——个数滚动窗口(Tumbling Count Windows)》中的代码做轻微的改动即可。为了简化样例,我们只看Key为E的元素的滑动。

word_count_data = [("E",3),("E",1),("E",4),("E",2),("E",6),("E",5)]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]) 

窗口为2,滑动距离为1

count_window会根据传入的第二参数决定是构建滚动(CountTumblingWindowAssigner)窗口还是滑动(CountSlidingWindowAssigner)窗口。

    def count_window(self, size: int, slide: int = 0):"""Windows this KeyedStream into tumbling or sliding count windows.:param size: The size of the windows in number of elements.:param slide: The slide interval in number of elements... versionadded:: 1.16.0"""if slide == 0:return WindowedStream(self, CountTumblingWindowAssigner(size))else:return WindowedStream(self, CountSlidingWindowAssigner(size, slide))

我们只要给count_window第二个参数传递一个不为0的值,即可达到滑动效果。

    # reducingwindows_size = 2sliding_size = 1reduced=keyed.count_window(windows_size, sliding_size) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))# # define the sinkreduced.print()# submit for executionenv.execute()

(E,2)
(E,2)
(E,2)
(E,2)
(E,2)

在这里插入图片描述

窗口为3,滑动距离为1

    # reducingwindows_size = 3sliding_size = 1reduced=keyed.count_window(windows_size, sliding_size) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))

(E,3)
(E,3)
(E,3)
(E,3)

在这里插入图片描述

窗口为3,滑动距离为2

    # reducingwindows_size = 3sliding_size = 2reduced=keyed.count_window(windows_size, sliding_size) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))

(E,3)
(E,3)

在这里插入图片描述

窗口为3,滑动距离为3

这个就等效于滚动窗口了,因为“滑”过了窗口大小。

    # reducingwindows_size = 3sliding_size = 3reduced=keyed.count_window(windows_size, sliding_size) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))

(E,3)
(E,3)

在这里插入图片描述

完整代码

from typing import Iterablefrom pyflink.common import Types
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, WindowFunction
from pyflink.datastream.window import CountWindowclass SumWindowFunction(WindowFunction[tuple, tuple, str, CountWindow]):def apply(self, key: str, window: CountWindow, inputs: Iterable[tuple]):return [(key,  len([e for e in inputs]))]word_count_data = [("E",3),("E",1),("E",4),("E",2),("E",6),("E",5)]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]) # reducingwindows_size = 3sliding_size = 1reduced=keyed.count_window(windows_size, sliding_size) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))# # define the sinkreduced.print()# submit for executionenv.execute()if __name__ == '__main__':word_count()

参考资料

  • https://nightlies.apache.org/flink/flink-docs-release-1.18/zh/docs/learn-flink/streaming_analytics/

相关文章:

0基础学习PyFlink——个数滑动窗口(Sliding Count Windows)

大纲 滑动(Sliding)和滚动(Tumbling)的区别样例窗口为2,滑动距离为1窗口为3,滑动距离为1窗口为3,滑动距离为2窗口为3,滑动距离为3 完整代码参考资料 在 《0基础学习PyFlink——个数…...

vue3+ts 提取公共方法

因为好多页面都会使用到这个效验规则,封装一个校检规则,方便维护 封装前 封装后...

C++ ->

C -> 是访问类或结构体对象的成员的运算符 注意这里不是直接的访问.是用于访问指向对象的指针的成员 下面的代码可以很好的理解如下&#xff1a; #include<iostream>using namespace std;class Func{public:int i,j;void myFunc(){cout<<"i"<&l…...

VR全景在医院的应用:缓和医患矛盾、提升医院形象

医患关系一直以来都是较为激烈的&#xff0c;包括制度的不完善、医疗资源紧张等问题也时有存在&#xff0c;为了缓解医患矛盾&#xff0c;不仅要提升患者以及家属对于医院的认知&#xff0c;还需要完善医疗制度&#xff0c;提高医疗资源的配置效率&#xff0c;提高服务质量。 因…...

【python基础】format格式化函数的使用

文章目录 前言一、format()内容匹配替换1、序号索引2、关键字3、列表索引4、字典索引5、通过类的属性6、通过魔法参数 二、format()数字格式化 前言 语法&#xff1a;str.format() 说明&#xff1a;一种格式化字符串的函数。 一、format()内容匹配替换 1、序号索引 在没有参…...

Java web(三):Http、Tomcat、Servlet

文章目录 一、Java web技术栈二、Http1.1 Http请求数据格式1.2 Http响应数据格式1.3 状态码 二、Tomcat2.1 介绍2.2 web项目结构2.3 IDEA中使用Tomcat 三、Servlet3.1 Servlet使用3.2 Servlet生命周期3.3 Servlet方法和体系结构3.4 urlPattern配置 四、Request4.1 获取请求数据…...

Java实现Hive UDF详细步骤 (Hive 3.x版本,IDEA开发)

这里写目录标题 前言1. 新建项目2.配置maven依赖3.编写代码4.打jar包5.上传服务器6.代码中引用 前言 老版本编写UDF时&#xff0c;需要继承 org.apache.hadoop.hive.ql.exec.UDF类&#xff0c;然后直接实现evaluate()方法即可。 由于公司hive版本比较高&#xff08;3.x&#x…...

Vue进阶(幺陆肆)Apache的Access.log分析总结

文章目录 一、前言二、常用指令 一、前言 前端项目排错阶段&#xff0c;可借助apache的Access.log进行请求日志查看。 二、常用指令 #查看80端口的tcp连接 #netstat -tan | grep "ESTABLISHED" | grep ":80" | wc -l #当前WEB服务器中联接次数最多的ip地…...

Apple 苹果发布 M3、M3 Pro 和 M3 Max 芯片

本心、输入输出、结果 文章目录 Apple 苹果发布 M3、M3 Pro 和 M3 Max 芯片前言M3、M3 Pro 和 M3 Max 芯片的性能相关资料图M3 Pro规格M3 Max规格弘扬爱国精神 Apple 苹果发布 M3、M3 Pro 和 M3 Max 芯片 编辑&#xff1a;简简单单 Online zuozuo 地址&#xff1a;https://blog…...

Linux常用命令及主流服务部署大全

目录 Linux 系统目录 一、常用操作命令 1、目录操作 2、文件内容操作&#xff08;查看日志&#xff0c;更改配置文件&#xff09; 3、压缩和解压缩 4、更改文件权限 二、各服务部署命令 1、增加虚拟内存 2、JDK 2.1 删除系统自带的openjdk 2.2 安装jdk 2.3 删除jd…...

list-watch集群调度

调度约束 Kubernetes 是通过 List-Watch **** 的机制进行每个组件的协作&#xff0c;保持数据同步的&#xff0c;每个组件之间的设计实现了解耦。 用户是通过 kubectl 根据配置文件&#xff0c;向 APIServer 发送命令&#xff0c;在 Node 节点上面建立 Pod 和 Container。…...

深度强化学习中的神经网络部分的作用是什么?一般如何选择合适的神经网络呢?

在深度强化学习中&#xff0c;神经网络部分通常用于实现值函数近似或策略近似&#xff0c;以帮助智能体学习如何在一个环境中做出决策以获得最大的累积奖励。这些神经网络在深度强化学习中扮演着重要的角色&#xff0c;具体作用如下&#xff1a; 1.值函数近似&#xff08;Valu…...

若依系统的数据导入功能设置

一、后端 Log(title "公交站牌", businessType BusinessType.IMPORT)PreAuthorize("ss.hasPermi(busStop:busStop:import)")PostMapping("/importData")public AjaxResult importData(MultipartFile file, boolean updateSupport) throws Exce…...

vue页面父组件与子组件相互调用方法和传递参数值

vue页面父组件与子组件相互调用方法和传递参数值 父组件页面定义 <el-button type"text" icon"el-icon-refresh" click"refreshClick" slot"label"></el-button> <leftList leftClick"loadModelClick" r…...

vim使用

概述 vi&#xff08;visual editor&#xff09;是Unix/Linux编辑器的一种。类似于win中notepad。vim&#xff08;vi improved&#xff09;加强版 安装vim&#xff1a; $ yum install vim -y四种模式 命令模式&#xff1a;快速进行复制、粘贴、删除等操作&#xff0c;还可以…...

人工智能基础_机器学习014_BGD批量梯度下降公式更新_进一步推导_SGD随机梯度下降和MBGD小批量梯度下降公式进一步推导---人工智能工作笔记0054

然后我们先来看BGD批量梯度下降,可以看到这里,其实这个公式来源于 梯度下降的公式对吧,其实就是对原始梯度下降公式求偏导以后的梯度下降公式,然后 使用所有样本进行梯度下降得来的,可以看到* 1/n 其实就是求了一个平均数对吧.所有样本的平均数. 然后我们看,我们这里* 1/n那么…...

Android STR研究之一

简介&#xff1a; 先上一段谷歌的介绍 谷歌的网站地址&#xff1a; 电源管理 | Android 开源项目 | Android Open Source Project (google.cn) 术语 STR&#xff1a; STR(Suspend To RAM)的意思是“挂起到内存”,它是一种瞬间开机技术(On Now)。 当系统进入“挂起”状态…...

单链表的详解实现

单链表 结构 单链表结构中有两个数据&#xff0c;一个是存储数据的&#xff0c;还有一个指针指向下一个节点。 该图就是一个简单单链表的结构图。 接口实现 SLNode* CreateNode(SLNDataType x);//申请节点 void SLTprint(SLNode* head);//打印链表 void SLTPushBack(SLNode*…...

抛弃 scp 改用 rsync,让 Linux 下文件传输高效无比

我们都使用过 scp 来传输文件。当传输在中途或甚至在 99% 时被中断时&#xff0c;&#xff08;每当我想起99%的中断传输时&#xff0c;我的心都很痛&#xff09;&#xff1b;让我们看看如何使用 rsync 来替代 scp&#xff0c;避免这样的不幸。 什么是rsync&#xff1f; Rsync…...

Leetcode 2919. Minimum Increment Operations to Make Array Beautiful

Leetcode 2919. Minimum Increment Operations to Make Array Beautiful 1. 解题思路2. 代码实现 题目链接&#xff1a;2919. Minimum Increment Operations to Make Array Beautiful 1. 解题思路 这一题就是一个动态规划的题目。 思路上来说&#xff0c;就是考察每一个没到…...

相机从app启动流程

一、流程框架图 二、具体流程分析 1、得到cameralist和对应的静态信息 目录如下: 重点代码分析: 启动相机前,先要通过getCameraIdList获取camera的个数以及id,然后可以通过getCameraCharacteristics获取对应id camera的capabilities(静态信息)进行一些openCamera前的…...

微服务商城-商品微服务

数据表 CREATE TABLE product (id bigint(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 商品id,cateid smallint(6) UNSIGNED NOT NULL DEFAULT 0 COMMENT 类别Id,name varchar(100) NOT NULL DEFAULT COMMENT 商品名称,subtitle varchar(200) NOT NULL DEFAULT COMMENT 商…...

什么是EULA和DPA

文章目录 EULA&#xff08;End User License Agreement&#xff09;DPA&#xff08;Data Protection Agreement&#xff09;一、定义与背景二、核心内容三、法律效力与责任四、实际应用与意义 EULA&#xff08;End User License Agreement&#xff09; 定义&#xff1a; EULA即…...

零基础在实践中学习网络安全-皮卡丘靶场(第九期-Unsafe Fileupload模块)(yakit方式)

本期内容并不是很难&#xff0c;相信大家会学的很愉快&#xff0c;当然对于有后端基础的朋友来说&#xff0c;本期内容更加容易了解&#xff0c;当然没有基础的也别担心&#xff0c;本期内容会详细解释有关内容 本期用到的软件&#xff1a;yakit&#xff08;因为经过之前好多期…...

html css js网页制作成品——HTML+CSS榴莲商城网页设计(4页)附源码

目录 一、&#x1f468;‍&#x1f393;网站题目 二、✍️网站描述 三、&#x1f4da;网站介绍 四、&#x1f310;网站效果 五、&#x1fa93; 代码实现 &#x1f9f1;HTML 六、&#x1f947; 如何让学习不再盲目 七、&#x1f381;更多干货 一、&#x1f468;‍&#x1f…...

【生成模型】视频生成论文调研

工作清单 上游应用方向&#xff1a;控制、速度、时长、高动态、多主体驱动 类型工作基础模型WAN / WAN-VACE / HunyuanVideo控制条件轨迹控制ATI~镜头控制ReCamMaster~多主体驱动Phantom~音频驱动Let Them Talk: Audio-Driven Multi-Person Conversational Video Generation速…...

安宝特方案丨船舶智造的“AR+AI+作业标准化管理解决方案”(装配)

船舶制造装配管理现状&#xff1a;装配工作依赖人工经验&#xff0c;装配工人凭借长期实践积累的操作技巧完成零部件组装。企业通常制定了装配作业指导书&#xff0c;但在实际执行中&#xff0c;工人对指导书的理解和遵循程度参差不齐。 船舶装配过程中的挑战与需求 挑战 (1…...

【Go语言基础【13】】函数、闭包、方法

文章目录 零、概述一、函数基础1、函数基础概念2、参数传递机制3、返回值特性3.1. 多返回值3.2. 命名返回值3.3. 错误处理 二、函数类型与高阶函数1. 函数类型定义2. 高阶函数&#xff08;函数作为参数、返回值&#xff09; 三、匿名函数与闭包1. 匿名函数&#xff08;Lambda函…...

【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 日志分析…...

GraphQL 实战篇:Apollo Client 配置与缓存

GraphQL 实战篇&#xff1a;Apollo Client 配置与缓存 上一篇&#xff1a;GraphQL 入门篇&#xff1a;基础查询语法 依旧和上一篇的笔记一样&#xff0c;主实操&#xff0c;没啥过多的细节讲解&#xff0c;代码具体在&#xff1a; https://github.com/GoldenaArcher/graphql…...