spark streaming 滑动窗口

spark streaming 滑动窗口滑动窗口DStream.window(windowlength,slidinginterval) batchinterval:批处理时间间隔,sparkstreaming将消息源(Kafka)的数据,以流的方式按批处理时间间隔切片,一个批处理间隔时间对应1个切片对应生成的1个RDDwindowlength:窗口时间长度,每个批处理间隔将会实际处理的RDD个数(1…n…

大家好,又见面了,我是你们的朋友全栈君。

滑动窗口

DStream.window(window length,sliding interval)

 

batch interval:批处理时间间隔,spark streaming将消息源(Kafka)的数据,以流的方式按批处理时间间隔切片,一个批处理间隔时间对应1个切片对应生成的1个RDD

window length :窗口时间长度,每个批处理间隔将会实际处理的RDD个数(1…n)。是批处理间隔的N(N>=1)倍。

sliding interval:滑动窗口时间长度,窗口操作执行的时间间隔。如果设置为=batch interval,则每个批处理时间间隔都会执行一次窗口操作,如果设置为=N*processingInterval(N>1,N为Int),则每N个批处理时间间隔会执行一次窗口操作。

 

假设spark streaming 从kafka的largest 偏移量处开始消费

对于一个新的消费者:

每隔一次batch interval,会更新一次offset(拉取的数据为该batch interval内进入kafka的实时数据)

每隔一次sliding interval,会进行生成windowed DStream 操作,并执行逻辑,最后更新一次offset。其中生成的 windowed DStream的数据源为当前最后  window length对应的N个RDD的和(N>=sliding interval,且N=n*batch interval)。

     

对于一个旧的消费者:

每隔一次batch interval,会更新一次offset(拉取的数据为该batch interval内进入kafka的实时数据+之前保存的offset位置到当前位置的历史数据)

每隔一次sliding interval,会进行生成windowed DStream 操作,并执行逻辑,最后更新一次offset。其中生成的 windowed DStream的数据源为当前最后  window length包含的N个RDD的和(N>=sliding interval,且N=n*batch interval)。

                  

1.如果,window length=3Min,sliding interval=1Min,batch interval=1Min,假设spark streaming 从kafka的largest 偏移量处开始消费。

上述语义为:每隔1分钟,将当前最后3分钟的数据生成一个windowed DStream(如果有多个RDD,则合并他们)

spark streaming 滑动窗口

在第一个分钟里,会从kafka里面拉取新进入kafka里的第一分钟的数据并封装为RDD存储到内存,并拉取当前最后1分钟的数据生成一个windowed DStream执行print等action操作(为什么是当前最后1分钟?因为当前只有1分钟的数据)

两分钟过去后,会从kafka里面拉取新进入kafka里的第2分钟的数据并封装为RDD存储到内存,并拉取当前最后2分钟的数据生成一个windowed DStream执行print等action操作

3分钟过去后,会从kafka里面拉取新进入kafka里的第3分钟的数据并封装为RDD存储到内存,并拉取当前最后3分钟的数据生成一个windowed DStream执行print等action操作

4分钟过去后,会从kafka里面拉取新进入kafka里的第4分钟的数据并封装为RDD存储到内存,并拉取当前最后3分钟的数据生成一个windowed DStream执行print等action操作

5分钟过去后,会从kafka里面拉取新进入kafka里的第5分钟的数据并封装为RDD存储到内存,并拉取当前最后3分钟的数据生成一个windowed DStream执行print等action操作

….

2. 如果,window length=3Min,sliding interval=2Min,batch interval=1Min,假设spark streaming 从kafka的largest 偏移量处开始消费。

上述语义为:每隔2分钟,将当前最后3分钟的数据生成一个windowed DStream(如果有多个RDD,则合并他们)

spark streaming 滑动窗口

 

在14个batch interval 里会执行7次窗口数据处理,除了第一个窗口长度为2个batch interval以外,其他都为3个batch interval。

在第一个分钟里,会从kafka里面拉取新进入kafka里的第一分钟的数据并封装为RDD存储到内存

两分钟过去后,会从kafka里面拉取新进入kafka里的第2分钟的数据并封装为RDD存储到内存,执行print等action操作,这次会执行2个RDD里面的数据。

3分钟过去后,会从kafka里面拉取新进入kafka里的第3分钟的数据并封装为RDD存储到内存,不会执行print等action操作

4分钟过去后,会从kafka里面拉取新进入kafka里的第4分钟的数据并封装为RDD存储到内存,并拉取当前最后3分钟的数据生成一个windowed DStream执行print等action操作

5分钟过去后,会从kafka里面拉取新进入kafka里的第5分钟的数据并封装为RDD存储到内存,不会执行print等action操作

6分钟过去后,会从kafka里面拉取新进入kafka里的第6分钟的数据并封装为RDD存储到内存,并拉取当前最后3分钟的数据生成一个windowed DStream执行print等action操作

7分钟过去后,会从kafka里面拉取新进入kafka里的第7分钟的数据并封装为RDD存储到内存,不会执行print等action操作

…..

在实际应用中:window length – sliding interval >=应用中给定的需要统计的累计最大时长,这样才不会因为当前窗口遗漏某些特殊时间段的数据。

如有这样一个逻辑:要求判断连续30分钟的数据满足条件A,则得出结果B

spark streaming 滑动窗口

如果,让window length=30,sliding interval=10,batch interval=10,即window length – sliding interval < 30分钟

那么 从第5分钟开始连续,直到35分钟时结束连续的这段数据,将不能正常得到结果B

如果,让window length=40,sliding interval=10,batch interval=10,即window length – sliding interval =30分钟

那么 从第5分钟开始连续,直到35分钟时结束连续的这段数据,将可以正常得到结果B

 

觉得还行的话,右上角点个赞哟。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/149657.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(1)
blank

相关推荐

  • 编程实现盗版QQ「建议收藏」

    编程实现盗版QQ「建议收藏」源码下载(附件1 )一、思路分析  (一) 数据包的角度 (二) 钩子角度 1. 钩子简介 2. 程序流程 ;二、 实现过程   (一) 定义宏   (二) 枚举进程找到QQ.exe   (三) 枚举所有窗口,找属于QQ.exe的窗口   (四) 获取用户名、密码、登陆按钮的句柄   (五) 创建钩子用的DLL   (六) 安装钩子  一直以来我对盗QQ这种技术都比较的好奇,最近为了练

  • mvc的各个部分都有哪些技术实现_计算机植入木马程序

    mvc的各个部分都有哪些技术实现_计算机植入木马程序MVC三个基本部分:模型(Model)、视图(View)和控制器(Controller)Model:负责业务对象与数据库的映射(ORM)View:负责与用户的交互Controller:接受用户的输入调用模型和视图完成用户的请求MVTDjango框架的MTV设计模式借鉴了MVC框架的思想,三部分为:Model、Template和ViewModel(模型):负责业务对象与数据库的对象(…

  • Vue上传文件到springboot

    Vue上传文件到springboot<el-uploadclass=”upload-demo”ref=”upload”accept=”image/png,image/jpg,image/jpeg”:file-list=”fileLists”:on-preview=”handlePreview”…

    2022年10月16日
  • java 常量池和运行时常量池_常量池中的字符串是对象吗

    java 常量池和运行时常量池_常量池中的字符串是对象吗简介:这几天在看Java虚拟机方面的知识时,看到了有几种不同常量池的说法,然后我就去CSDN、博客园等上找资料,里面说的内容真是百花齐放,各自争艳,因此,我好好整理了一下,将我自认为对的理解写下来与大家共同探讨:在Java的内存分配中,总共3种常量池:1.字符串常量池(StringConstantPool):1.1:字符串常量池在Java内存区域的哪个位置?在JDK6.0及之前版本,字符串

  • C6000系列DSP的中断系统

    C6000系列DSP的中断系统C6000系列DSP的中断系统上一篇介绍了C6455的GPIO系统,最后把GPIO4配置成了中断/事件模式,本文将介绍C6455的中断系统,介绍完基本概念后,给出把GPIO4映射到INT4的代码。完成了GPIO和中断的配置,我们就可以开始着手DSP与FPGA之间的通讯了。废话不多说,下面开始介绍C6455的中断系统。————————-

  • golang 2022.01.13激活码【2022.01最新】2022.03.06

    (golang 2022.01.13激活码)最近有小伙伴私信我,问我这边有没有免费的intellijIdea的激活码,然后我将全栈君台教程分享给他了。激活成功之后他一直表示感谢,哈哈~IntelliJ2021最新激活注册码,破解教程可免费永久激活,亲测有效,下面是详细链接哦~https://javaforall.cn/100143.html…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号