大数据-120 - Flink Window 窗口机制-滑动时间窗口、会话窗口-基于时间驱动基于事件驱动

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

  • Hadoop（已更完）
  • HDFS（已更完）
  • MapReduce（已更完）
  • Hive（已更完）
  • Flume（已更完）
  • Sqoop（已更完）
  • Zookeeper（已更完）
  • HBase（已更完）
  • Redis （已更完）
  • Kafka（已更完）
  • Spark（已更完）
  • Flink（正在更新！）

章节内容

上节我们完成了如下的内容：

  • Flink Window 背景总览
  • Flink Window 滚动时间窗口
  • 基于时间驱动
  • 基于事件驱动

在这里插入图片描述

滑动时间窗口

在这里插入图片描述

滑动窗口是固定窗口更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成。Flink 的滑动时间窗口（Sliding Window）是一种常用的窗口机制，适用于处理流式数据时需要在时间范围内定期计算的场景。滑动窗口会按照指定的窗口大小（window size）和滑动步长（slide interval）不断地划分数据，并对每个窗口内的数据进行聚合计算。

类型特点

窗口长度固定，可以有重叠。

  • 滑动窗口会有重叠部分，因此每个事件可能会被包含在多个窗口中。
  • 滑动窗口更适合定期计算某个时间范围内的聚合值，像是移动平均值、最近一段时间的活跃用户等场景。

关键参数

  • 窗口大小（window size）：每个窗口包含的时间范围，例如 10 秒。
  • 滑动步长（slide interval）：窗口每次滑动的时间步长，例如 5 秒。这意味着每隔 5 秒就会创建一个新的窗口，每个窗口覆盖的时间范围是 10 秒。

基于时间驱动

场景：我们可以每30秒计算一次最近一分钟用户购买的商品数

package icu.wzk;import org.apache.commons.math3.analysis.function.Sin;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;import java.text.SimpleDateFormat;
import java.util.Random;public class SlidingWindow {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999);SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = dataStreamSource.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");long timeMillis = System.currentTimeMillis();int random = new Random().nextInt(10);System.out.println("value: " + value + ", random: " + random +", timestamp: " + format.format(timeMillis));return Tuple2.of(value, random);}});KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = mapStream.keyBy(new KeySelector<Tuple2<String, Integer>, Tuple>() {@Overridepublic Tuple getKey(Tuple2<String, Integer> value) throws Exception {return Tuple1.of(value.f0);}});WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> timeWindow = keyedStream.timeWindow(Time.seconds(10), Time.seconds(5));timeWindow.apply(new MyTimeWindowFunction()).print();env.execute("SlidingWindow");}}

基于事件驱动

package icu.wzk;import org.apache.commons.math3.analysis.function.Sin;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;import java.text.SimpleDateFormat;
import java.util.Random;public class SlidingWindow {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999);SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = dataStreamSource.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");long timeMillis = System.currentTimeMillis();int random = new Random().nextInt(10);System.out.println("value: " + value + ", random: " + random +", timestamp: " + format.format(timeMillis));return Tuple2.of(value, random);}});KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = mapStream.keyBy(new KeySelector<Tuple2<String, Integer>, Tuple>() {@Overridepublic Tuple getKey(Tuple2<String, Integer> value) throws Exception {return Tuple1.of(value.f0);}});WindowedStream<Tuple2<String, Integer>, Tuple, GlobalWindow> globalWindow = keyedStream.countWindow(3, 2);globalWindow.apply(new MyCountWindowFuntion()).print();env.execute("SlidingWindow");}}

会话窗口

由一系列事件组合一个指定时间长度timeout间隙组成，类似于Web应用的Session，也就是一段时间没有接收到新数据会生成新的窗口。
Session窗口分配器通过Session活动来对元素进行分组，Session窗口跟滚动窗口和滑动窗口相比，不会有重叠和固定的开始时间和结束时间的情况。
Session窗口在一个固定的时间周期内不再收到元素，即非活动间隔产生，那么这个窗口就会关闭。
一个Session窗口通过一个Session间隔来配置，这个Session间隔定义了非活跃周期的长度，当这个非活跃周期产生，那么当前的Session将关闭并且后续的元素将被分配到新的Session窗口去。

类型特点

  • 会话窗口不重叠，没有固定的开始和结束时间
  • 于翻滚窗口和滑动窗口相反，当会话窗口在一段时间内没有接收到元素时会关闭会话窗口。
  • 后续的元素将会被分配到新的会话窗口

基于时间驱动

package icu.wzk;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;public class SessionWindow {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999);SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = dataStreamSource.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {return null;}});KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = mapStream.keyBy(new KeySelector<Tuple2<String, Integer>, Tuple>() {@Overridepublic Tuple getKey(Tuple2<String, Integer> value) throws Exception {return Tuple1.of(value.f0);}});WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> window = keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)));window.apply(new MyTimeWindowFunction()).print();env.execute("SessionWindow");}}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/web/52104.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

自定义 SpringBoot Starter

自定义 SpringBoot Starter

文章目录 一、自定义 starter1.1 创建 maven 项目1.2 创建邮件配置属性类1.3 创建模拟邮件发送服务类1.4 创建自动配置类1.5 spring.factories 相关配置1.6 打包成依赖 二、测试项目2.1 创建项目2.2 application.yml 配置2.3 测试应用 参考资料 本文源码位于 java-demos/spring…
阅读更多...
Restful风格接口开发

Restful风格接口开发

一、项目搭建 安装nestjs脚手架 // 安装nestjs脚手架 npm i nestjs/cli// 新建 nest new [名字]//选择要用的工具 npm / yarn / pnpm 文件信息&#xff1a; 【main.ts】&#xff1a; 入口文件&#xff0c;通过NestFactory&#xff08;由nestjs/core库抛出的对象&#x…
阅读更多...
微信小程序手写签名

微信小程序手写签名

微信小程序手写签名组件 该组件基于signature_pad封装&#xff0c;signature_pad本身是web端的插件&#xff0c;此处将插件代码修改为小程序端可用。 signature_pad.js /*!* Signature Pad v5.0.3 | https://github.com/szimek/signature_pad* (c) 2024 Szymon Nowak | Releas…
阅读更多...
九盾叉车U型区域警示灯，高效照明和安全警示

九盾叉车U型区域警示灯，高效照明和安全警示

叉车运作的环境比较复杂&#xff0c;在方便人们物流运输的同时也存在着很大的安全隐患&#xff0c;特别是叉车碰撞人的事故发生率很高&#xff0c;那我们该怎么在减少成本的同时又能避免碰撞事故的发生呢&#xff1f; 九盾叉车U型区域警示灯&#xff0c;仅需一盏灯安装在叉车尾…
阅读更多...
十一 面向对象技术(考点篇)试题

十一 面向对象技术(考点篇)试题

A &#xff1b;D&#xff0c;D。实际答案&#xff1a;C&#xff1b;D&#xff0c;D 考的很偏了。UML 2.0基础结构的设计目标是定义一个元语言的核心 UML 2.0 【InfrastructureLibrary】,通过对此核心的复用&#xff0c;除了可以定义一个自展的UML元模型&#xff0c;也可以 Infr…
阅读更多...
基于IP子网的VLAN典型配置举例（H3C,其他厂商同理）

基于IP子网的VLAN典型配置举例（H3C,其他厂商同理）

基于IP子网的VLAN典型配置举例 1. 组网需求 如下图所示&#xff0c;办公区的主机属于不同的网段192.168.5.0/24和192.168.50.0/24&#xff0c;Device C在收到来自办公区主机的报文时&#xff0c;根据报文的源IP地址&#xff0c;使来自不同网段主机的报文分别在指定的VLAN中传…
阅读更多...
7、Django Admin删除默认应用程序

7、Django Admin删除默认应用程序

admin文件 from django.contrib.auth.models import User, Groupadmin.site.unregister(User) admin.site.unregister(Group) 显示效果&#xff1a; 前 后
阅读更多...
基于FreeRTOS的STM32多功能手表

基于FreeRTOS的STM32多功能手表

前言 项目背景 项目演示 使用到的硬件 项目原理图 目前版本实现的功能 设计到的freertos知识 实现思路 代码讲解 初始化GPIO引脚、配置时钟 蜂鸣器初始化以及软件定时器创建 系统默认创建的defaultTaskHandle 创建七个Task&#xff0c;代表七个功能 ShowTimeTask …
阅读更多...
2024.9自然语言及语言处理设计开发工程师专项培训通知！

2024.9自然语言及语言处理设计开发工程师专项培训通知！

为进一步贯彻落实中共中央印发《关于深化人才发展体制机制改革的意见》和国务院印发《关于“十四五”数字经济发展规划》等有关工作的部署要求&#xff0c;深入实施人才强国战略和创新驱动发展战略&#xff0c;加强全国数字化人才队伍建设&#xff0c;持续推进人工智能从业人员…
阅读更多...
Epoll 用法

Epoll 用法

Epoll 监听 EPOLL_CTL_DEL EPOLL_CTL_ADD epoll_event event event.events event.data.fd
阅读更多...
双指针(3)_快慢指针_快乐数问题

双指针(3)_快慢指针_快乐数问题

个人主页&#xff1a;C忠实粉丝 欢迎 点赞&#x1f44d; 收藏✨ 留言✉ 加关注&#x1f493;本文由 C忠实粉丝 原创 双指针(3)_快慢指针_快乐数问题 收录于专栏【经典算法练习】 本专栏旨在分享学习C的一点学习笔记&#xff0c;欢迎大家在评论区交流讨论&#x1f48c; 目录 1.…
阅读更多...
Android Studio gradle下载太慢了！怎么办？（已解决）

Android Studio gradle下载太慢了！怎么办？（已解决）

Android Studio&#xff01;你到底干了什么&#xff1f;&#xff01; 不能高速下载gradle&#xff0c;我等如何进行app编程&#xff1f;&#xff01; 很简单&#xff0c;我修改gradle地址不就是了。 找到gradle-wrapper.properties文件 修改其中distributionUrl的地址。 将 ht…
阅读更多...
ES配合高德地图JS-API实现地理位置查询

ES配合高德地图JS-API实现地理位置查询

目录 实现功能点 技术选型 具体实现 Vue3整合高德地图JS API-2.0 添加商户&#xff1a;前端 添加商户&#xff1a;后端/ES 查询用户当前地理坐标 获取附近&#xff08;指定距离&#xff09;的商户 总结/测试Demo代码地址 测试概述&#xff1a;用户使用高德地图组件获取商户…
阅读更多...
SpringBoot2：学SpringBoot前的知识准备-用IDEA创建传统的webapp工程，并整合SpringMVC

SpringBoot2：学SpringBoot前的知识准备-用IDEA创建传统的webapp工程，并整合SpringMVC

1、IDEA创建工程 基于Maven模板创建的SpringMVC工程 工程创建好后&#xff0c;只有webapp目录 这里&#xff0c;我们需要手动创建java目录和resources配置文件目录 创建好后&#xff0c;配置下目录属性 最终结构 至此&#xff0c;工程就创建好了 2、配置Tomcat 参考&am…
阅读更多...
论文笔记：2023顶会SIGIR - Strategy-aware Bundle Recommender System

论文笔记：2023顶会SIGIR - Strategy-aware Bundle Recommender System

论文笔记&#xff1a;2023顶会SIGIR - Strategy-aware Bundle Recommender System
阅读更多...
【Python报错已解决】`Provisional headers are shown Learn more`

【Python报错已解决】`Provisional headers are shown Learn more`

&#x1f3ac; 鸽芷咕&#xff1a;个人主页 &#x1f525; 个人专栏: 《C干货基地》《粉丝福利》 ⛺️生活的理想&#xff0c;就是为了理想的生活! 文章目录 引言&#xff1a;一、问题描述&#xff1a;1.1 报错示例&#xff1a;1.2 报错分析&#xff1a;1.3 解决思路&#xff…
阅读更多...
145. 利用 Redis Bitmap实践： 用户签到统计

145. 利用 Redis Bitmap实践： 用户签到统计

文章目录 一、Redis Bitmap简介二、Bitmap 的主要应用三、Go使用Redis实现签到统计用户签到查询用户签到状态统计今年累计签到天数统计当月的签到情况 总结 在现代应用程序中&#xff0c;用户签到是一个常见的功能。我们通常使用 MySQL 数据库来存储用户的签到记录。然而&#…
阅读更多...
智能家居系统（基于STM32F103C8T6标准库+FreeRTOS+Qt串口开发实现）

智能家居系统（基于STM32F103C8T6标准库+FreeRTOS+Qt串口开发实现）

视频演示&#xff1a;基于STM32F103C8T6标准库FreeRTOSQt串口开发实现的智能家居项目_哔哩哔哩_bilibili 基于STM32F103C8T6标准库FreeRTOSQt串口开发实现的智能家居项目: https://pan.baidu.com/s/1f41gAfOOnlcQoKoMx3o84A?pwd6j2g 提取码: 6j2g 注&#xff1a;本项目为学习完…
阅读更多...
Windows I/O系统

Windows I/O系统

硬件存储体系 寄存器 处理器内部定义的存储体&#xff0c;它们除了存储功能&#xff0c;往往还兼有其他的能力&#xff0c;比如参与运算&#xff0c;地址解析&#xff0c;指示处理器的状态&#xff0c;等等。寄存器是由处理器内部专门的触发器电路实现的&#xff0c;处理器往…
阅读更多...
2024年高教杯国赛（B题）数学建模竞赛解题思路|完整代码论文集合

2024年高教杯国赛（B题）数学建模竞赛解题思路|完整代码论文集合

我是Tina表姐&#xff0c;毕业于中国人民大学&#xff0c;对数学建模的热爱让我在这一领域深耕多年。我的建模思路已经帮助了百余位学习者和参赛者在数学建模的道路上取得了显著的进步和成就。现在&#xff0c;我将这份宝贵的经验和知识凝练成一份全面的解题思路与代码论文集合…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023