十、Flink中的窗口

- 日理万妓 2024-04-19 11:32 59阅读 0赞

**目录**

窗口类型

窗口的实现方式

1、Tumbling Time Window 翻滚时间窗口

2、Sliding Time Window 滑动时间窗口

3、Tumbling Count Window 翻滚计数窗口

4、Session Window 会话窗口

--------------------

# 窗口类型 #

*  1. flink支持两种划分窗口的方式（time和count） 如果根据时间划分窗口，那么它就是一个time-window 如果根据数据划分窗口，那么它就是一个count-window
 *  2. flink支持窗口的两个重要属性（size和interval）

如果size=interval,那么就会形成tumbling-window(无重叠数据)

如果size>interval,那么就会形成sliding-window(有重叠数据)

如果size<interval,那么这种窗口将会丢失数据。比如每5秒钟，统计过去3秒的通过路口汽车的数据，将会漏掉2秒钟的数据。

*  3. 通过组合可以得出四种基本窗口：

\`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)

*  4. flink支持在stream上的通过key去区分多个窗口

# 窗口的实现方式 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70][]

## **1、Tumbling Time Window** 翻滚时间窗口 ##

假如我们需要统计每一分钟中用户购买的商品的总数，需要将用户的行为事件按每一分钟进行切分，这种切分被成为翻滚时间窗口（Tumbling Time Window）。翻滚窗口能将数据流切分成不重叠的窗口，每一个事件只能属于一个窗口。

object FlinkTransTimeWindow {
      def main(args: Array[String]): Unit = {
        //nc -l portNum (nc -l 9000)
        //  获取运行环境
        val env :StreamExecutionEnvironment =StreamExecutionEnvironment.getExecutionEnvironment
        //    链接socket获取输入数据
        val text =env.socketTextStream("masters",9000,',')
        //  注意：必需要加上这个隐式转换，否则下面的flatmap方法执行会报错
        import org.apache.flink.api.scala._
        //    解析数据，吧数据打平，分组，窗口计算，并且聚合求sum
        val windowCounts =text.flatMap(line=>line.split(",")
          .map(w=> WordWithCount(w,1L)) )//把单词转成（word，1）这种形式
          .keyBy("word")
          .timeWindow(Time.minutes(1))//指定窗口大小，指定间隔时间
          .sum("count")//sum或者reduce都可以
        windowCounts.print().setParallelism(1)//打印到控制台
        env.execute("socket window count scala")
    
      }
      case class WordWithCount(word:String,count:Long)
    }

## **2、Sliding Time Window** 滑动时间窗口 ##

我们可以每30秒计算一次最近一分钟用户购买的商品总数。这种窗口我们称为滑动时间窗口（Sliding Time Window）。在滑窗中，一个元素可以对应多个窗口。通过使用 DataStream API，我们可以这样实现：

object FlinkTransTimeWindow {
      def main(args: Array[String]): Unit = {
        //nc -l portNum (nc -l 9000)
        //  获取运行环境
        val env :StreamExecutionEnvironment =StreamExecutionEnvironment.getExecutionEnvironment
        //    链接socket获取输入数据
        val text =env.socketTextStream("masters",9000,',')
        //  注意：必需要加上这个隐式转换，否则下面的flatmap方法执行会报错
        import org.apache.flink.api.scala._
        //    解析数据，吧数据打平，分组，窗口计算，并且聚合求sum
        val windowCounts =text.flatMap(line=>line.split(",")
          .map(w=> WordWithCount(w,1L)) )//把单词转成（word，1）这种形式
          .keyBy("word")
          .timeWindow(Time.minutes(1),Time.seconds(30))//指定窗口大小，指定间隔时间
          .sum("count")//sum或者reduce都可以
        windowCounts.print().setParallelism(1)//打印到控制台
        env.execute("socket window count scala")
    
      }
      case class WordWithCount(word:String,count:Long)
    }

## **3、Tumbling Count Window** 翻滚计数窗口 ##

当我们想要每100个用户购买行为事件统计购买总数，那么每当窗口中填满100个元素了，就会对窗口进行计算，这种窗口我们称之为翻滚计数窗口（Tumbling Count Window），上图所示窗口大小为3个。通过使用 DataStream API，我们可以这样实现：

object FLinkTransWindow {
      def main(args: Array[String]): Unit = {
        //nc -l portNum (nc -l 9000)
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        val stream = env.socketTextStream("masters", port, ',')
        import org.apache.flink.api.scala._
    
        val streamKeyBy = stream.map(item => (item, 1L))
          .keyBy(0)
        val streamWindow = streamKeyBy
    //窗口大小5个单词,当同一个key每达到2次时会输出一次，并且当同一个key出现5次时也会执行一次，
          .countWindow(100) 
          .reduce((x, y) => (x._1, x._2 + y._2))
        streamWindow.print()
        env.execute("socket window count scala")
    
      }
      case class WordWithCount(word: String, count: Long)
    }

## **4、Session Window** 会话窗口 ##

在这种用户交互事件流中，我们首先想到的是将事件聚合到会话窗口中（一段用户持续活跃的周期），由非活跃的间隙分隔开。如上图所示，就是需要计算每个用户在活跃期间总共购买的商品数量，**如果用户30秒没有活动则视为会话断开**（假设raw data stream是单个用户的购买行为流）。Session Window 的示例代码如下：

object FlinkTransTimeWindow {
      def main(args: Array[String]): Unit = {
        //nc -l portNum (nc -l 9000)
        //  获取运行环境
        val env :StreamExecutionEnvironment =StreamExecutionEnvironment.getExecutionEnvironment
        //    链接socket获取输入数据
        val text =env.socketTextStream("masters",port,',')
        //  注意：必需要加上这个隐式转换，否则下面的flatmap方法执行会报错
        import org.apache.flink.api.scala._
        //    解析数据，吧数据打平，分组，窗口计算，并且聚合求sum
        val windowCounts =text.flatMap(line=>line.split(",")
          .map(w=> WordWithCount(w,1L)) )//把单词转成（word，1）这种形式
          .keyBy("word")
           .window(ProcessingTimeSessionWindows.withGap(Time.seconds(30)))
          .sum("count")//sum或者reduce都可以
        windowCounts.print().setParallelism(1)//打印到控制台
        env.execute("socket window count scala")
      }
      case class WordWithCount(word:String,count:Long)
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/10e345646a234d54af004290b8eaba49.png