[大数据] Flink Table API与SQL

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Flink Table API与SQL -> 正文阅读

[大数据]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>

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

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

在这里插入图片描述

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

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)

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

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

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

table.toAppendStream[(String,String)]

11.5.2 注册表

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

tableEnv.registerDataStrem("t_table",dataStream);

最后转换为留输出

tableEnv.toRetractStream(table,Row.class);

11.5.4 创建视图

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

11.5.4 创建零时表

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

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，会话窗口，窗口数据没有固定的大小，根据窗口数据活跃程度划分窗口，窗口数据无叠加。