Flink中的窗口
窗口
窗口类型
- flink支持两种划分窗口的方式(time和count) 如果根据时间划分窗口,那么它就是一个time-window 如果根据数据划分窗口,那么它就是一个count-window
- flink支持窗口的两个重要属性(size和interval)
- 如果size=interval,那么就会形成tumbling-window(无重叠数据)
- 如果size>interval,那么就会形成sliding-window(有重叠数据)
- 如果size
- 通过组合可以得出四种基本窗口:
-
time-tumbling-window
无重叠数据的时间窗口,设置方式举例:timeWindow(Time.seconds(5)) -
time-sliding-window
有重叠数据的时间窗口,设置方式举例:timeWindow(Time.seconds(5), Time.seconds(3)) -
count-tumbling-window
无重叠数据的数量窗口,设置方式举例:countWindow(5) -
count-sliding-window
有重叠数据的数量窗口,设置方式举例:countWindow(5,3)
flink支持在stream上的通过key去区分多个窗口
窗口的实现方式
上一张经典图:
- Tumbling Time Window
假如我们需要统计每一分钟中用户购买的商品的总数,需要将用户的行为事件按每一分钟进行切分,这种切分被成为翻滚时间窗口(Tumbling Time Window)。翻滚窗口能将数据流切分成不重叠的窗口,每一个事件只能属于一个窗口。
// 用户id和购买数量 stream val counts: DataStream[(Int, Int)] = ... val tumblingCnts: DataStream[(Int, Int)] = counts // 用userId分组 .keyBy(0) // 1分钟的翻滚窗口宽度 .timeWindow(Time.minutes(1)) // 计算购买数量 .sum(1)
- Sliding Time Window
我们可以每30秒计算一次最近一分钟用户购买的商品总数。这种窗口我们称为滑动时间窗口(Sliding Time Window)。在滑窗中,一个元素可以对应多个窗口。通过使用 DataStream API,我们可以这样实现:
val slidingCnts: DataStream[(Int, Int)] = buyCnts .keyBy(0) .timeWindow(Time.minutes(1), Time.seconds(30)) .sum(1)
- Tumbling Count Window
当我们想要每100个用户购买行为事件统计购买总数,那么每当窗口中填满100个元素了,就会对窗口进行计算,这种窗口我们称之为翻滚计数窗口(Tumbling Count Window),上图所示窗口大小为3个。通过使用 DataStream API,我们可以这样实现:
// Stream of (userId, buyCnts) val buyCnts: DataStream[(Int, Int)] = ... val tumblingCnts: DataStream[(Int, Int)] = buyCnts // key stream by sensorId .keyBy(0) // tumbling count window of 100 elements size .countWindow(100) // compute the buyCnt sum .sum(1)
- Session Window
在这种用户交互事件流中,我们首先想到的是将事件聚合到会话窗口中(一段用户持续活跃的周期),由非活跃的间隙分隔开。如上图所示,就是需要计算每个用户在活跃期间总共购买的商品数量,如果用户30秒没有活动则视为会话断开(假设raw data stream是单个用户的购买行为流)。Session Window 的示例代码如下:
// Stream of (userId, buyCnts) val buyCnts: DataStream[(Int, Int)] = ... val sessionCnts: DataStream[(Int, Int)] = vehicleCnts .keyBy(0) // session window based on a 30 seconds session gap interval .window(ProcessingTimeSessionWindows.withGap(Time.seconds(30))) .sum(1)
一般而言,window 是在无限的流上定义了一个有限的元素集合。这个集合可以是基于时间的,元素个数的,时间和个数结合的,会话间隙的,或者是自定义的。Flink 的 DataStream API 提供了简洁的算子来满足常用的窗口操作,同时提供了通用的窗口机制来允许用户自己定义窗口分配逻辑。
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
Flink的分布式缓存
分布式缓存 Flink提供了一个分布式缓存,类似于hadoop,可以使用户在并行函数中很方便的读取本地文件,并把它放在taskmanager节点中,防止task重复拉取。此缓存的工作机制如下:程序注册一个文件或者目录(本地或者远程文件系统,例如hdfs或者s3),通过ExecutionEnvironment注册缓存文件并为它起一个名称。当程序执行,Flink自动将文件或者目录复制到所有taskmanager节点的本地文件系统,仅会执行一次。用户可以通过这个指定的名称查找文件或者目录,然后从taskmanager节点的本地文件系统访问它。 示例 在ExecutionEnvironment中注册一个文件: //获取运行环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); //1:注册一个文件,可以使用hdfs上的文件 也可以是本地文件进行测试 env.registerCachedFile("/Users/wangzhiwu/WorkSpace/quickstart/text","a.txt...
- 下一篇
Flink中的Time
时间 时间类型 Flink中的时间与现实世界中的时间是不一致的,在flink中被划分为事件时间,摄入时间,处理时间三种。 如果以EventTime为基准来定义时间窗口将形成EventTimeWindow,要求消息本身就应该携带EventTime 如果以IngesingtTime为基准来定义时间窗口将形成IngestingTimeWindow,以source的systemTime为准。 如果以ProcessingTime基准来定义时间窗口将形成ProcessingTimeWindow,以operator的systemTime为准。 时间详解 Processing Time Processing Time 是指事件被处理时机器的系统时间。 当流程序在 Processing Time 上运行时,所有基于时间的操作(如时间窗口)将使用当时机器的系统时间。每小时 Processing Time 窗口将包括在系统时钟指示整个小时之间到达特定操作的所有事件。 例如,如果应用程序在上午 9:15 开始运行,则第一个每小时 Processing Time 窗口将包括在上午 9:15 到上午 10:00 ...
相关文章
文章评论
共有0条评论来说两句吧...