扫二维码与项目经理沟通
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
这篇文章主要讲解了“flink中窗口的作用是什么”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“flink中窗口的作用是什么”吧！
创新互联建站致力于网站设计制作、网站设计,成都网站设计,集团网站建设等服务标准化，推过标准化降低中小企业的建站的成本，并持续提升建站的定制化服务水平进行质量交付，让企业网站从市场竞争中脱颖而出。 选择创新互联建站，就选择了安全、稳定、美观的网站建设服务！
窗口计算是流式计算中常用的数据计算方式之一，通过按照固定时间或长度将数据流切分成不同的窗口，再对数据进行相应的聚合操作，得到一定时间范围内的统计结果，例如统计最近5分钟内某网站的点击数，此时，点击数据在不断产生，通过5分钟窗口将数据限定在固定时间范围内，就可以对该范围内的有界数据执行聚合，得到最近5分钟的网站点击数。
代码接口规则
stream.keyBy(...) //keyed类型数据集 .window(...) //指定窗口分配器类型 [.trigger(...)] //指定触发器类型(可选) [.evictor(...)] //指定evictor(可选) [.allowedLateness(...)] //指定是否延迟处理数据(可选) [.sideOutputLateData(...)] //指定Output Lag(可选) .reduce/aggregate/fold/apply() //指定窗口计算函数 [.getSideOutput(...)] //根据Tag输出数据(可选)
算子
Windows Assigner：指定窗口类型，定义如何将数据流分配到一个或多个窗口
Windows Trigger：指定窗口触发的时机，定义窗口满足什么样的条件触发计算；
Evictor：用于数据剔除
Lateness：标记是否处理迟到数据，当迟到数据到达窗口中是否触发计算。
Output Tag：标记输出标签，然后通过getSideOutput将窗口中的数据根据标签输出。
Windows Function：定义窗口上数据处理的逻辑，例如对数据进行sum操作。
在运用窗口计算时，Flink根据上游数据集是否为KeyedStream类型（将数据集按Key分区），对应的Window Assigner会不同，
上游数据集为KeyedStream类型，则调用DataStream API的Windwo()方法指定Windows Assigner，数据将根据Key在不同的Task实例中并行分别计算，最后得出针对每个Key统计的结果。
如果是Non-Keyed类型，则调用WindowsAll()方法来指定Windows Assigner，所有数据都被窗口算子路由到一个Task中计算，并得到结果。
建议数据进行KeyedStream处理，这样启动并行计算，加速效率。
flink支持两种类型的窗口，一种基于时间，窗口大小由开始和结束时间戳约束，一种基于数量，根据固定数量定义窗口大小。
根据Windows Assigner数据分配方式的不同将Windows分为4大类：滚动窗口（Tumbling Windows）、滑动窗口（Sliding Windows）、会话窗口（Session Windows）和全局窗口（Global Windows）
滚动窗口根据固定时间或大小切分，且窗口与窗口间元素互不重叠，适合于固定时间大小和周期统计某一指标的窗口计算。
DataStream API提供了基于Event Time和Process Time两种时间类型的Tumbling窗口，对应的Assigner分别为TumblingEventTimeWindows和TumblingProcessTimeWindows，窗口大小童工of()指定，时间单位分别为Time.milliseconds(x)、Time.seconds(x)或Time.minutes(x)，也可以是不同时间单位的组合。
如下实例，窗口时间按10S进行切分，窗口的时间是[1:00:00.000-1:00:09.999] 到[1:00:10.000-1:00:19.999]的等固定时间范围。
val inputStream:DataStream[T]= ... //定义Event Time Tumbling Windows val tumblingEventTimeWindows=inputStream.keyBy(_.id) //通过使用TumblingEventTimeWindows定义Event Time滚动窗口 .window(TumblingEventTimeWindows.of(Time.seconds(10))) .process(...) //定义窗口函数 //定义Process Time Tumbling Windows val tumblingProcessingTimeWindows = inputStream.keyBy(_.id) //通过TumblingProcessTimeWindows定义Evnet Time滚动窗口 .window(TumblingProcessTimeWindows.of(Times.seconds(10))) .process(...) //定义窗口函数
滑动窗口是一种常见的窗口类型，特点是在滚动窗口基础上增加了窗口滑动时间(Slide Time)，且允许窗口数据发生重叠。这种窗口不像滚动窗口按照Windows Size向前移动，而是根据设定的Slide Time向前滑动。窗口之间的数据重叠大小根据Windows Size和Slide time决定，当Slide Time小于Windows Size便会发生窗口重叠，Slide Size大于WindowsSize会出现窗口不连续，数据可能不会再任何一个窗口内计算。
DataStream API针对Sliding Windows根据不同时间类型Assigner，包括基于Event Time的SlidingEventTimeWindows和基于Process Time的SlidingProcessingTimeWindows。
实例如下，指定Windows Size为1h，Slide Time为10m。
val inputStream:DataStream[T]= ... //定义Event Time Sliding Windows val slidingEventTimeWindows=inputStream.keyBy(_.id) //通过使用SlidingEventTimeWindows定义Event Time滚动窗口 .window(SlidingEventTimeWindows.of(Time.hours(1),Time.minutes(10))) .process(...) //定义窗口函数 //定义Process Time Sliding Windows val slidingProcessTimeWindows = inputStream.keyBy(_.id) //通过SlidingProcessTimeWindows定义Evnet Time滚动窗口 .window(SlidingProcessTimeWindows.of(Time.hours(1),Time.minutes(10))) .process(...) //定义窗口函数
将某个时间段内活跃较高的数据聚合为一个窗口进行计算，窗口的触发条件为Session Gap，指规定时间内没有数据活跃接入，则任务窗口结束，触发窗口计算。
注意：如果数据一直不间断，会导致窗口始终不触发。
与滑动、滚动窗口不同，Session Windows不需要定义Windows Size和Slide Time，只需要定义session gap，规定不活跃数据的时间上线即可。
Session Windows比较适合非连续型数据处理或周期性产生数据的场景。DataStream API中可以创建基于Event Time和Process Time的Session Windows，对应的有Assigner分别为EventTimeSessionWindow和ProcessTimerSessionWindows。
实例代码如下：
val inputStream:DataStream[T]= ... //定义Event Time Session Windows val eventTimeSessionWindows=inputStream.keyBy(_.id) //通过使用EventTimeSessionWindows定义Event Time滚动窗口 .window(EventTimeSessionWindows.withGap(Time.milliseconds(10))) .process(...) //定义窗口函数 //定义Process Time Session Windows val processTimeSessionWindows = inputStream.keyBy(_.id) //通过ProcessTimeSessionWindows定义Evnet Time滚动窗口 .window(ProcessTimeSessionWindows.withGap(Time.milliseconds(10))) .process(...) //定义窗口函数
flink支持动态调整的Session Gap，需要实现SessionWindowTimeGapExtractor接口，并复写extract方法，完成Session Gap的抽取，然后将创建好的Session Gap抽取器传入ProcessiongTimeSessionWindows.withDynamicGap()方法即可。
val inputStream:DataStream[T]= ...
//定义Event Time Session Windows
val eventTimeSessionWindows=inputStream.keyBy(_.id)
//通过使用EventTimeSessionWindows定义Event Time滚动窗口
.window(EventTimeSessionWindows.withDynamicGap(
//实例化SessionWindowTimeGapExtractor接口
new SessionWindowTimeGapExtractor[String]{
override def extract(element:String):Long={
//动态指定并返回Session Gap
}
}
))
.process(...) //定义窗口函数
//定义Process Time Session Windows
val processTimeSessionWindows = inputStream.keyBy(_.id)
//通过ProcessTimeSessionWindows定义Evnet Time滚动窗口
.window(ProcessTimeSessionWindows.withDynamicGap(
//实例化SessionWindowTimeGapExtractor接口
new SessionWindowTimeGapExtractor[String]{
override def extract(element:String):Long={
//动态指定并返回Session Gap
}
}
))
.process(...) //定义窗口函数全局会话窗口将所有相同的key数据分配到单个窗口中计算，窗口没有起始和结束时间，窗口需要借助Triger触发计算，如果不指定，则不会触发计算。
使用全局窗口要非常谨慎，必须明确自己在整个窗口中统计出的结果是什么，并指定对应的触发器，同时指定相应的数据清理机制，否则数据将一直留在内存中。
val inputStream:DataStream[T]= ... val globalWindows = inputStream.keyBy(_.id) .window(GlobalWindows.create()) //通过GlobalWindows定义Global Windows .process()
flink定义的四种窗口，容易和时间窗口和事件窗口混淆，他们是不同维度的的窗口定义，需要特别注意下。
越长大越孤单，珍惜好身边人。
感谢各位的阅读，以上就是“flink中窗口的作用是什么”的内容了，经过本文的学习后，相信大家对flink中窗口的作用是什么这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是创新互联，小编将为大家推送更多相关知识点的文章，欢迎关注！
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
微信二维码
移动版官网