成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

一篇文章帶你深入理解FlinkSQL中的窗口

作者： 大數據老哥
大數據
時間語義，要配合窗口操作才能發揮作用。最主要的用途，當然就是開窗口、根據時間段做計算了。下面我們就來看看 Table API 和 SQL 中，怎么利用時間字段做窗口操作。

[[360693]]

前言

時間語義，要配合窗口操作才能發揮作用。最主要的用途，當然就是開窗口、根據時間段做計算了。下面我們就來看看 Table API 和 SQL 中，怎么利用時間字段做窗口操作。在 Table API 和 SQL 中，主要有兩種窗口：Group Windows 和 Over Windows

 

一、分組窗口(Group Windows) 分組窗口(Group Windows)會根據時間或行計數間隔，將行聚合到有限的組(Group)中，并對每個組的數據執行一次聚合函數。 Table API 中的 Group Windows 都是使用.window(w:GroupWindow)子句定義的，并且必須由 as 子句指定一個別名。為了按窗口對表進行分組，窗口的別名必須在 group by 子句中，像常規的分組字段一樣引用。例子：

  1. val table = input 
  2. .window([w: GroupWindow] as 'w) 
  3. .groupBy('w, 'a) 
  4. .select('a, 'w.start, 'w.end, 'w.rowtime, 'b.count) 

Table API 提供了一組具有特定語義的預定義 Window 類，這些類會被轉換為底層DataStream 或 DataSet 的窗口操作。

Table API 支持的窗口定義，和我們熟悉的一樣，主要也是三種：滾動(Tumbling)、滑動(Sliding和 會話(Session)。

1.1 滾動窗口

滾動窗口(Tumbling windows)要用 Tumble 類來定義，另外還有三個方法：

  • over：定義窗口長度
  • on：用來分組(按時間間隔)或者排序(按行數)的時間字段
  • as：別名，必須出現在后面的 groupBy 中

實現案例

1.需求

設置滾動窗口為10秒鐘統計id出現的次數。

2.數據準備

  1. sensor_1,1547718199,35.8 
  2. sensor_6,1547718201,15.4 
  3. sensor_7,1547718202,6.7 
  4. sensor_10,1547718205,38.1 
  5. sensor_1,1547718206,32 
  6. sensor_1,1547718208,36.2 
  7. sensor_1,1547718210,29.7 
  8. sensor_1,1547718213,30.9 

3.代碼實現

  1. package windows 
  2.  
  3. import org.apache.flink.streaming.api.TimeCharacteristic 
  4. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor 
  5. import org.apache.flink.streaming.api.scala._ 
  6. import org.apache.flink.streaming.api.windowing.time.Time 
  7. import org.apache.flink.table.api.scala._ 
  8. import org.apache.flink.table.api.{EnvironmentSettings, Table, Tumble} 
  9. import org.apache.flink.types.Row 
  10.  
  11. /** 
  12.  * @Package Windows 
  13.  * @File ：FlinkSQLTumBlingTie.java 
  14.  * @author 大數據老哥 
  15.  * @date 2020/12/25 21:58 
  16.  * @version V1.0 
  17.  *          設置滾動窗口 
  18.  */ 
  19. object FlinkSQLTumBlingTie { 
  20.   def main(args: Array[String]): Unit = { 
  21.     val env = StreamExecutionEnvironment.getExecutionEnvironment 
  22.     env.setParallelism(1) 
  23.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) 
  24.  
  25.     val settings = EnvironmentSettings.newInstance() 
  26.       .useBlinkPlanner() 
  27.       .inStreamingMode() 
  28.       .build() 
  29.     val tableEnv = StreamTableEnvironment.create(env, settings) 
  30.  
  31.     // 讀取數據 
  32.     val inputPath = "./data/sensor.txt" 
  33.     val inputStream = env.readTextFile(inputPath) 
  34.     
  35.  
  36.     // 先轉換成樣例類類型（簡單轉換操作） 
  37.     val dataStream = inputStream 
  38.       .map(data => { 
  39.         val arr = data.split(",") 
  40.         SensorReading(arr(0), arr(1).toLong, arr(2).toDouble) 
  41.       }) 
  42.       .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) { 
  43.         override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L 
  44.       }) 
  45.  
  46.     val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timestamp.rowtime as 'ts) 
  47.     // 注冊表 
  48.     tableEnv.createTemporaryView("sensor", sensorTable) 
  49.     // table 實現 
  50.     val resultTable = sensorTable 
  51.       .window(Tumble over 10.seconds on 'ts as 'tw) // 每10秒統計一次，滾動時間窗口 
  52.       .groupBy('id, 'tw) 
  53.       .select('id, 'id.count, 'tw.end) 
  54.     //sql 實現 
  55.     val sqlTable = tableEnv.sqlQuery( 
  56.       """ 
  57.         |select 
  58.         |id, 
  59.         |count(id) , 
  60.         |tumble_end(ts,interval '10' second) 
  61.         |from sensor 
  62.         |group by 
  63.         |id, 
  64.         |tumble(ts,interval '10' second) 
  65.         |""".stripMargin) 
  66.  
  67.     /*** 
  68.      * .window(Tumble over 10.minutes on 'rowtime as 'w) （事件時間字段 rowtime） 
  69.      * .window(Tumble over 10.minutes on 'proctime as 'w)（處理時間字段 proctime） 
  70.      * .window(Tumble over 10.minutes on 'proctime as 'w) (類似于計數窗口，按處理時間排序，10 行一組) 
  71.      */ 
  72.     resultTable.toAppendStream[Row].print("talbe") 
  73.     sqlTable.toRetractStream[Row].print("sqlTable") 
  74.      
  75.     env.execute("FlinkSQLTumBlingTie") 
  76.   } 
  77.  
  78.   case class SensorReading(id: String, timestamp: Long, temperature: Double) 
  79.  
  80. } 

運行結果

 

1.2 滑動窗口

滑動窗口(Sliding windows)要用 Slide 類來定義，另外還有四個方法：

  • over：定義窗口長度
  • every：定義滑動步長
  • on：用來分組(按時間間隔)或者排序(按行數)的時間字段
  • as：別名，必須出現在后面的 groupBy 中

實現案例

1.需求描述

設置窗口大小為10秒鐘設置滑動距離為5秒鐘，統計id的出現的次數。

2.數據準備

  1. sensor_1,1547718199,35.8 
  2. sensor_6,1547718201,15.4 
  3. sensor_7,1547718202,6.7 
  4. sensor_10,1547718205,38.1 
  5. sensor_1,1547718206,32 
  6. sensor_1,1547718208,36.2 
  7. sensor_1,1547718210,29.7 
  8. sensor_1,1547718213,30.9 

3.實現代碼

  1. package windows 
  2.  
  3. import org.apache.flink.streaming.api.TimeCharacteristic 
  4. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor 
  5. import org.apache.flink.streaming.api.scala._ 
  6. import org.apache.flink.streaming.api.windowing.time.Time 
  7. import org.apache.flink.table.api.{EnvironmentSettings, Slide, Table} 
  8. import org.apache.flink.table.api.scala._ 
  9. import org.apache.flink.types.Row 
  10. import windows.FlinkSQLTumBlingTie.SensorReading 
  11.  
  12. /** 
  13.  * @Package windows 
  14.  * @File ：FlinkSQLSlideTime.java 
  15.  * @author 大數據老哥 
  16.  * @date 2020/12/27 22:19 
  17.  * @version V1.0 
  18.  *          滑動窗口 
  19.  */ 
  20. object FlinkSQLSlideTime { 
  21.   def main(args: Array[String]): Unit = { 
  22.     //構建運行環境 
  23.     val env = StreamExecutionEnvironment.getExecutionEnvironment 
  24.     env.setParallelism(1) // 設置分區為1 方便后面測試 
  25.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //事件時間 
  26.  
  27.     val settings = EnvironmentSettings.newInstance() 
  28.       .useBlinkPlanner() 
  29.       .inStreamingMode() 
  30.       .build() 
  31.     // 創建表env 
  32.     val tableEnv = StreamTableEnvironment.create(env, settings) 
  33.  
  34.     // 讀取數據 
  35.     val inputPath = "./data/sensor.txt" 
  36.     val inputStream = env.readTextFile(inputPath) 
  37.  
  38.     // 先轉換成樣例類類型（簡單轉換操作） 
  39.     val dataStream = inputStream 
  40.       .map(data => { 
  41.         val arr = data.split(",") 
  42.         SensorReading(arr(0), arr(1).toLong, arr(2).toDouble) 
  43.       }) 
  44.       .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) { 
  45.         override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L 
  46.       }) 
  47.  
  48.     val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timestamp.rowtime as 'ts) 
  49.     // 注冊表 
  50.     tableEnv.createTemporaryView("sensor", sensorTable) 
  51.     // table API 實現 
  52.     val tableApi = sensorTable.window(Slide over 10.seconds every 5.seconds on 'ts as 'w) 
  53.       .groupBy('w, 'id) 
  54.       .select('id, 'id.count, 'w.end) 
  55.     val tableSql = tableEnv.sqlQuery( 
  56.       """ 
  57.         |select 
  58.         |id, 
  59.         |count(id), 
  60.         |HOP_END(ts,INTERVAL '10' SECOND, INTERVAL '5' SECOND )as w 
  61.         |from sensor 
  62.         |group by 
  63.         |HOP(ts,INTERVAL '10' SECOND, INTERVAL '5' SECOND),id 
  64.         |""".stripMargin) 
  65.  
  66.     tableApi.toAppendStream[Row].print("tableApi") 
  67.     tableSql.toAppendStream[Row].print("tableSql") 
  68.     /** 
  69. .window(Slide over 10.minutes every 5.minutes on 'rowtime as 'w) （事件時間字段 rowtime） 
  70. .window(Slide over 10.minutes every 5.minutes on 'proctime as 'w) （處理時間字段 proctime）  
  71. .window(Slide over 10.rows every 5.rows on 'proctime as 'w) (類似于計數窗口，按處理時間排序，10 行一組) 
  72.    **/ 
  73.     env.execute("FlinkSQLSlideTime") 
  74.   } 
  75. } 

4.運行結果

 

1.3 會話窗口

會話窗口(Session windows)要用 Session 類來定義，另外還有三個方法：

  • withGap：會話時間間隔
  • on：用來分組(按時間間隔)或者排序(按行數)的時間字段
  • as：別名，必須出現在后面的 groupBy 中實現案例

1.需求描述

設置一個session 為10秒鐘 統計id的個數

2.準備數據

  1. sensor_1,1547718199,35.8 
  2. sensor_6,1547718201,15.4 
  3. sensor_7,1547718202,6.7 
  4. sensor_10,1547718205,38.1 
  5. sensor_1,1547718206,32 
  6. sensor_1,1547718208,36.2 
  7. sensor_1,1547718210,29.7 
  8. sensor_1,1547718213,30.9 

3.編寫代碼

  1. package windows 
  2.  
  3. import org.apache.flink.streaming.api.TimeCharacteristic 
  4. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor 
  5. import org.apache.flink.streaming.api.scala._ 
  6. import org.apache.flink.streaming.api.windowing.time.Time 
  7. import org.apache.flink.table.api.{EnvironmentSettings, Session, Table} 
  8. import org.apache.flink.table.api.scala._ 
  9. import org.apache.flink.types.Row 
  10. import windows.FlinkSQLTumBlingTie.SensorReading 
  11.  
  12. /** 
  13.  * @Package windows 
  14.  * @File ：FlinkSqlSessionTime.java 
  15.  * @author 大數據老哥 
  16.  * @date 2020/12/27 22:52 
  17.  * @version V1.0 
  18.  */ 
  19. object FlinkSqlSessionTime { 
  20.   def main(args: Array[String]): Unit = { 
  21.     //構建運行環境 
  22.     val env = StreamExecutionEnvironment.getExecutionEnvironment 
  23.     env.setParallelism(1) // 設置分區為1 方便后面測試 
  24.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //事件時間 
  25.  
  26.     val settings = EnvironmentSettings.newInstance() 
  27.       .useBlinkPlanner() 
  28.       .inStreamingMode() 
  29.       .build() 
  30.     // 創建表env 
  31.     val tableEnv = StreamTableEnvironment.create(env, settings) 
  32.  
  33.     // 讀取數據 
  34.     val inputPath = "./data/sensor.txt" 
  35.     val inputStream = env.readTextFile(inputPath) 
  36.  
  37.     // 先轉換成樣例類類型（簡單轉換操作） 
  38.     val dataStream = inputStream 
  39.       .map(data => { 
  40.         val arr = data.split(",") 
  41.         SensorReading(arr(0), arr(1).toLong, arr(2).toDouble) 
  42.       }) 
  43.       .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) { 
  44.         override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L 
  45.       }) 
  46.  
  47.     val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timestamp.rowtime as 'ts) 
  48.     // 注冊表 
  49.     tableEnv.createTemporaryView("sensor", sensorTable) 
  50.  
  51.     val tableApi = sensorTable. 
  52.       window(Session withGap 10.seconds on 'ts as 'w) 
  53.       .groupBy('id, 'w) 
  54.       .select('id, 'id.count, 'w.end) 
  55.     val tableSQL = tableEnv.sqlQuery( 
  56.       """ 
  57.         |SELECT 
  58.         |id, 
  59.         |COUNT(id), 
  60.         |SESSION_END(ts, INTERVAL '10' SECOND) AS w 
  61.         |FROM sensor 
  62.         |GROUP BY 
  63.         |id, 
  64.         |SESSION(ts, INTERVAL '10' SECOND) 
  65.         |""".stripMargin) 
  66.     tableApi.toAppendStream[Row].print("tableApi") 
  67.     tableSQL.toAppendStream[Row].print("tableSQL") 
  68.  
  69.     /** 
  70.      * .window(Session withGap 10.minutes on 'rowtime as 'w) 事件時間字段 rowtime） 
  71.      * .window(Session withGap 10.minutes on 'proctime as 'w) 處理時間字段 proctime） 
  72.      */ 
  73.     env.execute("FlinkSqlSessionTime") 
  74.   } 
  75. } 

4.運行結果

 

二、 Over Windows

Over window 聚合是標準 SQL 中已有的(Over 子句)，可以在查詢的 SELECT 子句中定義。Over window 聚合，會針對每個輸入行，計算相鄰行范圍內的聚合。Over windows使用.window(w:overwindows*)子句定義，并在 select()方法中通過別名來引用。例子：

  1. val table = input 
  2. .window([w: OverWindow] as 'w) 
  3. .select('a, 'b.sum over 'w, 'c.min over 'w) 

Table API 提供了 Over 類，來配置 Over 窗口的屬性。可以在事件時間或處理時間，以及指定為時間間隔、或行計數的范圍內，定義 Over windows。

無界的 over window 是使用常量指定的。也就是說，時間間隔要指定 UNBOUNDED_RANGE，或者行計數間隔要指定 UNBOUNDED_ROW。而有界的 over window 是用間隔的大小指定的。

2.1 無界的 over window

  1. // 無界的事件時間 over window (時間字段 "rowtime") 
  2. .window(Over partitionBy 'a orderBy 'rowtime preceding UNBOUNDED_RANGE as 'w) 
  3. //無界的處理時間 over window (時間字段"proctime") 
  4. .window(Over partitionBy 'a orderBy 'proctime preceding UNBOUNDED_RANGE as 'w) 
  5. // 無界的事件時間 Row-count over window (時間字段 "rowtime") 
  6. .window(Over partitionBy 'a orderBy 'rowtime preceding UNBOUNDED_ROW as 'w) 
  7. //無界的處理時間 Row-count over window (時間字段 "rowtime") 
  8. .window(Over partitionBy 'a orderBy 'proctime preceding UNBOUNDED_ROW as 'w) 

2.2 有界的 over window

  1. // 有界的事件時間 over window (時間字段 "rowtime"，之前 1 分鐘) 
  2. .window(Over partitionBy 'a orderBy 'rowtime preceding 1.minutes as 'w) 
  3. // 有界的處理時間 over window (時間字段 "rowtime"，之前 1 分鐘) 
  4. .window(Over partitionBy 'a orderBy 'proctime preceding 1.minutes as 'w) 
  5. // 有界的事件時間 Row-count over window (時間字段 "rowtime"，之前 10 行) 
  6. .window(Over partitionBy 'a orderBy 'rowtime preceding 10.rows as 'w) 
  7. // 有界的處理時間 Row-count over window (時間字段 "rowtime"，之前 10 行) 
  8. .window(Over partitionBy 'a orderBy 'proctime preceding 10.rows as 'w) 

2.3 代碼練習

我們可以綜合學習過的內容，用一段完整的代碼實現一個具體的需求。例如，統計每個sensor每條數據，與之前兩行數據的平均溫度。

數據準備

  1. sensor_1,1547718199,35.8 
  2. sensor_6,1547718201,15.4 
  3. sensor_7,1547718202,6.7 
  4. sensor_10,1547718205,38.1 
  5. sensor_1,1547718206,32 
  6. sensor_1,1547718208,36.2 
  7. sensor_1,1547718210,29.7 
  8. sensor_1,1547718213,30.9 

代碼分析：

  1. package windows 
  2.  
  3. import org.apache.flink.streaming.api.TimeCharacteristic 
  4. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor 
  5. import org.apache.flink.streaming.api.scala._ 
  6. import org.apache.flink.streaming.api.windowing.time.Time 
  7. import org.apache.flink.table.api.{EnvironmentSettings, Over, Tumble} 
  8. import org.apache.flink.table.api.scala._ 
  9. import org.apache.flink.types.Row 
  10.  
  11. /** 
  12. * @Package windows 
  13. * @File ：FlinkSqlTumBlingOverTime.java 
  14. * @author 大數據老哥 
  15. * @date 2020/12/28 21:45 
  16. * @version V1.0 
  17. */ 
  18. object FlinkSqlTumBlingOverTime { 
  19.  def main(args: Array[String]): Unit = { 
  20.    // 構建運行環境 
  21.    val env = StreamExecutionEnvironment.getExecutionEnvironment 
  22.    env.setParallelism(1) // 設置并行度為1方便后面進行測試 
  23.    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 設置事件時間 
  24.  
  25.    val settings = EnvironmentSettings.newInstance() 
  26.      .useBlinkPlanner() 
  27.      .inStreamingMode() 
  28.      .build() 
  29.    //構建table Env 
  30.    val tableEnv = StreamTableEnvironment.create(env, settings) 
  31.  
  32.    // 讀取數據 
  33.    val inputPath = "./data/sensor.txt" 
  34.    val inputStream = env.readTextFile(inputPath) 
  35.    // 先轉換成樣例類類型（簡單轉換操作） 
  36.    // 解析數據 封裝成樣例類 
  37.    val dataStream = inputStream 
  38.      .map(data => { 
  39.        val arr = data.split(",") 
  40.        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble) 
  41.      }) 
  42.      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) { 
  43.        override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L 
  44.      }) 
  45.    // 將數據注冊成一張臨時表 
  46.    val dataTable = tableEnv.fromDataStream(dataStream,'id, 'temperature, 'timestamp.rowtime as 'ts) 
  47.    tableEnv.createTemporaryView("sensor",dataTable) 
  48.    var tableRes= dataTable.window( Over partitionBy 'id orderBy  'ts preceding 2.rows as 'ow) 
  49.     .select('id,'ts,'id.count over 'ow, 'temperature.avg over 'ow) 
  50.  
  51.   var tableSql= tableEnv.sqlQuery( 
  52.      """ 
  53.        |select 
  54.        |id, 
  55.        |ts, 
  56.        |count(id) over ow, 
  57.        |avg(temperature) over ow 
  58.        |from sensor 
  59.        |window ow as( 
  60.        | partition by id 
  61.        | order by ts 
  62.        | rows between 2 preceding and current row 
  63.        |) 
  64.        |""".stripMargin) 
  65.  
  66.    tableRes.toAppendStream[Row].print("tableRes") 
  67.    tableSql.toAppendStream[Row].print("tableSql") 
  68.    env.execute("FlinkSqlTumBlingOverTime") 
  69.  } 
  70.  case class SensorReading(id: String, timestamp: Long, temperature: Double) 
  71.  
  72. } 

運行結果

 

 

總結

好了到這里FlinkSql中窗口使用到這里就結束啦，喜歡的可以給了三連。其中FlinkSql中的窗口的用法還是比較多得，所有還是要多加練習。老話說的好，師傅領進門，修行在個人。

本文轉載自微信公眾號「 大數據老哥」，可以通過以下二維碼關注。轉載本文請聯系 大數據老哥公眾號。

 

責任編輯：武曉燕 來源： 大數據老哥
FlinkSQL排序
相關推薦

2020-11-27 08:02:41

Promise

2025-01-09 11:26:47

2021-10-15 07:57:04

Docker 日志容器

2018-11-21 08:00:05

Dubbo分布式系統

2022-02-21 09:44:45

Git開源分布式

2021-06-30 00:20:12

Hangfire.NET平臺

2023-05-12 08:19:12

Netty程序框架

2021-05-18 09:00:28

Python類class

2021-08-12 14:19:14

Slice數組類型內存

2020-09-29 15:13:14

C++語言開發

2023-05-08 08:21:15

JavaNIO編程

2021-09-27 09:18:30

ListIterato接口方法

2021-02-26 20:01:57

SVG濾鏡元素

2021-01-26 23:46:32

JavaScript數據結構前端

2023-07-30 15:18:54

JavaScript屬性

2021-05-18 08:30:42

JavaScript 前端JavaScript時

2021-03-09 14:04:01

JavaScriptCookie數據

2024-04-19 14:23:52

SwitchJavaScript開發

2020-12-08 08:09:49

SVG圖標Web

2021-06-24 09:05:08

JavaScript日期前端
點贊
收藏

51CTO技術棧公眾號

主站蜘蛛池模板： 91黄在线观看 | 成人免费三级电影 | 欧美a区 | 国产真实乱对白精彩久久小说 | 日韩中文一区二区三区 | 国产精品99久久久精品免费观看 | 黄网免费 | 久久一| 欧美久久一区 | 高清一区二区 | 在线观看深夜视频 | 成人亚洲 | 91新视频 | 欧美精品一区二区在线观看 | 久久成人免费 | 国外成人在线视频网站 | av大片在线 | 成人三级视频 | 国产欧美精品一区二区 | 日日碰狠狠躁久久躁婷婷 | 999久久久久久久久6666 | 成人免费视频网站在线看 | 亚洲国产精品91 | 午夜大片 | 91麻豆久久久 | 亚洲成人免费观看 | 午夜影院在线视频 | 岛国在线免费观看 | 精品国产乱码久久久久久久久 | 国产精品视频网 | 免费性视频 | 国产亚洲精品久久久久动 | av色站| 国产女人与拘做视频免费 | 精品福利在线 | 黑人巨大精品欧美一区二区免费 | 久久国产精品无码网站 | 久热伊人 | 精品国模一区二区三区欧美 | 亚洲女人天堂成人av在线 | 91精品国产91久久久久久最新 |