一、基本代码结构

<!--需要引入如下依赖
        <dependency>
                        <!-- 用于把sql生成一个执行计划，旧版本；新版本一般带有blink的 -->
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
                        <!-- 桥接器，用于转换tableapi和datastream，分为scala或者java版本 -->
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>-->
package tableapi
import datasource.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala._ //隐式转换
object TableAPITest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val outputTag = new OutputTag[SensorReading]("side")
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.getConfig.setAutoWatermarkInterval(5000)
    val inputStream: DataStream[String] = env.socketTextStream("127.0.0.1",7777)
    val dataStream : DataStream[SensorReading]= inputStream.map(data=>{
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0),arr(1).trim.toLong,arr(2).trim.toDouble)
    })
    //创建表执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
        //TableEnvironment是flink中集成Table API 和 SQL的核心概念，所有对表的操作都是基于TableEnvironment
    //如：注册cataLog、在catalog中注册表、执行SQL查询、注册用户自定义函数
    //第一种方式
    //基于流创建一个表，默认转换后的的Table Schema和DataStream中的字段一一对应，也可以单独指定处理
    val dataTable: Table = tableEnv.fromDataStream(dataStream)
    //val dataTable: Table = tableEnv.fromDataStream(dataStream,"id","temperature","timestamp")
    //基于名称作字段转换，可以调换顺序，而且使用timestamp作为字段名会报错
    //val dataTable: Table = tableEnv.fromDataStream(dataStream , 'timestamp as 'ts , 'id as 'myid , 'temperature)
    //基于位置作字段转换，根据字段的顺序重命名
    //val dataTable: Table = tableEnv.fromDataStream(dataStream , 'myid , 'ts)
    //调用tableapi进行转换
    val resultTable: Table = dataTable.select("id , temperature")
        .filter("id == 'sensor_1'")
    //把Table类型转换成流类型，并输出
    resultTable.toAppendStream[(String,Double)].print("data = ")
    //第二种方式，通过sql执行
    tableEnv.createTemporaryView("dataTable",dataTable)
    val sql: String = "select id,temperature from dataTable where id='sensor_1'"
    val sqlResTable: Table = tableEnv.sqlQuery(sql)
    sqlResTable.toAppendStream[(String,Double)].print("sql resData = ")
    env.execute("Table API Test")
  }
}

1.2、基本编程结构

    //基本编程结构
    val tableEnv
    //创建一个表，用于读取数据
    tableEnv.connect().createTemporaryTable("inputTable")
    //创建一个表，用于把计算结果输出
    tableEnv.connect().createTemporaryTable("outputTable")
    //通过Table API 查询算子，得到一张结果表
    val result = tableEnv.from("inputTable").select()
    //通过sql查询语句，得到一张结果表
    val result = tableEnv.sqlQuery()
    //将结果写入输出表中
    result.insertInto("outputTable")

1.3、Table环境的创建

package tableapi
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
import org.apache.flink.table.api.scala.{BatchTableEnvironment, StreamTableEnvironment}
object TableAPITest2 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //创建表执行环境
    val tableEnv: Any = StreamTableEnvironment.create(env)
    //1.1 基于老版本planner的流处理
    val settings = EnvironmentSettings.newInstance()
      .useOldPlanner()
      .inStreamingMode()
      .build()
    val oldStreamTableEnv = StreamTableEnvironment.create(env , settings)
    //1.2 基于老版本的批处理
    val batchEnv = ExecutionEnvironment.getExecutionEnvironment
    val oldBatchTableEnc: BatchTableEnvironment = BatchTableEnvironment.create(batchEnv)
    //1.3 基于blink planner 的流处理
    val blinkStreamSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val blinkStreamTableEnv = StreamTableEnvironment.create(env , blinkStreamSettings)
    //1.4 基于blink planner 的批处理
    val blinkBatchSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inBatchMode()
      .build()
    val blinkBatchTableEnv: TableEnvironment = TableEnvironment.create(blinkBatchSettings)
  }
}

二、表（Table）

TableEnvironment可以注册Catalog，并可以基于Catalog注册表
表Table是由一个“标识符”来指定的，由三个部分组成：Catalog名、数据库(database)和对象名，如果不指定则默认为default
表可以是常规的Table，也可以是虚拟的视图View
常规表一般可以用来描述外部数据，比如文件、数据库或者消息队列的数据，也可以直接从DataStream转换而来
视图View可以从现有的表中创建，通常是Table API或者SQL查询的一个结果集

2.1、创建表

TableEnvironment可以调用 .connect() 方法，连接外部系统，并调用 .createTemporaryTable() 方法，在Catalog中注册表
```scala import org.apache.flink.api.scala.ExecutionEnvironment import org.apache.flink.streaming.api.scala. import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table, TableEnvironment} import org.apache.flink.table.api.scala. import org.apache.flink.table.descriptors.{Csv, FileSystem, Kafka, Schema}

tableEnv .connect() //定义表的数据来源，和外部系统建立连接 .withFormat() //定义数据格式化方法 .withSchema() //定义表结构 .createTemporaryTable() //创建临时表

如： //2.1 从文件读取数据 tableEnv .connect(new FileSystem().path(“path/sensor.txt”)) //定义连接到文件系统 .withFormat(new Csv()) //指定以csv的格式进行数据格式化 .withSchema(new Schema() .field(“id”,DataTypes.STRING()) .field(“timestamp”,DataTypes.BIGINT()) .field(“temperature”,DataTypes.DOUBLE()) ) .createTemporaryTable(“sensorTable”) //

val inputTable: Table = tableEnv.from(“sensorTable”) //基于注册好的表，转换成Table类型的数据 inputTable.toAppendStream[(String,Long,Double)].print()

//2.2 从kafka读取数据 tableEnv.connect(new Kafka() .version(“0.11”) .topic(“sensor”) .property(“zookeeper.connect”,”localhost:2181”) .property(“bootstrap.servers”,”localhost:9092”) ) .withFormat(new Csv) .withSchema(new Schema() .field(“id”,DataTypes.STRING()) .field(“timestamp”,DataTypes.BIGINT()) .field(“temperature”,DataTypes.DOUBLE()) ) .createTemporaryTable(“kafkaInputTable”)

//2.3 创建ES的连接 tableEnv.connect( new Elasticsearch() .version(“6”) .host(“localhost”, 9200, “http”) .index(“sensor”) .documentType(“temp”) ) .inUpsertMode() // 指定是 Upsert 模式 .withFormat(new Json()) .withSchema( new Schema() .field(“id”, DataTypes.STRING()) .field(“count”, DataTypes.BIGINT()) ) .createTemporaryTable(“esOutputTable”); aggResultTable.insertInto(“esOutputTable”);


<a name="rdbx4"></a>
## 2.2、输出表
- 表的输出，是通过将数据写入TableSink来实现
- TableSink是一个通用接口，可以支持不同的文件格式、存储数据和消息队列
- 数据表最直接的方法就是，通过Table.insertInto() 的方法将一个Table写入到注册后的TableSink中
- 如resultTable.insertInto("outputTable")
```scala
package tableapi
import datasource.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, GroupedTable, Table}
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}
object FileOutPutTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    val inputStream: DataStream[String] = env.socketTextStream("127.0.0.1",7777)
    val dataStream = inputStream.map(data=>{
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0).trim , arr(1).trim.toLong , arr(2).trim.toDouble)
    })
    val inputTable: Table = tableEnv.fromDataStream(dataStream)
    tableEnv.createTemporaryView("inputTable" , inputTable)
    //1.1 简单转换1
    val resultTable1: Table = tableEnv.sqlQuery("select * from inputTable where id = 'sensor_1'")
    //1.2 简单转换2
    val resultTable2: Table = inputTable.select('id , 'timestamp , 'temperature).filter('id === "sensor_1")
    resultTable2.toAppendStream[(String,Long,Double)].print("result")
    //1.3 聚合转换
    val aggTable = inputTable
      .groupBy('id) //基于id进行分组
      .select('id , 'id.count as 'count) //返回id 和 id数量 两个字段
    aggTable.toRetractStream[(String,Long)].print("agg") // 聚合后的Table需要toRetractStream转换成DataStream，会输出一个带有Boolean的参数表示数据是否有效，旧的结果为false，最新的结果为true
    //2.1 输出到文件，而聚合操作是不能写到文件系统的
    tableEnv.connect(new FileSystem().path("C:\\ideaworkshop\\FlinkDemo\\LearnFirst\\src\\main\\resources\\tableApiOutput.txt"))
      .withFormat(new Csv)
      .withSchema(new Schema()
        .field("id",DataTypes.STRING())
        .field("timestamp",DataTypes.BIGINT())
        .field("temperature",DataTypes.DOUBLE())
      )
      .createTemporaryTable("outputTable")
    resultTable2.insertInto("outputTable")
    env.execute("FileOutPut Test")
  }
}

2.3、更新模式

对于流式查询，需要声明如何在表和外部连接器之间执行转换
与外部系统交换的消息类型，由更新模式（Update Mode）指定

追加模式（Append）
—表只是做插入操作，和外部连接器只交换插入消息。
撤回模式（Retract）
—表和外部连接器交换添加（Add）和撤回（Retract）消息
—插入（Insert）操作编码为Add消息；删除（Delete）编码为Retract消息；更新（Update）编码为上一条的Retract和下一条的Add消息
更新插入模式（Upsert）
—更新和插入都被编码为Upsert消息；删除编码为Delete消息。

2.4、查看执行计划

Table API提供了一种机制来解释计算表的逻辑和优化查询计划
可以通过TableEnvironment.explain(table) 方法或 TableEnvironment.explain() 方法完成，返回一个字符串，描述三个计划
- 优化前的逻辑查询计划
- 优化后的逻辑查询计划
- 实际执行计划

2.5、流处理和关系代数的区别

	关系代数（表）/SQL	流处理
处理的数据对象	字段元组的有界集合	字段元组的无限序列
查询对数据的访问	可以访问到完整的数据输入	无法访问所有数据，必须持续“等待”流式输入
查询终止条件	生成固定大小的结果集后终止	永不停止，根据持续收到的数据不断更新查询结果

2.6、动态表

动态表是Flink对流数据的Table API 和 SQL支持的核心概念
与表示批处理数据的静态表不同，动态表是随时间变化的（随着数据的到来更新变化）

持续查询（Continuous Query）

动态表可以像静态的批处理一样进行查询，查询一个动态表会产生持续查询（Continuous Query）
连续查询永远不会终止，并会生成另一个动态表
查询会不断更新其动态结果表，以反映其动态输入表上的更改

动态表和持续查询（流式表查询的处理过程）
1、流被转换为动态表
2、对动态表计算查询查询，生成新的动态表
3、生成的动态表被转换成流
10、Flink Table和Flink SQL - 图1

2.7、Table和DataStream转换

1、追加模式AppendMode：用于表只被插入的场景
sqlResTable.toAppendStream[(String,Double)].print("sql resData = ")
或tableEnv.toAppendStream[(String,Double)](sqlResTable)
2、撤回模式RetractMode：
- 用于任何场景，类似于更新模式的Retract，只有Insert和Delete两类操作
- 得到的数据会增加一个Boolean类型的标识位，用来表示新增还是删除的的数据
tableEnv.toRetractStream()[(String,Double)](sqlAggTable)

三、时间特性

基于时间的操作（比如Table API和SQL中的窗口操作），需要定义相关的事件语义和事件数据来源的信息
Table可以提供一个逻辑上的时间字段（即原来数据没有的），用于在表处理程序中，指示时间和访问相应的时间戳
时间属性，可以是每个表schema的一部分。一旦定义了时间属性，它就可以作为一个字段引用，并且可以在基于时间的操作中使用
时间属性的行为类似于常规时间戳(timestamp)，可以访问，并且进行计算

3.1、定义处理时间（ProcessingTime）

处理时间语义下，允许表处理程序根据机器的本地时间生成结果。它是时间的最简单概念。它既不需要提取时间戳，也不需要生成watermark

1、由DataStream转换成表时指定

在定义Schema期间，可以使用 .proctime ，指定字段名定义处理时间字段
这个proctime属性只能通过附件逻辑字段，来扩展物理schema。因此，只能在schema定义末尾定义它 ```scala val sensorTable = tableEnv.fromDataStream(dataStream,’id,’temperature,’timestamp,’pt.proctime)

package tableapi import datasource.SensorReading import org.apache.flink.streaming.api.TimeCharacteristic import org.apache.flink.streaming.api.scala. import org.apache.flink.table.api.{DataTypes, Table} import org.apache.flink.table.api.scala. import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema} import org.apache.flink.types.Row object ProcessingTimeDemo { def main(args: Array[String]): Unit = { val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
env.setParallelism(1)
val inputStream: DataStream[String] = env.socketTextStream("127.0.0.1",7777)
val dataStream : DataStream[SensorReading]= inputStream.map(data=>{
  val arr: Array[String] = data.split(",")
  SensorReading(arr(0),arr(1).trim.toLong,arr(2).trim.toDouble)
})
//创建表执行环境
val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
//第一种方式
//指定Processing Time
val inputTable: Table = tableEnv.fromDataStream(dataStream,'id,'timestamp as 'ts,'temperature,'pt.proctime)
tableEnv.createTemporaryView("inputTable" , inputTable)
inputTable.printSchema()
tableEnv.sqlQuery("select id,pt,temperature,ts from inputTable").toAppendStream[Row].print()
//第二种方式，指定Processing Time
tableEnv.connect(new FileSystem().path("C:\\ideaworkshop\\FlinkDemo\\LearnFirst\\src\\main\\resources\\sensor.txt"))
    .withSchema(new Schema()
      .field("id",DataTypes.STRING())
      .field("timestamp",DataTypes.BIGINT())
      .field("temperature",DataTypes.DOUBLE())
      .field("pt",DataTypes.TIMESTAMP(3)).proctime //追加一个字段，类型为DataTypes.TIMESTAMP(3)，且指定该字段为proctime，但是这种方法不是所有的都支持，如kafka的连接器不支持
    )
    .withFormat(new Csv)
    .createTemporaryTable("fileTable")
//第三种，在DDL里定义
env.execute("ProcessingTimeDemo")

} }



<a name="gFbCx"></a>
## 3.2、定义事件时间（Event Time）

- 事件事件语义，允许表处理程序根据每个记录中包含的时间生成结果。这样即使在有乱序事件或者延迟事件时，也可以获得争取二点结果。
- 为了处理无序事件，并区分流中的准时和迟到事件；Flink需要从事件数据中提取时间戳，并用来推进事件事件的进展
- 定义事件时间，有三种方法：
   - 由DataStream转换成Table时指定，使用 .rowTime 可以定义事件事件属性
   - 定义Table Schema时指定
   - 在创建表的DDL中定义

```scala
package tableapi

import datasource.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors.{Csv, FileSystem, Rowtime, Schema}

object EventTimeDemo {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val inputStream: DataStream[String] = env.socketTextStream("127.0.0.1",7777)
    val dataStream : DataStream[SensorReading]= inputStream.map(data=>{
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0),arr(1).trim.toLong,arr(2).trim.toDouble)
    })

    //创建表执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    //第一种方式，在转换DataStream时指定EventTime，
    val inputTable1: Table = tableEnv.fromDataStream(dataStream,'id,'timestamp.rowtime as 'ts,'temperature)//指定timestamp为时间字段
    val inputTable2: Table = tableEnv.fromDataStream(dataStream,'id,'timestamp as 'ts,'temperature,'rt.rowtime)//或者直接追加一个新的时间字段
    tableEnv.createTemporaryView("inputTable" , inputTable2)

    //第二种方式，定义schema时指定
    tableEnv.connect(new FileSystem().path("C:\\ideaworkshop\\FlinkDemo\\LearnFirst\\src\\main\\resources\\sensor.txt"))
      .withFormat(new Csv)
      .withSchema(new Schema()
        .field("id",DataTypes.STRING())
        .field("timestamp",DataTypes.BIGINT())
        .field("temperature",DataTypes.DOUBLE())
        .rowtime(new Rowtime()
          .timestampsFromField("timestamp") //从字段中提取时间戳,这里指定字段也可以是一个新的字段,
          .watermarksPeriodicBounded(1000) //watermark延迟1秒
        ) 
      )

    //第三种，使用DDL定义

  }
}

四、窗口

在Table API 和 SQL中，主要有两种窗口

Group Windows(分组窗口)
- 根据时间或行计数间隔，将行聚合到有限的组(Group)中，并对每个组的数据执行一次聚合函数（使用group by） ```scala val table = input.window([w:GroupWindow] as ‘w) // 定义窗口，别名为w
```
                        .groupBy('w,'a)        //按照字段a 和 窗口w分组;如果不指定字段a，则相当于没有keyBy，直接使用windowAll
                        .select('a,'b.sum) //聚合
```
  //Table API提供了一组具有特定语义的预定义Window类，这些类会被转换为底层DataStream或者DataSet //滚动窗口Tumble .window(Tumble over 10.minutes on ‘rowtime as ‘w) //事件事件滚动窗口 .window(Tumble over 10.minutes on ‘proctime as ‘w)//处理事件滚动窗口 .window(Tumble over 10.rows on ‘proctime as ‘w)//滚动计数窗口,会根据proctime排序,每10行数据取一组

//滑动窗口Slide .window(Slide over 10.minutes every 5.minutes on ‘rowtime as ‘w) //滑动事件时间窗口 .window(Slide over 10.minutes every 5.minutes on ‘proctime as ‘w) //滑动处理时间窗口 .window(Slide over 10.rows every 5.rows on ‘proctime as ‘w) //滑动计数窗口

//会话窗口Session .window(Session withGap 10.minutes on ‘rowtime as ‘w) //会话事件时间窗口，相隔超过10分钟，会话关闭 .window(Session withGap 10.minutes on ‘proctime as ‘w) //会话事件时间窗口

//SQL中的Group Windows //在SQL中定义窗口，需要定义在Group By子句中 TUMBLE(time_attr,interval)//滚动窗口，第一个参数是时间字段，第二个参数是窗口长度 HOP(time_attr,interval,interval)//滑动窗口，第一个参数是时间字段，第二个参数是窗口滑动步长，第三个参数是窗口长度 SESSION(time_attr,interval)//会话窗口，第一个参数是时间字段，第二个参数是窗口间隔


- **Over Windows**
   - 针对每个输入行，计算相邻行范围内（很多行）的聚合
   - Over Window聚合是标准sql中已有的（Over子句），可以在查询的SELECT子句中定义
   - Over Window 使用window(w:overwindow*)子句定义，并在select()方法中通过别名来定义
```scala
val table = input.window([w:OverWindow] as 'w)
                 .select('a , 'b.sum over 'w , 'c.min over 'w)

//无界Over Windows
//可以在时间时间或者处理时间，以及指定时间为时间间隔、或行计数的范围内，定义Over Windows
//1、无界的事件时间over windows
.window(Over partitionBy 'a orderBy 'rowtime preceding UNBOUNDED_RANGE as 'w) //排序字段不可少
//2、无界的处理时间over windows
.window(Over partitionBy 'a orderBy 'proctime preceding UNBOUNDED_RANGE as 'w)
//3、无界的事件时间Row-Count over windows
.window(Over partitionBy 'a orderBy 'rowtime preceding UNBOUNDED_ROW as 'w)
//4、无界的处理时间Row-Count over windows
.window(Over partitionBy 'a orderBy 'proctime preceding UNBOUNDED_ROW as 'w)

//有界的over window是用间隔大小来指定的
//1、有界的事件时间over windows
.window(Over partitionBy 'a orderBy 'rowtime preceding 1.minutes as 'w) //排序字段不可少
//2、有界的处理时间over windows
.window(Over partitionBy 'a orderBy 'proctime preceding 1.minutes as 'w)
//3、有界的事件时间Row-Count over windows
.window(Over partitionBy 'a orderBy 'rowtime preceding 10.rows as 'w)
//4、有界的处理时间Row-Count over windows
.window(Over partitionBy 'a orderBy 'proctime preceding 10.rows as 'w)

//SQL的例子
//使用Over做窗口聚合时，所有聚合必须在同一窗口上定义，也就是必须是相同的分区、排序和范围
//目前仅支持在当前行范围之前的窗口
//ORDER BY 必须在单一的时间属性上指定
select count(amount) OVER(
    PARTITION BY user
  ORDER BY proctime
  ROWS BETWEEN 2 PRECEDING AND CURRENT NOW
)

4.1、具体代码demo

package tableapi

import datasource.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.{Over, OverWindow, Table, Tumble}
import org.apache.flink.table.api.scala._
import org.apache.flink.types.Row

object TableWindowDemo {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val inputStream: DataStream[String] = env.readTextFile("C:\\ideaworkshop\\FlinkDemo\\LearnFirst\\src\\main\\resources\\TableWindowDemo.txt")

    val dataStream : DataStream[SensorReading]= inputStream.map(data=>{
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0),arr(1).trim.toLong,arr(2).trim.toDouble)
    }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(3)) {
      override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L
    })

    //创建表执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    val inputTable: Table = tableEnv.fromDataStream(dataStream,'id,'timestamp.rowtime as 'ts,'temperature)//指定timestamp为时间字段

    //一、Group Window
    //1、table api实现
    val resTable = inputTable.window(Tumble over 10.seconds on 'ts as 'w) //每10秒统计一次，滚动时间窗口
      .groupBy('w , 'id)
      .select('id , 'id.count , 'temperature.avg , 'w.end)


    //2、sql实现
    tableEnv.createTemporaryView("sensor",inputTable)
    val resultTable:Table = tableEnv.sqlQuery(
      """
        |select
        | id ,
        | count(id) ,
        | avg(temperature),
        | tumble_end(ts , interval '10' second)
        | from sensor
        | group by
        |   id ,
        |   tumble(ts , interval '10' second)
      """.stripMargin)


//    resTable.toRetractStream[Row].print("result")
//    resultTable.toRetractStream[Row].print("sql")

    //二、Group Window
    //统计每个sensor每条数据，与之前两行数据的平均温度
    //1、table api实现
    val overresultTable = inputTable.window( Over partitionBy 'id orderBy 'ts preceding 2.rows as 'ow)
        .select('id , 'ts , 'id.count over 'ow , 'temperature.avg over 'ow)
    overresultTable.toAppendStream[Row].print("overWindow")

    //2、sql实现
    tableEnv.sqlQuery(
      """
        |select
        | id,
        | ts,
        | count(id) over ow,
        | avg(temperature) over ow
        |from sensor
        |window ow as (
        | partition by id
        | order by ts
        | rows between 2 preceding and current row
        |)
      """.stripMargin
    )

    env.execute("Window Test")
  }
}

五、函数

5.1、基本代码

package udf

import datasource.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.{Table, Tumble}
import org.apache.flink.table.api.scala._
import org.apache.flink.table.functions.{ScalarFunction, TableAggregateFunction, TableFunction}
import org.apache.flink.types.Row
import org.apache.flink.util.Collector

//表聚合函数
//可以把一个表的数据，聚合为具有多个行和多列的结果表
object TableAggregateFunctionTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val inputStream: DataStream[String] = env.readTextFile("C:\\ideaworkshop\\FlinkDemo\\LearnFirst\\src\\main\\resources\\TableWindowDemo.txt")

    val dataStream : DataStream[SensorReading]= inputStream.map(data=>{
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0),arr(1).trim.toLong,arr(2).trim.toDouble)
    }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(3)) {
      override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L
    })

    //创建表执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    val inputTable: Table = tableEnv.fromDataStream(dataStream,'id,'timestamp.rowtime as 'ts,'temperature)//指定timestamp为时间字段
  }
}

5.2、ScalaFunction

    //调用自定义hash函数
    //1、table api
    val hashCode = new HashCode(23)
    val res = inputTable.select('id , 'ts , hashCode('id))
    res.toAppendStream[Row].print("res")

    //2、sql,需要在环境中注册udf函数
    tableEnv.createTemporaryView("sensor",inputTable)
    tableEnv.registerFunction("hashcode",new HashCode(23))//通过registerFunction在tableEnvironment中注册
    val sqlRes = tableEnv.sqlQuery(
      """
        |select
        | id,
        | ts,
        | hashcode(id)
        | from sensor
      """.stripMargin)
    sqlRes.toAppendStream[Row].print("sql")

    env.execute("scala function test")

//自定义标量函数
//实现值一对一转换的函数
//求值方法必须公开声明并命名为eval
class HashCode(factor : Int) extends ScalarFunction{

  def eval(s:String):Int = {
    s.hashCode * factor - 10000
  }

}

5.3、TableFunction

    //表函数的用法
    //1、table api使用
    val split = new Split("")
    val resTableFunc = inputTable
      .joinLateral(split('id) as ('words , 'length)) //使用joinLateral 侧向连接
      .select('id , 'ts , 'words , 'length )
    resTableFunc.toAppendStream[Row].print("res")

    //2、sql调用
    tableEnv.createTemporaryView("sensor",inputTable)
    tableEnv.registerFunction("split",new Split("-"))
    val resTablesqlFun = tableEnv.sqlQuery(
      """
        |select
        | id , ts , word , length
        | from
        | sensor , lateral table( split(id) ) as splitid(word,length)
      """.stripMargin)
    resTablesqlFun.toAppendStream[Row].print("sql")

    env.execute("table function test")
  }

}

//自定义表函数
//可以返回任意数量的行作为输出，而不是单个值
//求值方法必须公开声明并命名为eval
class Split(seperator : String) extends TableFunction[(String,Int)]{//指定返回值类型

  def eval(str : String):Unit={
    str.split(seperator).foreach(
      words=>{
        collect((words , words.length))
      }
    )
  }

}

5.4、AggregateFunction

//1、table api调用
    val myAvg = new MyAvg()
    val resTable = inputTable.groupBy('id)
      .aggregate(myAvg('temperature) as 'avgTemp)
      .select('id ,  'avgTemp)
    resTable.toRetractStream[Row].print("res")

//2、sql调用
    tableEnv.registerFunction("myAvg",myAvg)
    tableEnv.createTemporaryView("sensor",inputTable)
    val sqlRes = tableEnv.sqlQuery(
      """
        |select
        | id , myAvg(temperature)
        | from sensor
        | group by id
      """.stripMargin)
    sqlRes.toRetractStream[Row].print("sql")

    env.execute("AvgFunction Test")

class TempTempAcc{
  var  sum: Double = 0.0
  var  count:Int = 0
}

//自定义聚合函数，求平均值
//作用：可以把表中的数据，聚合成一个标量值
//自定义聚合函数来说至少需要createAccumulator、accumulate、getValue这三个方法，并且这三个方法是public 、not static的类型
class MyAvg extends AggregateFunction[Double , TempTempAcc]{ //泛型第一个参数表示输出值;第二个泛型参数表示中间处理类型
  //表示最终返回的结果，结果数据从acc中获取，这个方法在accumulate之后被调用。
  override def getValue(accumulator: TempTempAcc): Double = {
    accumulator.sum / accumulator.count //求平均值
  }

  //表示创建一个中间结果数据，用来保存聚合中间结果的数据结构
  override def createAccumulator(): (TempTempAcc) = {
    new TempTempAcc //初始化
  }

  //具体的处理计算函数对每行输入数据更新累加值
  //第一个参数表示中间值，第二个参数表示每行的传入值
  def accumulate(accumulator : TempTempAcc , temp: Double):Unit={
    accumulator.sum += temp
    accumulator.count += 1
  }
}

5.5、TableAggregateFunction

可以把一个表的数据，聚合为具有多个行和多列的结果表

    val top2Temp = new MyTableAggregateFun
    val resTable = inputTable.groupBy('id)
        .flatAggregate(top2Temp('temperature) as ('temp,'rank))
        .select('id , 'temp , 'rank)
    resTable.toRetractStream[Row].print("res")

    env.execute("table aggregate function test")
  }
}

class TopTwoAcc{
  var highestTemp : Double = Double.MinValue
  var sencondHighestTemp : Double = Double.MinValue
}

//自定义表聚合函数，提取所有温度值中最高的两个温度 , 输出(temp,rank)
class MyTableAggregateFun extends TableAggregateFunction[(Double,Int),TopTwoAcc]{ // 输出值类型 和 中间处理状态类型
  //
  override def createAccumulator(): TopTwoAcc = {
    new TopTwoAcc
  }

  //实现计算聚合结果的函数accumulate
  def accumulate(acc : TopTwoAcc , temp : Double):Unit = {

    //判断当前的温度值是否比前面的温度值大
    if (temp > acc.highestTemp){
      acc.sencondHighestTemp = acc.highestTemp
      acc.highestTemp = temp
    }else if(temp > acc.highestTemp && temp > acc.sencondHighestTemp){
      acc.sencondHighestTemp = temp
    }

  }

  //实现一个输出结果的方法，最终处理完所有数据时调用
  def emitValue(acc : TopTwoAcc , out: Collector[(Double,Int)]): Unit ={ //使用Collector 指定输出的类型，这里是一个二元组
    out.collect((acc.highestTemp,1))
    out.collect((acc.sencondHighestTemp,2))
  }

}