Flink学习笔记(9)-Table API 和 Flink SQL开发者社区

Flink学习笔记(9)-Table API 和 Flink SQL

// 创建表的执行环境
val tableEnv = StreamExecutionEnvironment.getExecutionEnvironment     
// 创建一张表，用于读取数据
tableEnv.connect(...).createTemporaryTable("inputTable")
// 注册一张表，用于把计算结果输出
tableEnv.connect(...).createTemporaryTable("outputTable")
// 通过 Table API 查询算子，得到一张结果表
val result = tableEnv.from("inputTable").select(...)
// 通过 SQL查询语句，得到一张结果表
val sqlResult  = tableEnv.sqlQuery("SELECT ... FROM inputTable ...")
// 将结果表写入输出表中
result.insertInto("outputTable")

    val tableEnv = StreamTableEnvironment.create(env)

val settings = EnvironmentSettings.newInstance()
  .useOldPlanner()
  .inStreamingMode()
  .build()
val tableEnv = StreamTableEnvironment.create(env, settings)

val batchEnv = ExecutionEnvironment.getExecutionEnvironment
val batchTableEnv = BatchTableEnvironment.create(batchEnv)

val bsSettings = EnvironmentSettings.newInstance()
  .useBlinkPlanner()
  .inStreamingMode()
  .build()
val bsTableEnv = StreamTableEnvironment.create(env, bsSettings)

val bbSettings = EnvironmentSettings.newInstance()
  .useBlinkPlanner()
  .inBatchMode()
  .build()
  val bbTableEnv = TableEnvironment.create(bbSettings)

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

val filePath = "***********"
tableEnvironment.connect(new FileSystem().path(filePath))
//      .withFormat(new OldCsv()) //定义读取数据之后的格式化方法
      // 新版CSV文件系统需要引入依赖
//    <dependency>
//      <groupId>org.apache.flink</groupId>
//      <artifactId>flink-csv</artifactId>
//      <version>1.10.0</version>
//    </dependency>
      .withFormat(new Csv())
      .withSchema(
        new Schema()
          .field("id",DataTypes.STRING())
          .field("timestamp",DataTypes.BIGINT())
          .field("temperature",DataTypes.DOUBLE())
      .createTemporaryTable("CSVTable")
    //转换成流打印输出
    val table = tableEnvironment.from("CSVTable")
    table.toAppendStream[(String,Long,Double)].print()

tableEnvironment.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("KafkaTable")
    //转换成流打印输出
    val KafkaTable = tableEnvironment.from("KafkaTable")
    KafkaTable.toAppendStream[(String,Long,Double)].print()

// 定义样例类，传感器id，时间戳，温度
case class SensorReading(id: String, timestamp: Long, temperature: Double)
val sensorTable: Table = tableEnv.from("sensorTable")
val resultTable: Table = sensorTable
    .select("id, temperature")
    .filter("id = 'sensor_1'")

    //简单聚合查询
  val resultTable: Table =   sensorTable
    .groupBy("id")
    .select('id,'timestamp.max as 'max_timestamp)

val resultSqlTable: Table = tableEnv
   .sqlQuery("select id, temperature from sensorTable where id ='sensor_1'")

val resultSqlTable: Table = tableEnvironment.sqlQuery(
        |select
        | id,max(timestamp) timestamp
        |from
        | CSVTable
        |group by
        |""".stripMargin)

val inputStream:DataStream[String] = environment.readTextFile("")
val datastream:DataStream[SensorReading] = inputStream.map(
    data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
val sensorTable: Table = tableEnv.fromDataStream(dataStream)

val inputStream:DataStream[String] = environment.readTextFile("")
val datastream:DataStream[SensorReading] = inputStream.map(
    data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
val sensorTable = tableEnv.fromDataStream(dataStream,'id, 'timestamp, 'temperature)

val sensorTable = tableEnv.fromDataStream(dataStream, 
                  'timestamp as 'ts, 'id as 'myId, 'temperature)

val sensorTable = tableEnv.fromDataStream(dataStream, 'myId, 'ts)

tableEnv.createTemporaryView("sensorView", dataStream)
tableEnv.createTemporaryView("sensorView", 
        dataStream, 'id, 'temperature, 'timestamp as 'ts)

tableEnv.createTemporaryView("sensorView", sensorTable)

tableEnv.connect(...)     
        .createTemporaryTable("outputTable")
val resultSqlTable: Table = ...
resultTable.insertInto("outputTable")

tableEnv.connect(new FileSystem().path("output.txt")) // 定义到文件系统的连接  
.withFormat(new Csv()) 
  .withSchema(new Schema()      
  .field("id", DataTypes.STRING())      .
  field("temp", DataTypes.Double())
  .createTemporaryTable("outputTable")     // 创建临时表
  resultTable.insertInto("outputTable")    // 输出表

tableEnv.connect(
  new Kafka()    
      .version("0.11")    
      .topic("sinkTest")    
      .property("zookeeper.connect", "localhost:2181")    
      .property("bootstrap.servers", "localhost:9092"))  
      .withFormat( new Csv() )  
      .withSchema( new Schema()    
      .field("id", DataTypes.STRING())    
      .field("temp", DataTypes.DOUBLE())  
      .createTemporaryTable("kafkaOutputTable")
resultTable.insertInto("kafkaOutputTable")

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")

val sinkDDL: String =  
                |create table jdbcOutputTable (   
                |  id varchar(20) not null,   
                |  cnt bigint not null   
                |) with (   
                |  'connector.type' = 'jdbc',  
                |  'connector.url' = 'jdbc:mysql://localhost:3306/test',   
                |  'connector.table' = 'sensor_count',   
                |  'connector.driver' = 'com.mysql.jdbc.Driver',   
                |  'connector.username' = 'root',    
                |  'connector.password' = '123456'    
                """.stripMargin
tableEnv.sqlUpdate(sinkDDL)    // 执行 DDL创建表
aggResultSqlTable.insertInto("jdbcOutputTable")

val resultStream: DataStream[Row] = tableEnv.toAppendStream[Row](resultTable)

val aggResultStream: DataStream[(Boolean, (String, Long))] = tableEnv
       .toRetractStream[(String, Long)](aggResultTable)

val explaination: String = tableEnv.explain(resultTable)
println(explaination)

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

.withSchema(new Schema() 
.field("id", DataTypes.STRING()) 
.field("timestamp", DataTypes.BIGINT()) 
.field("temperature", DataTypes.DOUBLE()) 
.field("pt", DataTypes.TIMESTAMP(3)) 
.proctime()
)

val sinkDDL: String = 
                    |create table dataTable (    
                    |  id varchar(20) not null,
                    |  ts bigint,    
                    |  temperature double,    
                    |  pt AS PROCTIME()   
                    |) with (  
                    |  'connector.type' = 'filesystem',   
                    |  'connector.path' = '/sensor.txt',  
                    |  'format.type' = 'csv'   
                """.stripMargin
tableEnv.sqlUpdate(sinkDDL)

// 将 DataStream转换为 Table，并指定时间字段
val sensorTable = tableEnv.fromDataStream(dataStream, 
                       'id, 'timestamp.rowtime, 'temperature)
// 或者，直接追加时间字段
val sensorTable = tableEnv.fromDataStream(dataStream, 
                       'id, 'temperature, 'timestamp, 'rt.rowtime)

.withSchema(new Schema() 
.field("id", DataTypes.STRING())
.field("timestamp", DataTypes.BIGINT())  
.rowtime(   
    new Rowtime()      
        .timestampsFromField("timestamp")    // 从字段中提取时间戳     
        .watermarksPeriodicBounded(1000)    // watermark延迟1秒  
.field("temperature", DataTypes.DOUBLE())
)

val sinkDDL: String =  
                |create table dataTable (    
                |  id varchar(20) not null,   
                |  ts bigint,   
                |  temperature double,   
                |  rt AS TO_TIMESTAMP( FROM_UNIXTIME(ts) ),    
                |  watermark for rt as rt - interval '1' second    
                |) with (    
                |  'connector.type' = 'filesystem',   
                |  'connector.path' = '/sensor.txt',   
                |  'format.type' = 'csv'    
                """.stripMargin
tableEnv.sqlUpdate(sinkDDL)

val table = input  
            .window([w: GroupWindow] as 'w) // 定义窗口，别名为 w  
            .groupBy('w, 'a)      // 按照字段 a和窗口 w分组  
            .select('a, 'b.sum)  // 聚合

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

// 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 (类似于计数窗口，按处理时间排序，10行一组)
.window(Tumble over 10.rows on 'proctime as 'w)

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

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

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

// 无界的事件时间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 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)

  TUMBLE(time_attr, interval)

  HOP(time_attr, interval, interval)

  SESSION(time_attr, interval)

TUMBLE_START(time_attr, interval)
TUMBLE_END(time_attr, interval)
TUMBLE_ROWTIME(time_attr, interval)
TUMBLE_PROCTIME(time_attr, interval)

SELECT COUNT(amount) OVER (
  PARTITION BY use
  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 use
  ORDER BY proctime
  ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)

SQL:
value1 = value2
value1 > value2
Table API：
ANY1 === ANY2
ANY1 > ANY2

SQL：
boolean1 OR boolean2
boolean IS FALSE
NOT boolean
Table API：
BOOLEAN1 || BOOLEAN2
BOOLEAN.isFalse
!BOOLEAN

SQL：
numeric1 + numeric2
POWER(numeric1, numeric2)
Table API：
NUMERIC1 + NUMERIC2
NUMERIC1.power(NUMERIC2)

SQL：
string1 || string2
UPPER(string)
CHAR_LENGTH(string)
Table API：
STRING1 + STRING2
STRING.upperCase()
STRING.charLength()

SQL：
DATE string
TIMESTAMP string
CURRENT_TIME
INTERVAL string range
Table API：
STRING.toDate
STRING.toTimestamp
currentTime()
NUMERIC.days
NUMERIC.minutes

SQL：
COUNT(*)
SUM([ ALL | DISTINCT ] expression)
RANK()
ROW_NUMBER()
Table API：
FIELD.count
FIELD.sum0

// 自定义一个标量函数
class HashCode( factor: Int ) extends ScalarFunction {
  def eval( s: String ): Int = {
    s.hashCode * facto
}

def main(args: Array[String]): Unit = {
  val env = StreamExecutionEnvironment.getExecutionEnvironment
  env.setParallelism(1)
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  val settings = EnvironmentSettings
    .newInstance()
    .useOldPlanner()
    .inStreamingMode()
    .build()
  val tableEnv = StreamTableEnvironment.create( env, settings )
  // 定义好 DataStream
  val inputStream: DataStream[String] = env.readTextFile("..\\sensor.txt")
  val dataStream: DataStream[SensorReading] = inputStream
    .map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
    .assignAscendingTimestamps(_.timestamp * 1000L)
  // 将 DataStream转换为 Table，并指定时间字段
  val sensorTable = tableEnv.fromDataStream(dataStream, 'id, 'timestamp.rowtime, 'temperature)
  // Table API中使用
  val hashCode = new HashCode(10)
  val resultTable = sensorTable
    .select( 'id, hashCode('id) )
  // SQL 中使用
  tableEnv.createTemporaryView("sensor", sensorTable)
  tableEnv.registerFunction("hashCode", hashCode)
  val resultSqlTable = tableEnv.sqlQuery("select id, hashCode(id) from sensor")
  // 转换成流，打印输出
  resultTable.toAppendStream[Row].print("table")
  resultSqlTable.toAppendStream[Row].print("sql")
  env.execute()
}

// 自定义TableFunction
class Split(separator: String) extends TableFunction[(String, Int)]{
  def eval(str: String): Unit = {
    str.split(separator).foreach(
      word => collect((word, word.length))
}

// Table API中调用，需要用joinLateral
    val resultTable = sensorTable
      .joinLateral(split('id) as ('word, 'length))   // as对输出行的字段重命名
      .select('id, 'word, 'length)
// 或者用leftOuterJoinLateral
    val resultTable2 = sensorTable
      .leftOuterJoinLateral(split('id) as ('word, 'length))
      .select('id, 'word, 'length)
    // 转换成流打印输出
    resultTable.toAppendStream[Row].print("1")
    resultTable2.toAppendStream[Row].print("2")

   tableEnv.createTemporaryView("sensor", sensorTable)
    tableEnv.registerFunction("split", split)
    val resultSqlTable = tableEnv.sqlQuery(
        |select id, word, length
        |from
        |sensor, LATERAL TABLE(split(id)) AS newsensor(word, length)
      """.stripMargin)
    // 或者用左连接的方式
    val resultSqlTable2 = tableEnv.sqlQuery(
        |SELECT id, word, length
        |FROM
        |senso
|  LEFT JOIN 
|  LATERAL TABLE(split(id)) AS newsensor(word, length) 
|  ON TRUE
      """.stripMargin
    // 转换成流打印输出
    resultSqlTable.toAppendStream[Row].print("1")
    resultSqlTable2.toAppendStream[Row].print("2")

createAccumulator()
accumulate()
getValue()

retract()
merge()
resetAccumulator()

// 定义AggregateFunction的Accumulato
class AvgTempAcc {
  var sum: Double = 0.0
  var count: Int = 0
class AvgTemp extends AggregateFunction[Double, AvgTempAcc] {
  override def getValue(accumulator: AvgTempAcc): Double =
    accumulator.sum / accumulator.count
  override def createAccumulator(): AvgTempAcc = new AvgTempAcc
  def accumulate(accumulator: AvgTempAcc, temp: Double): Unit ={
    accumulator.sum += temp
    accumulator.count += 1
}

// 创建一个聚合函数实例
val avgTemp = new AvgTemp()
// Table API的调用 
val resultTable = sensorTable.groupBy('id)
  .aggregate(avgTemp('temperature) as 'avgTemp)
  .select('id, 'avgTemp)
// SQL的实现
tableEnv.createTemporaryView("sensor", sensorTable)
tableEnv.registerFunction("avgTemp", avgTemp)
val resultSqlTable = tableEnv.sqlQuery(
    |SELECT
    |id, avgTemp(temperature)
    |FROM
    |senso
    |GROUP BY id
  """.stripMargin)
// 转换成流打印输出
resultTable.toRetractStream[(String, Double)].print("agg temp")
resultSqlTable.toRetractStream[Row].print("agg temp sql")

createAccumulator()
accumulate()

retract()
merge()
resetAccumulator()
emitValue()
emitUpdateWithRetract()

// 先定义一个 Accumulator 
class Top2TempAcc{
  var highestTemp: Double = Int.MinValue
  var secondHighestTemp: Double = Int.MinValue
// 自定义 TableAggregateFunction
class Top2Temp extends TableAggregateFunction[(Double, Int), Top2TempAcc]{
  override def createAccumulator(): Top2TempAcc = new Top2TempAcc
  def accumulate(acc: Top2TempAcc, temp: Double): Unit ={
    if( temp > acc.highestTemp ){
      acc.secondHighestTemp = acc.highestTemp
      acc.highestTemp = temp
    } else if( temp > acc.secondHighestTemp ){
      acc.secondHighestTemp = temp
  def emitValue(acc: Top2TempAcc, out: Collector[(Double, Int)]): Unit ={
    out.collect(acc.highestTemp, 1)
    out.collect(acc.secondHighestTemp, 2)
}

// 创建一个表聚合函数实例
val top2Temp = new Top2Temp()
// Table API的调用
val resultTable = sensorTable.groupBy('id)

Flink学习笔记(9)-Table API 和 Flink SQL

Flink学习笔记(9)-Table API 和 Flink SQL

一、Table API 和 Flink SQL 是什么？

二、基本程序结构

三、 创建 TableEnvironment

3.1 配置老版本 planner 的流式查询TableEnvironment

3.2 配置老版本 planner 的批式查询TableEnvironment

3.3 配置 blink planner 的流式查询TableEnvironment

3.4 配置 blink planner 的批式查询TableEnvironment

四、表（Table）

4.1 创建表

4.1.1 连接到外部文件系统（CSV）

4.1.2 连接到Kafka

4.2 表的查询

4.3 表的查询-SQL

4.4 将 DataStream 转换成表

4.5 创建临时视图（Temporary View）

4.5.1 基于 DataStream 创建临时视图

4.5.2 基于 Table 创建临时视图

4.6 输出表

4.6.1 输出到文件

4.6.2 输出到 Kafka

4.6.3 输出到ES

4.6.4 输出到 MySql

4.7 将 Table 转换成 DataStream

五、查看执行计划

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

七、动态表（Dynamic Tables）

八、时间特性

8.1 定义处理时间（Processing Time）

8.2 定义事件时间（Event Time）

九、窗口

9.1 Group Windows

9.1.1 滚动窗口

9.1.2 滑动窗口

9.1.3 会话窗口

9.2 Over Windows

9.3 SQL中窗口的定义

9.3.1 Group Windows

9.3.2 Over Windows

十、函数

10.1 系统内置函数

10.2 UDF

10.2.1 标量函数（Scalar Functions）

10.2.2 表函数（Table Functions）

10.2.3 聚合函数（Aggregate Functions）

10.2.4 表聚合函数（Table Aggregate Functions）