自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

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

作者： 大數(shù)據(jù)老哥
大數(shù)據(jù)
時(shí)間語(yǔ)義，要配合窗口操作才能發(fā)揮作用。最主要的用途，當(dāng)然就是開(kāi)窗口、根據(jù)時(shí)間段做計(jì)算了。下面我們就來(lái)看看 Table API 和 SQL 中，怎么利用時(shí)間字段做窗口操作。

[[360693]]

前言

時(shí)間語(yǔ)義，要配合窗口操作才能發(fā)揮作用。最主要的用途，當(dāng)然就是開(kāi)窗口、根據(jù)時(shí)間段做計(jì)算了。下面我們就來(lái)看看 Table API 和 SQL 中，怎么利用時(shí)間字段做窗口操作。在 Table API 和 SQL 中，主要有兩種窗口：Group Windows 和 Over Windows

 

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

  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 提供了一組具有特定語(yǔ)義的預(yù)定義 Window 類(lèi)，這些類(lèi)會(huì)被轉(zhuǎn)換為底層DataStream 或 DataSet 的窗口操作。

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

1.1 滾動(dòng)窗口

滾動(dòng)窗口(Tumbling windows)要用 Tumble 類(lèi)來(lái)定義，另外還有三個(gè)方法：

  • over：定義窗口長(zhǎng)度
  • on：用來(lái)分組(按時(shí)間間隔)或者排序(按行數(shù))的時(shí)間字段
  • as：別名，必須出現(xiàn)在后面的 groupBy 中

實(shí)現(xiàn)案例

1.需求

設(shè)置滾動(dòng)窗口為10秒鐘統(tǒng)計(jì)id出現(xiàn)的次數(shù)。

2.數(shù)據(jù)準(zhǔn)備

  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.代碼實(shí)現(xiàn)

  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 大數(shù)據(jù)老哥 
  15.  * @date 2020/12/25 21:58 
  16.  * @version V1.0 
  17.  *          設(shè)置滾動(dòng)窗口 
  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.     // 讀取數(shù)據(jù) 
  32.     val inputPath = "./data/sensor.txt" 
  33.     val inputStream = env.readTextFile(inputPath) 
  34.     
  35.  
  36.     // 先轉(zhuǎn)換成樣例類(lèi)類(lèi)型（簡(jiǎn)單轉(zhuǎn)換操作） 
  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.     // 注冊(cè)表 
  48.     tableEnv.createTemporaryView("sensor", sensorTable) 
  49.     // table 實(shí)現(xiàn) 
  50.     val resultTable = sensorTable 
  51.       .window(Tumble over 10.seconds on 'ts as 'tw) // 每10秒統(tǒng)計(jì)一次，滾動(dòng)時(shí)間窗口 
  52.       .groupBy('id, 'tw) 
  53.       .select('id, 'id.count, 'tw.end) 
  54.     //sql 實(shí)現(xiàn) 
  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) （事件時(shí)間字段 rowtime） 
  69.      * .window(Tumble over 10.minutes on 'proctime as 'w)（處理時(shí)間字段 proctime） 
  70.      * .window(Tumble over 10.minutes on 'proctime as 'w) (類(lèi)似于計(jì)數(shù)窗口，按處理時(shí)間排序，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. } 

運(yùn)行結(jié)果

 

1.2 滑動(dòng)窗口

滑動(dòng)窗口(Sliding windows)要用 Slide 類(lèi)來(lái)定義，另外還有四個(gè)方法：

  • over：定義窗口長(zhǎng)度
  • every：定義滑動(dòng)步長(zhǎng)
  • on：用來(lái)分組(按時(shí)間間隔)或者排序(按行數(shù))的時(shí)間字段
  • as：別名，必須出現(xiàn)在后面的 groupBy 中

實(shí)現(xiàn)案例

1.需求描述

設(shè)置窗口大小為10秒鐘設(shè)置滑動(dòng)距離為5秒鐘，統(tǒng)計(jì)id的出現(xiàn)的次數(shù)。

2.數(shù)據(jù)準(zhǔn)備

  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.實(shí)現(xiàn)代碼

  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 大數(shù)據(jù)老哥 
  16.  * @date 2020/12/27 22:19 
  17.  * @version V1.0 
  18.  *          滑動(dòng)窗口 
  19.  */ 
  20. object FlinkSQLSlideTime { 
  21.   def main(args: Array[String]): Unit = { 
  22.     //構(gòu)建運(yùn)行環(huán)境 
  23.     val env = StreamExecutionEnvironment.getExecutionEnvironment 
  24.     env.setParallelism(1) // 設(shè)置分區(qū)為1 方便后面測(cè)試 
  25.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //事件時(shí)間 
  26.  
  27.     val settings = EnvironmentSettings.newInstance() 
  28.       .useBlinkPlanner() 
  29.       .inStreamingMode() 
  30.       .build() 
  31.     // 創(chuàng)建表env 
  32.     val tableEnv = StreamTableEnvironment.create(env, settings) 
  33.  
  34.     // 讀取數(shù)據(jù) 
  35.     val inputPath = "./data/sensor.txt" 
  36.     val inputStream = env.readTextFile(inputPath) 
  37.  
  38.     // 先轉(zhuǎn)換成樣例類(lèi)類(lèi)型（簡(jiǎn)單轉(zhuǎn)換操作） 
  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.     // 注冊(cè)表 
  50.     tableEnv.createTemporaryView("sensor", sensorTable) 
  51.     // table API 實(shí)現(xiàn) 
  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) （事件時(shí)間字段 rowtime） 
  70. .window(Slide over 10.minutes every 5.minutes on 'proctime as 'w) （處理時(shí)間字段 proctime）  
  71. .window(Slide over 10.rows every 5.rows on 'proctime as 'w) (類(lèi)似于計(jì)數(shù)窗口，按處理時(shí)間排序，10 行一組) 
  72.    **/ 
  73.     env.execute("FlinkSQLSlideTime") 
  74.   } 
  75. } 

4.運(yùn)行結(jié)果

 

1.3 會(huì)話窗口

會(huì)話窗口(Session windows)要用 Session 類(lèi)來(lái)定義，另外還有三個(gè)方法：

  • withGap：會(huì)話時(shí)間間隔
  • on：用來(lái)分組(按時(shí)間間隔)或者排序(按行數(shù))的時(shí)間字段
  • as：別名，必須出現(xiàn)在后面的 groupBy 中實(shí)現(xiàn)案例

1.需求描述

設(shè)置一個(gè)session 為10秒鐘 統(tǒng)計(jì)id的個(gè)數(shù)

2.準(zhǔn)備數(shù)據(jù)

  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.編寫(xiě)代碼

  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 大數(shù)據(jù)老哥 
  16.  * @date 2020/12/27 22:52 
  17.  * @version V1.0 
  18.  */ 
  19. object FlinkSqlSessionTime { 
  20.   def main(args: Array[String]): Unit = { 
  21.     //構(gòu)建運(yùn)行環(huán)境 
  22.     val env = StreamExecutionEnvironment.getExecutionEnvironment 
  23.     env.setParallelism(1) // 設(shè)置分區(qū)為1 方便后面測(cè)試 
  24.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //事件時(shí)間 
  25.  
  26.     val settings = EnvironmentSettings.newInstance() 
  27.       .useBlinkPlanner() 
  28.       .inStreamingMode() 
  29.       .build() 
  30.     // 創(chuàng)建表env 
  31.     val tableEnv = StreamTableEnvironment.create(env, settings) 
  32.  
  33.     // 讀取數(shù)據(jù) 
  34.     val inputPath = "./data/sensor.txt" 
  35.     val inputStream = env.readTextFile(inputPath) 
  36.  
  37.     // 先轉(zhuǎn)換成樣例類(lèi)類(lèi)型（簡(jiǎn)單轉(zhuǎn)換操作） 
  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.     // 注冊(cè)表 
  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) 事件時(shí)間字段 rowtime） 
  71.      * .window(Session withGap 10.minutes on 'proctime as 'w) 處理時(shí)間字段 proctime） 
  72.      */ 
  73.     env.execute("FlinkSqlSessionTime") 
  74.   } 
  75. } 

4.運(yùn)行結(jié)果

 

二、 Over Windows

Over window 聚合是標(biāo)準(zhǔn) SQL 中已有的(Over 子句)，可以在查詢(xún)的 SELECT 子句中定義。Over window 聚合，會(huì)針對(duì)每個(gè)輸入行，計(jì)算相鄰行范圍內(nèi)的聚合。Over windows使用.window(w:overwindows*)子句定義，并在 select()方法中通過(guò)別名來(lái)引用。例子：

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

Table API 提供了 Over 類(lèi)，來(lái)配置 Over 窗口的屬性。可以在事件時(shí)間或處理時(shí)間，以及指定為時(shí)間間隔、或行計(jì)數(shù)的范圍內(nèi)，定義 Over windows。

無(wú)界的 over window 是使用常量指定的。也就是說(shuō)，時(shí)間間隔要指定 UNBOUNDED_RANGE，或者行計(jì)數(shù)間隔要指定 UNBOUNDED_ROW。而有界的 over window 是用間隔的大小指定的。

2.1 無(wú)界的 over window

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

2.2 有界的 over window

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

2.3 代碼練習(xí)

我們可以綜合學(xué)習(xí)過(guò)的內(nèi)容，用一段完整的代碼實(shí)現(xiàn)一個(gè)具體的需求。例如，統(tǒng)計(jì)每個(gè)sensor每條數(shù)據(jù)，與之前兩行數(shù)據(jù)的平均溫度。

數(shù)據(jù)準(zhǔn)備

  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 大數(shù)據(jù)老哥 
  15. * @date 2020/12/28 21:45 
  16. * @version V1.0 
  17. */ 
  18. object FlinkSqlTumBlingOverTime { 
  19.  def main(args: Array[String]): Unit = { 
  20.    // 構(gòu)建運(yùn)行環(huán)境 
  21.    val env = StreamExecutionEnvironment.getExecutionEnvironment 
  22.    env.setParallelism(1) // 設(shè)置并行度為1方便后面進(jìn)行測(cè)試 
  23.    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 設(shè)置事件時(shí)間 
  24.  
  25.    val settings = EnvironmentSettings.newInstance() 
  26.      .useBlinkPlanner() 
  27.      .inStreamingMode() 
  28.      .build() 
  29.    //構(gòu)建table Env 
  30.    val tableEnv = StreamTableEnvironment.create(env, settings) 
  31.  
  32.    // 讀取數(shù)據(jù) 
  33.    val inputPath = "./data/sensor.txt" 
  34.    val inputStream = env.readTextFile(inputPath) 
  35.    // 先轉(zhuǎn)換成樣例類(lèi)類(lèi)型（簡(jiǎn)單轉(zhuǎn)換操作） 
  36.    // 解析數(shù)據(jù) 封裝成樣例類(lèi) 
  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.    // 將數(shù)據(jù)注冊(cè)成一張臨時(shí)表 
  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. } 

運(yùn)行結(jié)果

 

 

總結(jié)

好了到這里FlinkSql中窗口使用到這里就結(jié)束啦，喜歡的可以給了三連。其中FlinkSql中的窗口的用法還是比較多得，所有還是要多加練習(xí)。老話說(shuō)的好，師傅領(lǐng)進(jìn)門(mén)，修行在個(gè)人。

本文轉(zhuǎn)載自微信公眾號(hào)「 大數(shù)據(jù)老哥」，可以通過(guò)以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系 大數(shù)據(jù)老哥公眾號(hào)。

 

責(zé)任編輯：武曉燕 來(lái)源： 大數(shù)據(jù)老哥
FlinkSQL排序
相關(guān)推薦

2020-11-27 08:02:41

Promise

2025-01-09 11:26:47

2018-11-21 08:00:05

Dubbo分布式系統(tǒng)

2021-10-15 07:57:04

Docker 日志容器

2022-02-21 09:44:45

Git開(kāi)源分布式

2023-05-12 08:19:12

Netty程序框架

2021-06-30 00:20:12

Hangfire.NET平臺(tái)

2021-05-18 09:00:28

Python類(lèi)class

2021-08-12 14:19:14

Slice數(shù)組類(lèi)型內(nèi)存

2020-09-29 15:13:14

C++語(yǔ)言開(kāi)發(fā)

2023-07-30 15:18:54

JavaScript屬性

2023-05-08 08:21:15

JavaNIO編程

2021-01-26 23:46:32

JavaScript數(shù)據(jù)結(jié)構(gòu)前端

2021-03-09 14:04:01

JavaScriptCookie數(shù)據(jù)

2021-06-24 09:05:08

JavaScript日期前端

2021-09-27 09:18:30

ListIterato接口方法

2023-09-06 14:57:46

JavaScript編程語(yǔ)言

2024-01-30 13:47:45

2024-04-19 14:23:52

SwitchJavaScript開(kāi)發(fā)

2020-12-08 08:09:49

SVG圖標(biāo)Web
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)