flink实战—时间窗口(Time Windows)的原理和使用

最新推荐文章于 2025-03-19 21:57:12 发布

一铭

最新推荐文章于 2025-03-19 21:57:12 发布

阅读量3.3w

点赞数 5

分类专栏：实战-flink 实战-flink实战文章标签： flink时间窗口 flink原理滑动窗口翻滚窗口 flink实战

本文链接：https://blog.csdn.net/zg_hover/article/details/87592060

版权

实战-flink 同时被 2 个专栏收录

19 篇文章

订阅专栏

实战-flink实战

19 篇文章

订阅专栏

概述

flink中支持多种窗口，包括：时间窗口，session窗口，count窗口等，本文简单介绍这些窗口的原理，并通过例子说明如何使用这些窗口。

时间窗口(Time Windows)

最简单常用的窗口形式是基于时间的窗口，flink支持两种时间窗口：

一种是翻滚时间窗口(tumbling time window)
一种是滑动时间窗口(sliding time window)

翻滚时间窗口(tumbling time window)

翻滚时间窗口的窗口是固定的，比如设定一个1分钟的时间窗口，该时间窗口将只计算当前1分钟内的数据，而不会管前1分钟或后1分钟的数据。

如上图所示，编写了一个1分钟的翻滚窗口，用来收集最后一分钟的值，并在1分钟结束时输出它们的总和。

从上图可见，该窗口只会计算从当前计时开始的1分钟内的数据，当1分钟完成时输出结果。然后，从完成这一刻起开始计算1分钟内的数据，依次类推。

一个翻滚窗口的定义如下：

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val text = env.socketTextStream("localhost", 9999)

    val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
      .map { (_, 1) }
      .keyBy(0)
      .timeWindow(Time.seconds(5))  //定义一个5秒的翻滚窗口
      .sum(1)

滑动时间窗口(sliding time window)

滑动窗口，顾名思义，该时间窗口是滑动的。所以，从概念上讲，这里有两个方面的概念需要理解：

窗口：需要定义窗口的大小
滑动：需要定义在窗口中滑动的大小，但理论上讲滑动的大小不能超过窗口大小

下面我们来看一个图例：

如上图所示，定义了一个1分钟的滑动窗口。在第一个滑动窗口中，将值9,6,8和4相加，得到结果27。接着，窗口滑动半分钟（例如，在我们的示例中为2个值），此时窗口中的值为8,4和7,3，产生结果22，以此类推。

可以在Flink中定义1分钟的滑动窗口，每30秒滑动一次，定义如下：

stream.timeWindow(Time.minutes(1), Time.seconds(30))

定义中，1分钟为窗口时间，30秒为滑动时间。

滑动窗口使用例子

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object WindowWordCount {
  def main(args: Array[String]) {

    println("start word count")
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val text = env.socketTextStream("localhost", 9999)

    val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
      .map { (_, 1) }
      .keyBy(0)
      .timeWindow(Time.seconds(5), Time.seconds(3))  // 定义了一个滑动窗口，窗口大小为5秒，每3秒滑动一次
      .sum(1)

    counts.print()
    println("end word count")

    env.execute("Window Stream WordCount")
    println("exit now!")
  }
}

可以在另一个终端通过命令nc -lk 9999输入一些数据，查看一下效果，并理解一下。

this is a test, time windows.

为了，省去编译和打包的麻烦，可以直接在flink的scala-shell中，修改一下上面的程序，把evn改成senv。

    val text = senv.socketTextStream("localhost", 9999)

    val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } } .map { (_, 1) } .keyBy(0) .timeWindow(Time.seconds(5), Time.seconds(3)) .sum(1)
    counts.print()
    senv.execute("Window Stream WordCount")