Flink Table API与SQL

11、Table API与SQL

• Flink 对批处理和流处理，提供了统一的上层 API
• Table API 是一套内嵌在 Java 和 Scala 语言中的查询API，它允许以非常直观的方式组合来自一些关系运算符的查询
• Flink 的 SQL 支持基于实现了 SQL 标准的 Apache Calcite

11.1 需要引入的 pom 依赖

<dependency>
    <groupId>org.apache.flink</groupId> 
    <artifactId>flink-table-planner_2.12</artifactId> 
    <version>1.9.3</version> 
</dependency> 
<dependency> 
    <groupId>org.apache.flink</groupId> 
    <artifactId>flink-table-api-scala-bridge_2.12</artifactId> 
    <version>1.9.3</version> 
</dependency>
1
2
3
4
5
6
7
8
9
10

11.2、两种planner区别（old/blink）

• 在 Flink 1.9 中会存在两个 Planner：Flink Planner 和 Blink Planner。

1. 批流统一：Blink将批处理作业，视为流式处理的特殊情况。所以，blink不支持表和DataSet之间的转换，批处理作业将不转换为DataSet应用程序，而是跟流处理一样，转换为DataStream程序来处理。
2. 因为批流统一，Blink planner也不支持BatchTableSource，而使用有界的StreamTableSource代替。
3. Blink planner只支持全新的目录，不支持已弃用的ExternalCatalog。
4. 旧planner和Blink planner的FilterableTableSource实现不兼容。旧的planner会把PlannerExpressions下推到filterableTableSource中，而blink planner则会把Expressions下推。
5. 基于字符串的键值配置选项仅适用于Blink planner。
6. PlannerConfig在两个planner中的实现不同。
7. Blink planner会将多个sink优化在一个DAG中（仅在TableEnvironment上受支持，而在StreamTableEnvironment上不受支持）。而旧planner的优化总是将每一个sink放在一个新的DAG中，其中所有DAG彼此独立。
8. 旧的planner不支持目录统计，而Blink planner支持。

• 两个Planner：flink-table-planner 和 flink-table-planner-blink。
• 两个Bridge：flink-table-api-scala-bridge 和 flink-table-api-java-bridge，从图中可以看出，Flink Planner 和 Blink Planner 都会依赖于具体的 JAVA API，也会依赖于具体的Bridge，通过 Bridge 可以将 API 操作相应的转化为 Scala 的 DataStream、DataSet，或者转化为 JAVA 的 DataStream 或者 Data Set。

11.3、基本程序结构

def main(args: Array[String]): Unit = {
    //创建flink环境对象
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //获取数据
    val inputStream = env.readTextFile("..\\sensor.txt")
    //map为javabean对象
    val dataStream = inputStream .map( data => { 
        val dataArray = data.split(",") 
        SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble) 
    }
    // 基于 env 创建 tableEnv
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
                                      .useOldPlanner()
                                      .inStreamingMode()
                                      .build() 
    //创建表的环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)
//方式一：
    //从一条流创建一张表 
	val fTable: Table = tableEnv.fromDataStream(dataStream)
	// 从表里选取特定的数据 
    val selectedTable: Table = fTable.select('id, 'temperature) .filter("id = 'sensor_1'")
	//转化为DataStream
    val selectedStream: DataStream[(String, Double)] = selectedTable .toAppendStream[(String, Double)] 
    //打印输出
    selectedStream.print() 

//方式 二
	//注册一张表
	val rTable: Table = tableEnv.registerDataStream("t_SensorReading",dataStream)
	//sql
	var senTable:Table = rTable.sql("select * from t_SensorReading")
	//转化为DataStream
	val senStream: DataStream[(String, Double)] = senTable.toRetractStream(Row.class)
	//输出
	senStream.print()
	
    env.execute("table test")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

11.4. TableEnvironment

• TableEnvironment 组成部分如下：

  • flink-table-common：这个包中主要是包含 Flink Planner 和 Blink Planner 一些共用的代码。
  • flink-table-api-java：这部分是用户编程使用的 API，包含了大部分的 API。
  • flink-table-api-scala：这里只是非常薄的一层，仅和 Table API 的 Expression 和 DSL 相关。

• Flink 1.9 中保留了 5 个 TableEnvironment，在实现上是 5 个面向用户的接口，在接口底层进行了不同的实现。

  5 个接口包括一个 TableEnvironment 接口，两个 BatchTableEnvironment 接口，两个 StreamTableEnvironment 接口

  • org/apache/flink/table/api/TableEnvironment.java

  • org/apache/flink/table/api/java/BatchTableEnvironment.java

  • org/apache/flink/table/api/scala/BatchTableEnvironment.scala

  • org/apache/flink/table/api/java/StreamTableEnvironment.java

  • org/apache/flink/table/api/scala/StreamTableEnvironment.scala

• 解析的过程涉及到了三层：Table API/SQL，Blink Planner，Runtime
• 用于统一流和批处理
  • Table API是Scala和Java语言集成查询API，可以非常直观的方式组合来自关系算子的查询(e.g. 选择，过滤和连接).
  • Flink的SQL支持基于实现SQL标准的Apache Calcite。无论输入是批输入（DataSet）还是流输入（DataStream），任一接口中指定的查询都具有相同的语义并指定相同的结果。

11.5、表(Table)

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

11.5.1 创建 TableEnvironment

TableEnvironment 是 Table API 和 SQL 的核心概念。它负责:

• 在内部的 catalog 中注册 Table
• 注册外部的 catalog
• 加载可插拔模块
• 执行 SQL 查询
• 注册自定义函数 （scalar、table 或 aggregation）
• 将 DataStream 或 DataSet 转换成 Table
• 持有对 ExecutionEnvironment 或 StreamExecutionEnvironment 的引用

通过静态方法 BatchTableEnvironment.create() 或者 StreamTableEnvironment.create() 中创建。

11.5.2 动态表

如果流中的数据类型是 case class 可以直接根据 case class 的结构生成 table

tableEnv.fromDataStream(dataStream) 
1

或者根据字段顺序单独命名

tableEnv.fromDataStream(dataStream,’id,’timestamp .......) 
1

最后的动态表可以转换为流进行输出

table.toAppendStream[(String,String)] 
1

11.5.2 注册表

如果流中的数据类型是 case class 可以直接根据 case class 的结构生成 table

tableEnv.registerDataStrem("t_table",dataStream);
1

最后转换为留输出

tableEnv.toRetractStream(table,Row.class);
1

11.5.4 创建视图

//创建视图
tableEnv.createTemporaryView("t_table",dataStream);
//转换成流
tableEnv.toAppendStream(table, Row.class).print();
1
2
3
4

11.5.4 创建零时表

//创建零时表
tableEnv.createTemporaryTable("inputTable",dataStream);
//转换成流
tableEnv.toAppendStream(inputTable, Row.class).print();
1
2
3
4

11.6. TableAPI语法

11.6.1. Scan, Projection, and Filter

• Scan
• select
• as
• where/Fliter

11.6.2. Column Operations

• addColumns
• AddOrReplaceColumns
• DropColumns
• Renamecolumns

11.6.3. Aggregations

• groupBy Aggregation
• groupBy window aggregation
• over window aggregation Streaming
• distinct aggregation
• distinct

11.6.4. Joins

• inner join
• outer join
• time-windowed join
• joinLateral
• leftouterjoinLateral

11.6.5. Set Operations

• union
• unionAll
• intersect
• intersectAll
• minus
• minusAll
• in

11.6.6. OrderBy, Offset & Fetch

• order by
• offset
• fetch

11.6.7. Insert

• insert into

11.6.8. Window

• Table 可以提供一个逻辑上的时间字段，用于在表处理程序中，指示时间和访问相应的时间戳。

  • 时间属性，可以是每个表 schema 的一部分。一旦定义了时间属性，它就可以作为一个字段引用，并且可以在基于时间的操作中使用。

• Event Time事件时间

  • 为了处理无序事件，并区分流中的准时和迟到事件；Flink 需要从事件数据中，提取时间戳，并用来推进事件时间的进展（watermark）。

  • 在DataStream 转换成 Table，schema 的定义期间，使用.rowtime 可以定义事件时间属性。注意，必须在转换的数据流中分配时间戳和 watermark。

  • 在将数据流转换为表时，有两种定义时间属性的方法。根据指定的.rowtime 字段名是否存在于数据流的架构中，timestamp 字段可以：

    ​ 1、作为新字段追加到 schema

    ​ 2、替换现有字段

11.7. FlinkSQL 语法

• Flink SQL 核心算子的语义设计参考了 1992、2011 等 ANSI-SQL 标准，Flink 使用 Apache Calcite解析 SQL ，Calcite 支持标准的 ANSI SQL。

11.7.1. FlinkSQL常用算子

• SELECT 用于从 DataSet/DataStream 中选择数据，用于筛选出某些列。

• WHERE 用于从数据集/流中过滤数据，与 SELECT 一起使用，用于根据某些条件对关系做水平分割，即选择符合条件的记录。

• DISTINCT 用于从数据集/流中去重根据 SELECT 的结果进行去重。

• GROUP BY 是对数据进行分组操作。

• UNION 用于将两个结果集合并起来，要求两个结果集字段完全一致，包括字段类型、字段顺序。不同于 UNION ALL 的是，UNION 会对结果数据去重。

• JOIN 用于把来自两个表的数据联合起来形成结果表，Flink 支持的 JOIN 类型包括：

  • JOIN - INNER JOIN

  • LEFT JOIN - LEFT OUTER JOIN

  • RIGHT JOIN - RIGHT OUTER JOIN

  • FULL JOIN - FULL OUTER JOIN

11.7.2. Group Window

• 根据窗口数据划分的不同，目前 Apache Flink 有如下 3 种 Bounded Window：
  • Tumble，滚动窗口，窗口数据有固定的大小，窗口数据无叠加；
  • Hop，滑动窗口，窗口数据有固定大小，并且有固定的窗口重建频率，窗口数据有叠加；
  • Session，会话窗口，窗口数据没有固定的大小，根据窗口数据活跃程度划分窗口，窗口数据无叠加。

11.7.3. Flink的内置函数

• 比较函数
• 逻辑函数
• 算术函数
• 字符串函数
• 时间函数