flink学习之窗口处理函数
窗口处理函数
什么是窗口处理函数
Flink 本身提供了多层 API,DataStream API 只是中间的一环,在更底层,我们可以不定义任何具体的算子(比如 map(),filter(),或者 window()),而只是提炼出一个统一的“处理”(process)操作——它是所有转换算子的一个概括性的表达,可以自定义处理逻辑,所以这一层接口就被叫作“处理函数”(process function)。

如何使用窗口处理函数
进行窗口计算,我们可以直接调用现成的简单聚合方法(sum()/max()/min()),也可以通过调用 reduce()或 aggregate()来自定义一般的增量聚合函数(ReduceFunction/AggregateFucntion);
而对于更加复杂、需要窗口信息和额外状态的一些场景,我们还可以直接使用全窗口函数、把数据全部收集保存在窗口内,等到触发窗口计算时再统一处理。窗口处理函数就是一种典型的全窗口函数。
窗 口 处 理 函 数 ProcessWindowFunction 的 使 用 与 其 他 窗 口 函 数 类 似 , 也 是 基 于
WindowedStream 直接调用方法就可以,只不过这时调用的是 process()。
stream.keyBy(_._1).window( TumblingEventTimeWindows.of(Time.seconds(10)) ).process(new MyProcessWindowFunction)
rocessWindowFunction 是一个继承了 AbstractRichFunction 的抽象类,它有四个类型
参数:
IN:input,数据流中窗口任务的输入数据类型。
OUT:output,窗口任务进行计算之后的输出数据类型。
KEY:数据中键 key 的类型。
W:窗口的类型,是 Window 的子类型。一般情况下我们定义时间窗口,W 就是
TimeWindow。
因为全窗口函数不是逐个处理元素的,所以处理数据的方法在这里并不是 processElement(),
而是改成了 process()。方法包含四个参数。
- key:窗口做统计计算基于的键,也就是之前 keyBy()用来分区的字段。
- context:当前窗口进行计算的上下文,它的类型就是
- ProcessWindowFunction 内部定 义的抽象类 Context。
- elements:窗口收集到用来计算的所有数据,这是一个可迭代的集合类型。 out:用来发送数据输出计算结果的收集器,类型为Collector。
可以明显看出,这里的参数 elements 不再是一个输入数据,而是窗口中所有数据的集合。
ProcessWindowFunction 中除.process()方法外,并没有 onTimer()方法,而是多出了一个
clear()方法,这主要是方便我们进行窗口的清理工作。
至于另一种窗口处理函数 ProcessAllWinfdowFunction,它的用法非常类似。区别在于它基
于的是 AllWindowedStream,相当于对没有 keyBy()的数据流直接开窗并调用 process()方法:
stream.windowAll( TumblingEventTimeWindows.of(Time.seconds(10)) )
.process(new MyProcessAllWindowFunction)
相关文章:
flink学习之窗口处理函数
窗口处理函数 什么是窗口处理函数 Flink 本身提供了多层 API,DataStream API 只是中间的一环,在更底层,我们可以不定义任何具体的算子(比如 map(),filter(),或者 window()),而只是…...
Python 基于pytorch从头写GPT模型;实现gpt实战
1.GPT简介 GPT(Generative Pre-trained Transformer)模型是一种基于Transformer架构的生成式预训练模型,由OpenAI开发。它采用了无监督学习的方式进行预训练,然后通过微调适应特定的任务。GPT模型的结构由多层Transformer解码器组…...
2023年NOC大赛(学而思赛道)创意编程Python初中组决赛真题
2023年NOC大赛(学而思赛道)创意编程Python初中组决赛真题 题目总数:7 总分数:100 编程题 第 1 题 问答题 二进制回文 编程实现: 输入一个正整数,判断它的二进制形式是否是回文数,如果是输出True…...
头歌C++之Switch控制语句编程实训
目录 第1关:根据输入数字判断是星期几 本关必读 本关任务 测试说明 第2关:根据输入的数值和运算符做相应运算 本关必读 本关任务 测试说明 第3关:根据输入年月计算该月份的天数 本关必读 本关任务...
CNN卷积理解
1 卷积的步骤 1 过滤器(卷积核)(Filter或Kernel): 卷积层使用一组可学习的过滤器来扫描输入数据(通常是图像)。每个过滤器都是一个小的窗口,包含一些权重,这些权重通过训…...
DataKit迁移MySQL到openGauss
前言 本文将分享DataKit迁移MySQL到openGauss的项目实战,供广大openGauss爱好者参考。 1. 下载操作系统 https://www.openeuler.org/zh/download https://support.huawei.com/enterprise/zh/doc/EDOC1100332931/1a643956 https://support.huawei.com/enterprise…...
Dockerfile里ADD * 保留原来的目录结构
1、问题 给新模块写Dockerfile,很多静态资源分散在各个目录,于是Dockerfile里我直接一句: ADD ./* /dest/镜像出来后,启动容器,进入容器种后发现:文件拷贝成功,但原来的目录结构都不在了&…...
C++ 利用容器适配器,仿函数实现栈,队列,优先级队列(堆),反向迭代器,deque的介绍与底层
C 利用容器适配器,仿函数实现栈,队列,优先级队列【堆】,反向迭代器,deque的介绍与底层 一.容器适配器的介绍二.利用容器适配器实现栈和队列1.stack2.queue 三.仿函数介绍1.什么是仿函数2.仿函数的使用3.函数指针的使用1.函数指针的用处2.利用函数指针完成回调3.利用仿函数完成回…...
C语言实战系列二:简单超市收银系统
从一个简单的超市收银系统,我们来练习一个系统如何设计,然后如何实现的思路。 在Ubuntu环境下使用C语言编写一个简单的超市收银系统。以下是一个基本的示例,涵盖了商品管理、购物车、交易处理等功能。 代码 #include <stdio.h> #inc…...
coding推送代码Jenkins自动构建部署
实现功能:我们向coding推送代码,通过webhook自动通知Jenkins,实现自动构建部署 coding 项目设置 / 开发者选项 / Service Hook 输入以下参数 发送POST请求服务 URL:htttp://xxx用户名:xxx密码:xxx Jen…...
Kettle-Docker部署+Sqlserver数据同步Mysql+Start定时任务
一. 背景介绍 1. ETL是什么 ETL(Extract-Transform-Load),即数据抽取、转换、装载的过程。它是一种思想,主要是说,从不同的数据源获取数据,并通过对数据进行处理(格式,协议等转换&a…...
《微信小程序开发从入门到实战》学习九十三
7.1 视图容器组件 7.1.3 swiper与swiper-item组件 swiper组件的显示效果如下图所示: indicator-dots、indicator-color和indicator-active-color三个属性用于设置swiper组件下方的指示点。设置指示点的颜色时,可以使用HexColor,也可以使用r…...
Java服务端使用freemarker+wkhtmltoimage生成Echart图片
目录 1.通过 freemarker 将ftl转成html 1.1 freemarker 手册: 1.2 添加freemarker maven依赖 1.3 添加 echart-test.ftl 模版文件 1.4 添加 FreemarkerTool 工具类 1.5 添加测试main方法 1.6 运行,生成echart-test-时间戳.html 文件 2. 通过wkhtmltoimage将html 转为p…...
一款颜值与实力并存的翻页时钟(免费)
FliTik是一款颜值与实力并存的翻页时钟,安卓端是完全免费的,无任何广告,极简风 ,软件默认是12小时制,可以在设置中启用24小时制,并且还支持设置显示秒钟、日期、文案,滴答声和语音报时。 支持横…...
Objective-C方法的声明实现及调用
1.无参数的方法 1)声明 a.位置:在interface括弧的外面 b.语法: - (返回值类型)方法名称; interface Person : NSObject -(void) run; end 2)实现 a.位置:在implementation中实现 b.语法:加大括弧将方法实现的代码写在大括孤之中 …...
第十四届蓝桥杯国赛 C++ B 组 C 题——班级活动(AC)
目录 1. 班级活动1. 问题描述2. 输入格式3. 输出格式4. 样例输入5. 样例输出6. 样例说明7. 评测用例规模与约定8. 原题链接 2. 解题思路3. AC_Code 1. 班级活动 前置知识点:思维,分类讨论 1. 问题描述 小明的老师准备组织一次班级活动。班上一共有 n…...
GraphQL的力量:简化复杂数据查询
1. GraphQL GraphQL 是一种由 Facebook 开发并于 2015 年公开发布的数据查询和操作语言,也是运行在服务端的运行时(runtime)用于处理 API 查询的一种规范。不同于传统的 REST API,GraphQL 允许客户端明确指定它们需要哪些数据&am…...
python环境安装sklearn及报错解决
安装 如刚开始安装,还未遇到问题请直接从重新安装库开始看,如果遇到报错,从问题开始看 问题 python安装sklearn报错 ,报错信息如下 File "<stdin>", line 1pip install scikit-learn^ SyntaxError: invalid s…...
log4j:WARN Please initialize the log4j system properly的解决办法
背景:很多次创建新项目log4j都出现以下2个警告: log4j:WARN No appenders could be found for logger (org.springframework.boot.ApplicationServletEnvironment).log4j:WARN Please initialize the log4j system properly 网上查询都是在说缺少以下jar…...
虹科分享丨汽车技术的未来:Netropy如何测试和确保汽车以太网的性能
来源:艾特保IT 虹科分享丨汽车技术的未来:Netropy如何测试和确保汽车以太网的性能 原文链接:https://mp.weixin.qq.com/s/G8wihrzqpJJOx5i0o63fkA 欢迎关注虹科,为您提供最新资讯! #汽车以太网 #车载网络 #Netropy …...
KubeSphere 容器平台高可用:环境搭建与可视化操作指南
Linux_k8s篇 欢迎来到Linux的世界,看笔记好好学多敲多打,每个人都是大神! 题目:KubeSphere 容器平台高可用:环境搭建与可视化操作指南 版本号: 1.0,0 作者: 老王要学习 日期: 2025.06.05 适用环境: Ubuntu22 文档说…...
解决本地部署 SmolVLM2 大语言模型运行 flash-attn 报错
出现的问题 安装 flash-attn 会一直卡在 build 那一步或者运行报错 解决办法 是因为你安装的 flash-attn 版本没有对应上,所以报错,到 https://github.com/Dao-AILab/flash-attention/releases 下载对应版本,cu、torch、cp 的版本一定要对…...
Axios请求超时重发机制
Axios 超时重新请求实现方案 在 Axios 中实现超时重新请求可以通过以下几种方式: 1. 使用拦截器实现自动重试 import axios from axios;// 创建axios实例 const instance axios.create();// 设置超时时间 instance.defaults.timeout 5000;// 最大重试次数 cons…...
UR 协作机器人「三剑客」:精密轻量担当(UR7e)、全能协作主力(UR12e)、重型任务专家(UR15)
UR协作机器人正以其卓越性能在现代制造业自动化中扮演重要角色。UR7e、UR12e和UR15通过创新技术和精准设计满足了不同行业的多样化需求。其中,UR15以其速度、精度及人工智能准备能力成为自动化领域的重要突破。UR7e和UR12e则在负载规格和市场定位上不断优化…...
今日学习:Spring线程池|并发修改异常|链路丢失|登录续期|VIP过期策略|数值类缓存
文章目录 优雅版线程池ThreadPoolTaskExecutor和ThreadPoolTaskExecutor的装饰器并发修改异常并发修改异常简介实现机制设计原因及意义 使用线程池造成的链路丢失问题线程池导致的链路丢失问题发生原因 常见解决方法更好的解决方法设计精妙之处 登录续期登录续期常见实现方式特…...
Springboot社区养老保险系统小程序
一、前言 随着我国经济迅速发展,人们对手机的需求越来越大,各种手机软件也都在被广泛应用,但是对于手机进行数据信息管理,对于手机的各种软件也是备受用户的喜爱,社区养老保险系统小程序被用户普遍使用,为方…...
Java 二维码
Java 二维码 **技术:**谷歌 ZXing 实现 首先添加依赖 <!-- 二维码依赖 --><dependency><groupId>com.google.zxing</groupId><artifactId>core</artifactId><version>3.5.1</version></dependency><de…...
基于Java+MySQL实现(GUI)客户管理系统
客户资料管理系统的设计与实现 第一章 需求分析 1.1 需求总体介绍 本项目为了方便维护客户信息为了方便维护客户信息,对客户进行统一管理,可以把所有客户信息录入系统,进行维护和统计功能。可通过文件的方式保存相关录入数据,对…...
安全突围:重塑内生安全体系:齐向东在2025年BCS大会的演讲
文章目录 前言第一部分:体系力量是突围之钥第一重困境是体系思想落地不畅。第二重困境是大小体系融合瓶颈。第三重困境是“小体系”运营梗阻。 第二部分:体系矛盾是突围之障一是数据孤岛的障碍。二是投入不足的障碍。三是新旧兼容难的障碍。 第三部分&am…...
JS手写代码篇----使用Promise封装AJAX请求
15、使用Promise封装AJAX请求 promise就有reject和resolve了,就不必写成功和失败的回调函数了 const BASEURL ./手写ajax/test.jsonfunction promiseAjax() {return new Promise((resolve, reject) > {const xhr new XMLHttpRequest();xhr.open("get&quo…...
