flink笔记16 flink table windows(Group Windows/Over Windows)

Posted 2022-11-22 Aurora1217

目录

1.介绍

2.Group Windows(分组窗口)

tumbling window(滚动窗口)

Sliding Windows(滑动窗口)

Session Windows(会话窗口)

实例

3.Over Windows

无界的Over Windows

有界的Over Windows

实例

4.SQL中的Group Windows和OverWindows

Group Windows

Over Windows

---

1.介绍

时间语义，要配合窗口操作才能发挥作用。最主要的用途就是开窗口、根据时间段做计算

在Table API和SQL中，主要有两种窗口：Group Windows 和 Over Windows

• Group Windows 根据时间或行计数间隔将组行聚合成有限的组，并对每个组计算一次聚合函数
• Over Windows 窗口内聚合为每个输入行在其相邻行范围内计算一个聚合

2.Group Windows(分组窗口)

注意点：

• Group Windows 是使用 .window子句定义的，并且必须由as子句指定一个别名
• 为了按窗口对表进行分组，窗口的别名必须在 group by 子句中
• Table API 提供了一组具有特定语义的预定义Window 类，这些类会被转换为底DataStream 或 DataSet 的窗口操作
• 分组窗口分为三种：滚动窗口(tumbling window)、滑动窗口(Sliding Windows)、会话窗口(Session Windows)

tumbling window(滚动窗口)

将行分配给固定长度的不重叠的连续窗口，Tumbling windows 通过 Tumble类来定义

// Tumbling Event-time Window (.rowtime)
.window(Tumble over 10.minutes on 'rowtime as 'w)

// Tumbling Processing-time Window (.proctime)
.window(Tumble over 10.minutes on 'proctime as 'w)

// Tumbling Row-count Window (.proctime)
.window(Tumble over 10.rows on 'proctime as 'w)

Sliding Windows(滑动窗口)

滑动窗口具有固定的大小，并按指定的滑动间隔滑动。如果滑动间隔小于窗口大小，则滑动窗口重叠。因此，可以将行分配给多个窗口。

// Sliding Event-time Window (.rowtime)
.window(Slide over 10.minutes every 5.minutes on 'rowtime as 'w)

// Sliding Processing-time window (.proctime)
.window(Slide over 10.minutes every 5.minutes on 'proctime as 'w)

// Sliding Row-count window (.proctime)
.window(Slide over 10.rows every 5.rows on 'proctime as 'w)

Session Windows(会话窗口)

会话窗口没有固定的大小，但它们的界限是由不活动间隔定义的，即如果在定义的间隔期内没有事件出现，会话窗口将关闭。

// Session Event-time Window (.rowtime)
.window(Session withGap 10.minutes on 'rowtime as 'w)

// Session Processing-time Window (.proctime)
.window(Session withGap 10.minutes on 'proctime as 'w)

实例

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.table.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment, createTypeInformation
import org.apache.flink.table.api.Tumble
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.types.Row

case class sensorReading(id:String,timestamp:Long,temperature:Double)

object groupWindows 
  def main(args: Array[String]): Unit = 
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val inputStream = env.readTextFile("src/main/resources/sensor.txt")
    val dataStream = inputStream
      .map(data =>
        val arr = data.split(",")
        sensorReading(arr(0),arr(1).toLong,arr(2).toDouble)
      )
      .assignAscendingTimestamps(_.timestamp * 1000)

    val tableEnv = StreamTableEnvironment.create(env)
    val dataTable = tableEnv.fromDataStream(dataStream,'id,'temperature,'timestamp.rowtime as 'ts)

    val resultTable = dataTable
      .window( Tumble over 10.seconds on 'ts as 'w )
      .groupBy( 'id, 'w )
      .select( 'id, 'id.count, 'w.end )

    resultTable.toAppendStream[Row].print("resultTable")
    resultTable.printSchema()

    env.execute("group Window test")
  

sensor.txt

sensor_1,1619492175,36.1
sensor_2,1619492176,36.6
sensor_3,1619492177,36.5
sensor_3,1619492178,36.1
sensor_1,1619492179,36.8
sensor_3,1619492180,36.1
sensor_1,1619492200,36.5
sensor_2,1619492209,36.5

结果

3.Over Windows

注意：

• Over Windows聚合是标准的SQL(子句)已有的，并在查询的SELECT子句中定义
• Over windows 使用 .window子句定义，并在 select() 方法中通过别名来引用
• Table API 提供了 Over 类，来配置 Over 窗口的属性

Over Window 分为两大类，无界的Over Window和有界的Over Window

 无界的 over window 是使用常量指定的。也就是说，时间间隔要指定UNBOUNDED_RANGE，或者行计数间隔要指定 UNBOUNDED_ROW。

有界的 over window 是用间隔的大小指定的。

无界的Over Windows

// 无界的事件时间over window (时间字段 "rowtime")
.window(Over partitionBy 'a orderBy 'rowtime preceding UNBOUNDED_RANGE as 'w)

//无界的处理时间over window (时间字段"proctime")
.window(Over partitionBy 'a orderBy 'proctime preceding UNBOUNDED_RANGE as 'w)

// 无界的事件时间Row-count over window (时间字段 "rowtime")
.window(Over partitionBy 'a orderBy 'rowtime preceding UNBOUNDED_ROW as 'w)

//无界的处理时间Row-count over window (时间字段 "rowtime")
.window(Over partitionBy 'a orderBy 'proctime preceding UNBOUNDED_ROW as 'w)

有界的Over Windows

// 有界的事件时间over window (时间字段 "rowtime"，之前1分钟)
.window(Over partitionBy 'a orderBy 'rowtime preceding 1.minutes as 'w)

// 有界的处理时间over window (时间字段 "rowtime"，之前1分钟)
.window(Over partitionBy 'a orderBy 'proctime preceding 1.minutes as 'w)

// 有界的事件时间Row-count over window (时间字段 "rowtime"，之前10行)
.window(Over partitionBy 'a orderBy 'rowtime preceding 10.rows as 'w)

// 有界的处理时间Row-count over window (时间字段 "rowtime"，之前10行)
.window(Over partitionBy 'a orderBy 'proctime preceding 10.rows as 'w)

实例

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Over
import org.apache.flink.table.api.scala._
import org.apache.flink.types.Row

case class sensorReading(id:String,timestamp:Long,temperature:Double)

object overwindow_test 
  def main(args: Array[String]): Unit = 
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream: DataStream[String] = env.readTextFile("src/main/resources/sensor.txt")

    val dataStream = inputStream
      .map(data =>
        val arr = data.split(",")
        sensorReading(arr(0),arr(1).toLong,arr(2).toDouble)
      )

    val tableEnv = StreamTableEnvironment.create(env)

    val dataTable = tableEnv.fromDataStream(dataStream,'id,'temperature,'timestamp,'pt.proctime)
    dataTable.printSchema()

    val resultTable = dataTable
      .window( Over partitionBy 'id orderBy 'pt preceding 2.rows as 'ow )
      .select( 'id, 'pt, 'id.count over 'ow, 'temperature.avg over 'ow )

    resultTable.toAppendStream[Row].print("over")

    env.execute("overwindow test")
  

结果：

4.SQL中的Group Windows和OverWindows

Group Windows

Group Windows 在 SQL 查询的 Group BY 子句中定义

• TUMBLE(time_attr, interval) 定义一个滚动窗口

– 第一个参数是时间字段，第二个参数是窗口长度。

• HOP(time_attr, interval, interval) 定义一个滑动窗口

– 第一个参数是时间字段，第二个参数是窗口滑动步长，第三个是窗口长度

• SESSION(time_attr, interval) 定义一个会话窗口 

– 第一个参数是时间字段，第二个参数是窗口间隔（Gap）

Over Windows

• 用 Over 做窗口聚合时，所有聚合必须在同一窗口上定义，也就是说必须是相同的分区、排序和范围
• 目前仅支持在当前行范围之前的窗口
• ORDER BY 必须在单一的时间属性上指定

SELECT COUNT(amount) OVER (
  PARTITION BY user
  ORDER BY proctime
  ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)
FROM Orders

// 也可以做多个聚合
SELECT COUNT(amount) OVER w, SUM(amount) OVER w
FROM Orders
WINDOW w AS (
  PARTITION BY user
  ORDER BY proctime
  ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)

Flink Table Function ：System (Built-in) Functions

参考：Flink-分组窗口 | Over Windows | SQL 中的 Group Windows | SQL 中的 Over Windows_SmallScorpion-CSDN博客_flink over window