Flink实操 : Window操作

r囧r小猫 2022-11-05 00:55 267阅读 0赞

### . ###

*  一 .前言
 *  二 .实现
 *   *  2.1. 窗口分类
     *  2.2. 时间窗口
     *   *  2.2.1. tumbling-time-window (翻滚窗口-无重叠数据)
         *  2.2.2.sliding-time-window (滑动窗口-有重叠数据)
         *  2.2.3. 小结
     *  2.3. Count-Window
     *   *  2.3.1. tumbling-count-window (无重叠数据)
         *  2.3.2. sliding-count-window (有重叠数据)
     *  2.4. Window Session
     *  2.4. Window Global
     *  2.6. Window apply

# 一 .前言 #

Flink 认为 Batch 是 Streaming 的一个特例，所以 Flink 底层引擎是一个流式引擎，在上面实现了流处理和批处理。  
而窗口（window）就是从 Streaming 到 Batch 的一个桥梁。  
Flink 提供了非常完善的窗口机制。

*  时间类型

1.  事件时间(Event Time) : 事件发生的时间,一旦确定之后再也不会改变.
2.  处理时间(Processing Time) : 消息被计算引擎处理的时间. (以各个计算节点的本地时间为准)
3.  摄取时间(IngestionTime) : 事件进入流处理系统的时间. (使用其被读取的那一刻时间作为摄取时间.)  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

在流处理应用中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理。  
当然我们可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，  
例如：在过去的1分钟内有多少用户点击了我们的网页。在这种情况下，我们必须定义一个窗口，用来收集最近一分钟内的数据，并对这个窗口内的数据进行计算。

如果在数据流上，截取固定大小的一部分，这部分是可以进行统计的。

截取方式主要有三种:

> 1.  根据`时间`进行截取(time-driven-window)，比如每1分钟统计一次或每10分钟统计一次。
> 2.  根据`消息数量`进行截取(data-driven-window)，比如每5个数据统计一次或每50个数据统计一次。
> 3.  根据`Session`来进行截取, 会话超时后产生一个新的窗口.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

# 二 .实现 #

## 2.1. 窗口分类 ##

Flink提供了三类默认窗口: **时间窗口(Time Window)** 、**计数窗口(Count Window)** 、 和 **会话窗口(Session Window)** 三种.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

*  **时间窗口(Time Window)**
    
     *  滚动时间器窗口 ( Tumble Time Window ) : 按照时间进行窗口划分,每次窗口的 滑动距离等于窗口的长度. (无重复数据)
     *  滑动时间器窗口 ( Sliding Time Window ) : 按照时间来进行窗口划分,根据指定的时间步长进行滑动. (数据可能丢失或者重复计算)
 *  **计数窗口(Count Window)**
    
     *  滚动计数器窗口 ( Tumble Count Window ) : 累积固定个数的元素就视为一个窗口,该类型无法像时间窗口一样事先切分好. (按照跟窗口步长一样的长度滑动,无重复数据)
     *  滑动计数器窗口 ( Sliding Count Window ) : 累积固定个数的元素就视为一个窗口,每超过一定个数的原则个数,产生一个新的窗口. (按照指定的个数步长滑动,数据可能丢失或有重复数据)
 *  **会话窗口(Session Window)**
    
     *  会话窗口 ( Session Window ) : 当超过一段时间,该窗口没有收到新的数据元素,则视为该窗口结束, 所以无法事先确认窗口的长度,元素个数. 窗口之间也不会相互重叠.

## 2.2. 时间窗口 ##

### 2.2.1. tumbling-time-window (翻滚窗口-无重叠数据) ###

**滚动时间器窗口 ( Tumble Time Window )** : 按照时间进行窗口划分,每次窗口的 滑动距离等于窗口的长度. (无重复数据)

按照`时间`来进行窗口划分,每次窗口的`滑动距离`等于窗口的长度,这样数据不会重复计算.

代码如下:

import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    
    
    // 要先打开 服务端,否则报错 :: nc -lk 9999
    // 滚动时间计数器窗口
    object StreamingTumblingTimeWindow { 
      def main(args: Array[String]): Unit = { 
        // 1. 创建流处理环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        // 2.定义数据流来源
        val textStream = env.socketTextStream("localhost",9999)
        // 3.处理数据
        val data = textStream.flatMap(x=> x.split(" ")).filter( x=>{  null!=x && x.trim.length != 0}).map(x=>{ (x,1)})
    
    
        //4. 根据key做聚合操作..
       val keyByData = data.keyBy(_._1)
    
        //5. 执行统计操作，每5秒钟统计一下各单词的数量
        val result = keyByData.window(TumblingProcessingTimeWindows.of(Time.seconds(5))).sum(1)
    
        //6. 显示统计结果
        result.print()
        //7. 触发流计算
        env.execute()
      }
    }

### 2.2.2.sliding-time-window (滑动窗口-有重叠数据) ###

\*\*滑动时间器窗口 ( Sliding Time Window ) \*\* : 按照时间来进行窗口划分,根据指定的时间步长进行滑动. (数据可能丢失或者重复计算)

按照时间来进行窗口划分,每次窗口的滑动距离小于窗口的长度,这样数据就会有一部分重复计算.

代码如下:

import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    
    
    // 要先打开 服务端,否则报错 :: nc -lk 9999
    // 滑动时间计数器窗口
    object StreamingSlidingTimeWindow { 
      def main(args: Array[String]): Unit = { 
        // 1. 创建流处理环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        // 2.定义数据流来源
        val textStream = env.socketTextStream("localhost",9999)
        // 3.处理数据
        val data = textStream.flatMap(x=> x.split(" ")).filter( x=>{  null!=x && x.trim.length != 0}).map(x=>{ (x,1)})
    
        //4. 根据key做聚合操作..
        val keyByData = data.keyBy(_._1)
    
        //5. 执行统计操作，步长为2, 每5秒钟统计一下各单词的数量
        // TumblingProcessingTimeWindows
        val result= keyByData.window(TumblingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(2))).sum(1)
        //6. 显示统计结果
        result.print()
        //7. 触发流计算
        env.execute()
      }
    }

### 2.2.3. 小结 ###

> 1.  如果窗口计算时间 > 窗口时间，会出现数据丢失
> 2.  如果窗口计算时间 < 窗口时间，会出现数据重复计算
> 3.  如果窗口计算时间 = 窗口时间，数据不会被重复计算

窗口计算时间 > 窗口时间

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-CxQIiGos-1614760997462)(../../BigDataVideo/Flink/%E5%9F%BA%E7%A1%80%E8%AE%B2%E4%B9%89/assets/1554437419868.png)\]][img-CxQIiGos-1614760997462_.._.._BigDataVideo_Flink_E5_9F_BA_E7_A1_80_E8_AE_B2_E4_B9_89_assets_1554437419868.png]

窗口计算时间 < 窗口时间

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mV8RvAhD-1614760997465)(assets/1554437468041.png)\]][img-mV8RvAhD-1614760997465_assets_1554437468041.png]

窗口计算时间 = 窗口时间

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HuApXzsX-1614760997472)(../../BigDataVideo/Flink/%E5%9F%BA%E7%A1%80%E8%AE%B2%E4%B9%89/assets/1554437536600.png)\]][img-HuApXzsX-1614760997472_.._.._BigDataVideo_Flink_E5_9F_BA_E7_A1_80_E8_AE_B2_E4_B9_89_assets_1554437536600.png]

## 2.3. Count-Window ##

### 2.3.1. tumbling-count-window (无重叠数据) ###

**滚动计数器窗口 ( Tumble Count Window )** : 累积固定个数的元素就视为一个窗口,该类型无法像时间窗口一样事先切分好. (按照跟窗口步长一样的长度滑动,无重复数据)

> 按照单词个数统计

代码如下:

import org.apache.flink.api.scala.createTypeInformation
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, StreamExecutionEnvironment}
    
    // 滚动
    object StreamingCountTumblingWindow { 
      def main(args: Array[String]): Unit = { 
        // 1. 创建流处理环境
        val env : StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        // 2.定义数据流来源
        val scoketText : DataStream[String] = env.socketTextStream("localhost",9999)
        // 3.处理数据
        val data = scoketText.flatMap(_.split(" ")).filter( x=>{ x != null && x.trim.length >0 }).map(x=>(x,1))
        //4. 根据key做聚合操作..
        val keyedStream : KeyedStream[(String,Int),String] = data.keyBy(_._1)
        //5. 相同的key出现三次才做一次sum聚合
        val res = keyedStream.countWindow(3).sum(1)
        //6. 显示统计结果
        res.print()
        //7. 触发流计算
        env.execute()
    
      }
    }

### 2.3.2. sliding-count-window (有重叠数据) ###

滑动计数器窗口 ( Sliding Count Window ) : 累积固定个数的元素就视为一个窗口,每超过一定个数的原则个数,产生一个新的窗口. (按照指定的个数步长滑动,数据可能丢失或有重复数据)

窗口长度和滑动窗口的操作：窗口长度是5，滑动长度是3

import org.apache.flink.api.scala.createTypeInformation
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, StreamExecutionEnvironment}
    
    
    object StreamingCountSlidingWindow { 
      def main(args: Array[String]): Unit = { 
        // 1. 创建流处理环境
        val env : StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    
        // 2.定义数据流来源
        val scoketText : DataStream[String] = env.socketTextStream("localhost",9999)
        // 3.处理数据
        val data = scoketText.flatMap(_.split(" ")).filter( x=>{ x != null && x.trim.length >0 }).map(x=>(x,1))
        //4. 根据key做聚合操作..
        val keyedStream : KeyedStream[(String,Int),String] = data.keyBy(_._1)
        //5. 相同的key出现三次才做一次sum聚合 , 步长为2, 窗口为3 .
        val res = keyedStream.countWindow(3,2).sum(1).setParallelism(1)
        //6. 显示统计结果
        res.print()
        //7. 触发流计算
        env.execute()
    
      }
    }

## 2.4. Window Session ##

**会话窗口 ( Session Window )** : 当超过一段时间,该窗口没有收到新的数据元素,则视为该窗口结束, 所以无法事先确认窗口的长度,元素个数. 窗口之间也不会相互重叠.

import org.apache.flink.api.scala.createTypeInformation
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, StreamExecutionEnvironment}
    import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    
    
    object StreamingStaticSessionWindow { 
      def main(args: Array[String]): Unit = { 
        // 1. 创建流处理环境
        val env : StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    
        // 2.定义数据流来源
        val scoketText : DataStream[String] = env.socketTextStream("localhost",9999)
        // 3.处理数据
        val data = scoketText.flatMap(_.split(" ")).filter( x=>{ x != null && x.trim.length >0 }).map(x=>(x,1))
        //4. 根据key做聚合操作..
        val keyedStream : KeyedStream[(String,Int),String] = data.keyBy(_._1)
        //5. 相同的key出现三次才做一次sum聚合 , 步长为2, 窗口为3 .
        val res = keyedStream.window(EventTimeSessionWindows.withGap(Time.milliseconds(100))).sum(1)
        //6. 显示统计结果
        res.print()
        //7. 触发流计算
        env.execute()
    
      }
    }

## 2.4. Window Global ##

全局窗口赋值器将具有相同键的所有元素分配给同一个全局窗口。此窗口方案仅在还指定自定义触发器时有用。否则，不会执行任何计算，因为全局窗口没有可以处理聚合元素的自然结束。

val input: DataStream[T] = ...
    
    input
        .keyBy(<key selector>)
        .window(GlobalWindows.create())
        .<windowed transformation>(<window function>)

## 2.6. Window apply ##

apply方法可以进行一些自定义处理，通过匿名内部类的方法来实现。当有一些复杂计算时使用。

**用法**

1.  实现`WindowFunction`类
2.  指定该类的泛型为`[输入数据类型, 输出数据类型, keyBy中使用分组字段的类型, 窗口类型]`

**示例**

使用apply方法来实现单词统计

**步骤**

1.  获取流处理运行环境
2.  构建socket流数据源，并指定IP地址和端口号
3.  对接收到的数据转换成单词元组
4.  使用`keyBy`进行分流（分组）
5.  使用`timeWinodw`指定窗口的长度（每3秒计算一次）
6.  实现一个WindowFunction匿名内部类
    
     *  在apply方法中实现聚合计算
     *  使用Collector.collect收集数据
7.  打印输出
8.  启动执行
9.  在Linux中，使用`nc -lk 端口号`监听端口，并发送单词

**参考代码**

import org.apache.flink.streaming.api.scala.{ DataStream, StreamExecutionEnvironment}
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.scala.function.WindowFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    object WindowApply { 
    
      def main(args: Array[String]): Unit = { 
        // 1. 获取流处理运行环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    
        // 2. 构建socket流数据源，并指定IP地址和端口号
        val socketDataStream = env.socketTextStream("localhost", 9999)
    
        // 3. 对接收到的数据转换成单词元组
        val wordcountDataStream: DataStream[(String, Int)] = socketDataStream.flatMap { 
          text =>
            text.split(" ").map(_ -> 1)
        }
    
        // 4. 使用`keyBy`进行分流（分组）
        val groupedDataStream = wordcountDataStream.keyBy(_._1)
    
        // 5. 使用`timeWinodw`指定窗口的长度（每3秒计算一次）
        val windowedDataStream = groupedDataStream.timeWindow(Time.seconds(3))
    
        // 6. 实现一个WindowFunction匿名内部类
        val resultDataStream: DataStream[(String, Int)] = windowedDataStream.apply(new WindowFunction[(String, Int), (String, Int), String, TimeWindow] { 
    
          // - 在apply方法中实现聚合计算
          override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = { 
            val resultWordCount: (String, Int) = input.reduce { 
              (wc1, wc2) =>
                (wc1._1, wc1._2 + wc2._2)
            }
            // - 使用Collector.collect收集数据
            out.collect(resultWordCount)
          }
        })
    
        // 7. 打印输出
        resultDataStream.print()
    
        // 8. 启动执行
        env.execute("App")
    
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221023/2aac362c85b641978881f3af0ee36972.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221023/f397c2bf5caf49eabb89c8091959cd17.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20221023/1d72c218eaff48aea3ad825a78f915fc.png
[img-CxQIiGos-1614760997462_.._.._BigDataVideo_Flink_E5_9F_BA_E7_A1_80_E8_AE_B2_E4_B9_89_assets_1554437419868.png]: /images/20221023/806f6a3b7daf43bf8633daf10607de12.png
[img-mV8RvAhD-1614760997465_assets_1554437468041.png]: /images/20221023/f25e226cd9194845b41355dbfe7de9a9.png
[img-HuApXzsX-1614760997472_.._.._BigDataVideo_Flink_E5_9F_BA_E7_A1_80_E8_AE_B2_E4_B9_89_assets_1554437536600.png]: /images/20221023/037638a1925348fa92925860f87d9c31.png