整体介绍

什么是Table API和Flink SQL

Flink本身是批流统一的处理框架，所以Table API和SQL，就是批流统一的上层处理API。目前处于活跃的开发阶段。
Table API 是一套内嵌在Java和Scala语言中的查询API，它允许我们以非常直观的方式，组合来自一些原酸关系符的查询（比如select、filter和join）。而对于Flink SQL，就是直接可以再代码中写SQL，来实现一些查询（Query）操作。Flink的SQL支持，基于实现SQL标准的Apache Calcite（Apache开源SQL解析工具）。
无论输入是批输入还是流式输入，在这两套API中，指定的查询都具有相同的语义，得到相同的结果。

需要引入的依赖

        <!-- flink-table-planner:
    planner计划器，是table API最主要的部分，提供了运行时环境和生成程序执行计划的planner. 
-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <!-- flink-table-planner-blink -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <!-- flink-table-api-java-bridge:
    bridge桥接器，主要负责table API和DataStream/DataSet API的连接支持，按照语言分为java和scala 
-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <!-- flink-csv: -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-csv</artifactId>
            <version>1.10.1</version>
        </dependency>
        <!-- flink-json -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>1.10.1</version>
        </dependency>
        <!-- flink-jdbc -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-jdbc_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>

两种planner(old&blink)的区别

批流统一：Blink将批处理作业，视为流式处理的特殊情况。所以，blink不支持表和DataSet之间的转换，批处理作业将不转换为DataSet应用程序，而是跟流处理一样，转换为DataStream程序来处理。
因为批流统一，Blink planner也不支持BatchTableSource，而使用有界的StreamTableSource代替。
Blink planner只支持全新的目录，不支持已弃用的exteralCatalog。
旧planner和blink planner的filterableTableSource实现不兼容。旧的planner会把PlannerExpressions下推到filterableTableSource中，而blink planner则会把Expression下推。
PlannerConfig在两个planner中的实现不同。
Blink planner会将多个sink优化在一个DAG中（仅在TableEnvironment上受支持，而在StreamTableEnvironment上不受支持）。而旧planner的优化总是将每一个sink放在一个新的DAG中，其中所有DAG彼此独立。
旧的planner不支持目录统计，而Blink planner支持。

API调用

基本程序结构

Table API和SQL的程序结构，与流式处理的程序结构类似；也可以近似地认为有这么几步：首先创建执行环境，然后定义source、transform和sink。

StreamTableEnvironment tableEnv = ... // 创建表的执行环境

// 创建一张表，用于读取数据
tableEnv.connect(...).createTemporaryTable("inputTable");
// 注册一张表，用于把计算结果输出
tableEnv.connect(...).createTemporaryTable("outputTable");

// 通过Table API查询算子，得到一张结果表
Table result = tableEnv.from("inputTable").select(...);
// 通过SQL查询语句，得到一张结果表
Table sqlResult = tableEnv.sqlQuery("select ... from inputTable ...");

// 将结果表写入输出表中
result.insertInto("outputTable");

创建表环境

创建表环境最简单的方式，就是基于流处理执行环境，调用 create 方法直接创建：
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

表环境（TableEnvironment）是Flink中集成Table API & SQL的核心概念。它负责：

注册catalog（目录）
在内部catalog中注册表
执行SQL查询
注册用户自定义函数
将DataStream或DataSet转换为表
保存对 ExecutionEnvironment StreamExecutionEnvironment 的引用。

在创建TableEnv的时候，可以多传入一个 EnvironmentSettings 或者 TableConfig 参数，可以用来配置 TableEnvironment 的一些特征。

// 1. 创建流式环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 1.1 基于老版本planner的流处理(Flink-Streaming-Query)
EnvironmentSettings oldStreamSettings = EnvironmentSettings.newInstance()
        .useOldPlanner()
        .inStreamingMode()
        .build();
StreamTableEnvironment oldStreamTableEnv = StreamTableEnvironment.create(env, oldStreamSettings);

// 1.2 基于老版本planner的批处理(Flink-Batch-Query)
ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment oldBatchTableEnv = BatchTableEnvironment.create(batchEnv);


// 1.3 基于Blink的流处理(Blink-Streaming-Query)
EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
        .useBlinkPlanner()
        .inStreamingMode()
        .build();
StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSettings);

// 1.4 基于Blink的批处理(Blink-Batch-Query)
EnvironmentSettings blinkBatchSettings = EnvironmentSettings.newInstance()
        .useBlinkPlanner()
        .inBatchMode()
        .build();
TableEnvironment blinkBatchTableEnv = TableEnvironment.create(blinkBatchSettings);

在Catalog中注册表

表（Table）

TableEnvironment 可以注册目录 Catalog，并可以基于Catalog注册表。它会维护一个 Catalog-Table 表之间的map。
表（Table）是由一个“标识符”来指定的，由3部分组成：Catalog名，数据库（database）名和对象名（表名）。如果没有指定目录或数据库，就使用当前的默认值。
表可以是常规的（Table，表），或者虚拟的（View，视图）。常规表（Table）一般可以用来描述外部数据，比如文件、数据库表或消息队列的数据，也可以直接从DataStream转换而来。视图可以从现有的表中创建，通常是Table API或者SQL查询的一个结果。

连接到文件系统（Csv格式）

连接外部系统在Catalog中注册表，直接调用 tableEnv.connect() 就可以，里面参数要传入一个 ConnectorDescriptor，也就是connector描述器。对于文件系统的 connector而言，flink内部已经提供了，就叫做FileSystem()。

新的描述其就叫Csv()，符合 RFC-4180 标准的新format描述器，需要引入flink-csv

/*
<!-- flink-csv -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-csv</artifactId>
    <version>1.10.1</version>
</dependency>
 */
tableEnv.connect(new FileSystem().path(filePath))
        .withFormat(new Csv())
        .withSchema(new Schema()
                .field("id", DataTypes.STRING())
                .field("timestamp", DataTypes.BIGINT())
                .field("temp", DataTypes.DOUBLE())
        ) // 定义表结构
        .createTemporaryTable("inputTable");

连接到Kafka

Kafka的连接器flink-kafka-connector中，1.10版本的已经提供了Table API的支持。我们可以再connect方法中直接传入一个叫做Kafka的类，这就是Kafka连接器的描述器ConnectorDescriptor。

package com.zh.apitest.tableapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

/**
 * author: zhanghui
 * project: big-data-learning
 * package: com.zh.apitest.tableapi
 * filename: TableTest4_KafkaPipeLine
 * date: 2021/12/26 16:24
 * description: tableAPI kafka pipeline
 */
public class TableTest4_KafkaPipeLine {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2. 连接kafka，读取数据
        tableEnv.connect(new Kafka()
                    .version("0.11") // 定义Kafka的版本
                    .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("temp", DataTypes.DOUBLE())
                ).createTemporaryTable("inputTable");

        // 3. 查询转换
        Table sensorTable = tableEnv.from("inputTable");

        // 简单转换
        Table resultTable = sensorTable
                .select("id, temp")
                .filter("id === 'sensor_6'");

        // 统计聚合
        Table aggTable = sensorTable
                .groupBy("id")
                .select("id, id.count as count, temp.avg as avgTemp");

        // 4. 连接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("temp", DataTypes.DOUBLE())
                ).createTemporaryTable("outputTable");

        resultTable.insertInto("outputTable");

        env.execute();
    }
}

<br />当然也可以连接到`Elasticsearch、MySql、HBase、Hive`等外部系统，实现方式基本上是类似的。

表的查询

利用外部系统的连接器Connector，我们可以读写数据，并在环境的Catalog中注册表。
Flink给我们提供了两种查询方式：Table API 和 SQL。

Table API调用

Table API是集成在Scala和Java语言内的查询API。与SQL不同，Table API的查询不会用字符串表示，而是在宿主语言中一步步调用完成的。
Table API基于代表一张“表”的Table类，并提供了一整套操作处理的方法API。这些方法会返回一个新的Table对象，这个对象就表示对输入表应用转换操作的结果。有些关系型转换操作，可以由多个方法调用组成，构成链式调用结构。
例如：table.select(...).filter(...)，其中select(...)表示选择表中指定的字段，filter(...)表示筛选条件。

Table inputTable = tableEnv.from("inputTable");
Table resultTable = inputTable
        .select("id, temp")
        .filter("id = 'sensor_6'");

SQL查询

Flink的SQL继承，基于的是Apache Calcite，它实现了SQL标准，在Flink中，用常规字符串来定义SQL查询语句。SQL查询的结果，是一个新的Table。

// inputTable为一个表
Table resultSqlTable = tableEnv.sqlQuery("select id, temperature from inputTable where id = 'sensor_1' ");

// 聚合统计
// TableAPI实现
Table aggTable = inputTable
        .groupBy("id")
        .select("id, id.count as count, temp.avg as avgTemp");

// SQL实现
tableEnv.sqlQuery("select id, temp from inputTable where id = 'sensor_6'");
Table sqlAggTable = tableEnv.sqlQuery("select id, count(id) as cnt, avg(temp) as avgTemp from inputTable group by id");

这里Table API里指定的字段，前面加了一个单引号'，这是Table API中定义的Expression类型的写法，可以很方便的标识一个表中的字段。
字段可以直接全部用双引号引起来，也可以用半边单引号 + 字段名的方式。以后的代码中，一般都用后一种形式。

将DataStream转换成表

Flink允许我们把Table和DataStream做转换：我们可以基于一个DataStream，先流式地读取数据源，然后map成POJO，再把它转成Table。Table的列字段（column fields），就是POJO里的字段，这样就不用再麻烦地定义schema了。

代码表达

直接使用tableEnv.fromDataStream()就可以了。默认转换后的Table Schema和DataStream中的字段一一对应，也可以单独指定。
允许更换字段的顺序、重命名，或者只选取某些字段出来，相当于做了一个map操作（或者Table API的select操作）

DataStream<String> inputStream = env.readTextFile("sensor.txt");

DataStream<SensorReading> dataStream = inputStream.map(line -> {
    String[] fields = line.split(",");
    return new SensorReading(fields[0], Long.valueOf(fields[1]), Double.valueOf(fields[2]));
    });

Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp.rowtime as ts, temperature");

数据类型与Table Schema

DataStream中的数据类型，与表的Schema之间的对应关系，是按照表中的字段名来对应的（name-based mapping），所以还可以用as做重命名。

Table sensorTable = tableEnv.fromDataStream(dataStream, "timestamp as ts, id as myId, temperature");

Flink的DataStream和DataSet API支持多种类型。组合类型，比如元组（内置Scala和java元组）、POJO、Scala case类和Flink的Row类型等，允许具有多个类型的嵌套数据结构，这些字段可以在Table的表达式中访问。其他类型，则被视为原子类。

创建临时视图（Temporary View）

创建临时视图的一种方式，直接从DataStream转换而来。可以直接对应字段转换；也可以在转换的时候，指定响应的字段。

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

// 还可以直接基于Table创建视图
tableEnv.createTemporaryView("sensorView", sensorTable);

View和Table的Schema完全相同。在Table API中，可以认为View和Table是等价的。

输出表

表的输出，是通过将数据写入TableSink来实现的。TableSink是一个通用接口，可以支持不同的文件格式、存储数据库和消息队列。

具体实现，输出表最直接的方法，就是通过Table.insertInto()方法将一个Table写入注册过的TableSink中。

输出到文件

// 注册输出表
tableEnv.connect(
    new FileSystem().path("flink-FlinkTutorial/src/main/resources/sensor.txt")
) // 定义到文件系统的连接
    .withFormat() // 定义格式化方法，Csv格式
    .withSchema(new Schema()
               .field("id", DataTypes.STRING())
               .field("temp", DataTypes.DOUBLE())
               ) // 定义表结构
    .createTemporaryTable("outputTable"); // 创建临时表

resultSqlTable.insertInto("outputTable");

更新模式（Update Mode）

在流处理过程中，表的处理并不像传统定义的那样简单。
对于流式查询（Streaming Queries），需要声明如何在（动态）表和外部连接器之间执行转换。与外部系统交换的消息类型，由更新模式（update mode）指定。
Flink Table API中的更新模式有以下三种：

1）追加模式（Append Mode）

在追加模式下，表（动态表）和外部连接器只交换插入（Insert）消息。

2）撤回模式（Retract Mode）

在撤回模式下，表和外部连接器交换的是：添加（Add）和撤回（Retract）消息。

插入（Insert）会被编码为添加消息；
删除（Delete）则编码为撤回消息；
更新（Update）则会编码为，已更新行（上一行）的撤回消息，和更新行（新行）的添加消息。

在此模式下，不能定义key，这一点跟upsert模式完全不同。

3）Upsert（更新插入）模式

在Upsert模式下，动态表和外部连接器交换Upsert和Delete消息。
这个模式需要一个唯一key，通过这个key可以传递更新消息。为了正确应用消息，外部连接器需要知道这个唯一key的属性。

插入（Insert）和更新（Update）都被编码为Upsert消息；
删除（Delete）编码为Delete消息。

这种模式和Retract模式的主要区别在于，Update操作是用单个消息编码的，所以效率会更高。

输出到Kafka

除了输出到文件，也可以输出到Kafka。我们可以结合前面Kafka作为输入数据，构建数据管道，Kafka进，Kafka出。

package com.zh.apitest.tableapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

/**
 * author: zhanghui
 * project: big-data-learning
 * package: com.zh.apitest.tableapi
 * filename: TableTest4_KafkaPipeLine
 * date: 2021/12/26 16:24
 * description: tableAPI kafka pipeline
 */
public class TableTest4_KafkaPipeLine {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 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("temp", DataTypes.DOUBLE())
                ).createTemporaryTable("inputTable");

        // 3. 查询转换
        Table sensorTable = tableEnv.from("inputTable");

        // 简单转换
        Table resultTable = sensorTable
                .select("id, temp")
                .filter("id === 'sensor_6'");

        // 统计聚合
        Table aggTable = sensorTable
                .groupBy("id")
                .select("id, id.count as count, temp.avg as avgTemp");

        // 4. 连接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("temp", DataTypes.DOUBLE())
                ).createTemporaryTable("outputTable");

        resultTable.insertInto("outputTable");

        env.execute();
    }
}

输出到ElasticSearch

ElasticSearch的Connector可以在Upsert（update+ insert，更新插入）模式下操作，这样就可以使用Query定义的键（key）与外部系统交换upsert/delete消息。
另外，对于“仅追加”（append-only）的查询，connector还可以在append模式下操作，这样就可以与外部系统只交换insert消息。

es目前支持的数据格式，只有json，而Flink本身并没有对应的支持

// 引入JSON依赖
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-json</artifactId>
    <version>1.10.1</version>
</dependency>

// 输出到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");

输出到MySql

Flink专门为Table API的JDBC连接提供了flink-jdbc连接器，我们需要先引入flink-jdbc依赖
jdbc连接的代码实现比较特殊，因为没有对应的java/scala类实现ConnectorDescriptor，所以不能直接tableEnv.connect()。不过Flink SQL留下了执行DDL的接口：tableEnv.sqlUpdate()。

// flink-jdbc
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-jdbc_2.12</artifactId>
    <version>1.10.1</version>
</dependency>

// 输出到Mysql
String sinkDDL = "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'";

tableEnv.sqlUpdate(sinkDDL); // 执行DDL创建表
aggResultSqlTable.insertInto("jdbcOutputTable");

将表转换成DataStream

表可以转换成DataStream或DataSet。这样，自定义流处理或批处理程序就可以继续在Table API或SQL查询的结果上运行了。
将表转换为DataStream或DataSet时，需要指定生成的数据类型，即要将表的每一行转换成的数据类型。通常，最方便的转换类型就是Row。当然，因为结果的所有字段类型都是明确的，我们也经常会用元组类型来表示。

表作为流式查询的结果，是动态更新的。所以，将这种动态查询转换成的数据流，同样需要对表的更新操作进行编码，进而有不同的转换模式。

TableAPI中表到DataStream有两种模式：

追加模式（Append Mode）：用于表只会被插入（Insert）操作更改的场景。
撤回模式（Retract Mode）：用于任何场景。有些类似于更新模式中Retract模式，它只有Insert和Delete两类操作。得到的数据会增加一个Boolean类型的标识位（返回的第一个字段），用他来表示到底是新增的数据（Insert），还是被删除的数据（老数据，Delete）

DataStream<Row> resultStream = tableEnv.toAppendStream(resultTable, Row.class);
DataStream<Tuple2<Boolean, Row>> aggResultStream = tableEnv.toRetractStream(aggResultTable, Row.class);

resultStream.print("result");
aggResultStream.print("aggResult");

没有经过groupBy之类的聚合操作，可以直接用toAppendStream来转换；而如果经过了聚合，有更新操作，一般就必须用toRetractStream。

Query的解释和执行

Table API提供了一种机制来解释（Explain）计算表的逻辑和优化查询计划。这是通过TableEnvironment.explain(table)方法或TableEnvironment.explain()方法完成的。
explain方法会返回一个字符串，描述三个计划：

未优化的逻辑查询计划
优化后的逻辑查询计划
实际执行计划

我们可以在代码中查看执行计划：
```
String expaination = tableEnv.explain(resultTable);
System.out.println(explaination);
```
Query的释放和执行过程，老planner和blink大体是一致的，又有所不同。整体来讲，Query都会表示成一个逻辑查询计划，然后分为两步解释：

优化查询计划
解释成DataStream或者DataSet程序

而Blink版本是批流统一的，所以所有的Query，只会被解释成DataStream程序；另外在批处理的环境TableEnvironment下，Blink版本要找到tableEnv.execute()执行调用才开始解释。

流处理中的特殊概念

Table API和SQL，本质上还是基于关系型表的操作方式：而关系型表、关系代数，以及SQL本身，一边是有界的，更适合批处理的场景。这就导致在进行流处理的过程中，理解会稍微复杂一些，需要引入一些特殊概念。

流处理和关系代数（表，及SQL）的区别

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

可以看到，其实关系代数（主要就是指关系型数据库中的表）和SQL，主要就是针对批处理的，这和流处理有天生的隔阂。

动态表（Dynamic Tables）

因为流处理面对的数据，是连续不断的，这和我们熟悉的关系型数据库中保存的“表”完全不同。如果我们把流数据转换成Table，然后执行类似于table的select操作，结果就不是一成不变的，而是随着新的数据的到来，会不停的更新。
随着新数据的到来，不停地在之前的基础上更新结果。得到的表，在Flink Table API概念来面，这就叫做“动态表”（Dynamic Tables）
动态表示Flink对流数据的Table API和SQL支持的核心概念。与表示批处理数据的静态表不同，动态表是随时间变化的。动态表可以像静态的批处理表一样进行查询，查询一个动态表会产生持续查询（Continuous Query）。连续查询永远不会终止，并会生成另一个动态表。查询（Query）会不断更新其结果表，以反映其动态输入表上的更改。

流式持续查询的过程

流式持续查询的过程为：

流被转换为动态表
对动态表计算连续查询，生成新的动态表。
生成的动态表被转换回流。

将流转换为表（Table）

为了处理带有关系查询的流，必须先将其转换为表。
从概念上将，流的每个数据记录，都被解释为对结果表的插入（Insert）修改。因为流式持续不断的，而且之前的输出结果无法改变。本质上，我们其实是从一个、只有插入操作的changelog（更新日志）流，来构建一个表。

比如，我们现在的输入数据，就是用户在网站上的访问行为，数据类型（Schema）如下：

[
    user: VARCHAR, // 用户名
  cTime: TIMESTAMP, // 访问某个URL的时间戳
  url: VARCHAR // 用户访问的URL
]

持续查询（Continuous Query）

持续查询，会在动态表上做计算处理，并作为结果生成新的动态表。与批处理查询不同，连续查询从不终止，并根据输入表的更新更新其结果表。
在任何时间点，连续查询的结果在语义上，等同于在输入表的快照上，以批处理模式执行的统一查询的结果。

这个Query很简单，是一个分组聚合做count统计的查询。它将用户字段上的clicks表分组，并统计访问的URL数。图中显示了随着时间的推移，当clicks表被其他行更新时如何计算查询。

将动态表转换成流

与常规的数据库一样，动态表可以通过插入（Insert）、更新（Update）和删除（Delete）更改，进行持续的修改。将动态表转换为流或将其写入外部系统时，需要对这些更改进行编码。Flink的Table API和SQL支持三种方式对动态表的更改进行编码。

1）仅追加（Append-Only）流

仅通过插入（Insert）更改，来修改的动态表，可以直接转换为“仅追加”流。这个流中发出的数据，就是动态表中新增的每一行。

2）撤回（Retract）流

Retract流是包含两类消息的流，添加（Add）消息和撤回（Retract）消息。
动态表通过将INSERT编码为add消息、DELETE编码为retract消息、UPDATE编码为被更改行（前一行）的retract消息和更新后行（新行）的add消息，转换为retract流。

2）Upsert（更新插入）流

Upsert流包含两种类型的消息：Upsert消息和delete消息。转换为upsert流的动态表，需要由唯一的键（key）。
通过将INSERT和UPDATE更改编码为upsert消息，将DELETE更改编码为DELETE消息，就可以将具有唯一键（Unique Key）的动态表转换为流。

需要注意的是，在代码里将动态表转换为DataStream时，仅支持Append和Retract流。而向外部系统输出动态表的TableSink接口，则可以有不同的实现，比如之前我们讲到的ES，就可以有Upsert模式。

时间特性

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

处理时间（Processing Time）

处理时间语义下，允许表处理程序根据机器生成结果。它是时间的最简单概念。它既不需要提取时间戳，也不需要生成watermark。
定义处理时间属性有三种方法：在DataStream转化时直接定义指定；在定义Table Schema时指定；在创建表的DDL中指定。

1）DataStream转换成Table时指定

由DataStream转换成表时，可以再后面指定字段名来定义Schema。在定义Schema期间，可以使用 .proctime，定义处理时间字段。
注意，这个proctime属性只能通过附加逻辑字段，来扩展物理schema。因此，只能在schema定义的末尾定义它。

// 定义好 DataStream
DataStream<String> inputStream = env.readTextFile("\\sensor.txt");
DataStream<SensorReading> dataStream = inputStream
    .map( line -> {
        String[] fields = line.split(",");
        return new SensorReading(fields[0], Long.valueOf(fields[1]), Double.valueOf(fields[2]));
    });

// 将DataStream转换为Table，并指明时间字段
Table sensorTable = tableEnv.fromDataStream(dataStream, "id, temperature, timestamp, pt.proctime");

2）定义Table Schema时指定

这种方法其实很简单，只要在定义Schema的时候，加上一个新的字段，并制定成proctime就可以了。

tableEnv.connect(
    new FileSystem().path("..\\sensor.txt"))
    .withFormat(new Csv()
                .field("id", DataTypes.STRING())
                .field("timestamp", DataTypes.BIGINT())
                .field("temperature", DataTypes.DOUBLE())
                .fields("pt", DataTypes.TIMESTAMP(3)).proctime()
               ) // 定义表结构
    .createTemporaryTable("inputTable"); // 创建临时表

3）创建表的DDL中指定

在创建表的DDL中，增加一个字段并制定为proctime，也可以指定当前的时间字段。

运行下面这段DDL，必须使用Blink Planner

String sinkDDL = "create table dataTable ("
    + " id varchar(20) not null,"
    + " ts bigint,"
    + " tenmperature double,"
    + " pt as PROCTIME()"
    + " ) with ("
    + " 'connector.type' = 'filesystem',"
    + " 'connector.path' = '/sensor.txt',"
    + " 'format.type' = 'csv')";

tableEnv.sqlUpdate(sinkDDL);

事件时间（Event Time）

事件时间语义，允许表处理程序根据每个记录中包含的时间生成结果。这样即使在有乱序或者延迟事件时，也可以获得正确的结果。
为了处理无序事件，并区分流中的准时和迟到事件；Flink需要从事件数据中提取时间戳，并用来推进事件时间的进展（watermark）。

1）DataStream转换成Table时指定

在DataStream转换成Table，schema的定义期间，使用.rowtime可以定义事件时间属性。注意，必须在转换的数据流中分配时间戳和watermark。
在将数据流转换成表时，有两种定义时间属性的方法。根据指定的.rowtime字段名是否存在于数据流的架构中，timestamp字段可以：

作为新字段追加到schema
替换现有字段

在这两种情况下，定义的事件时间戳字段，都将保存DataStream中事件时间戳的值。

// 1. 读取数据
DataStreamSource<String> inputStream = env.readTextFile("flink-FlinkTutorial/src/main/resources/sensor.txt");

// 2. 转换成POJO
DataStream<SensorReading> dataStream = inputStream.map(line -> {
    String[] fields = line.split(",");
    return new SensorReading(fields[0], Long.valueOf(fields[1]), Double.valueOf(fields[2]));
}).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.minutes(2)) {
    @Override
    public long extractTimestamp(SensorReading element) {
        return element.getTimestamp() * 1000;
    }
});

// 3. 创建表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 4. 将流转换成表，定义时间特性
Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, rt.rowtime");

2）定义Table Schema时指定

这种方法只要在定义Schema的时候，将事件时间字段，并制定成rowtime就可以了。

tableEnv.connect(new FileSystem().path("sensor.txt"))
    .withFormat(new Csv())
    .withSchema(new Schema()
               .field("id", DataTypes.STRING())
               .field("timestamp", DataTypes.BIGINT())
               .rowtime(
                       new Rowtime()
                           .timestampsFromField("timestamp") // 从字段中提取时间戳
                           .watermarksPeriodBounded(1000) // watermark 延迟1秒
               )
               .field("temperature", DataTypes.DOUBLE())
               ) // 定义表结构
    .createTemporaryTable("inputTable"); // 创建临时表

3）创建表的DDL中指定

事件时间属性，是使用create table DDL中的watermark语句定义的。watermark语句，定义现有事件时间字段上的watermark生成表达式，该表达式将事件时间字段标记为事件时间顺序。

String sinkDDL = "create table dataTable ("
    + " id varchar(20) not null,"
    + " ts bigint,"
    + " temperature double,"
    + " rt as TO_TIMESTAMP(FROM_UNIXTIME(ts)),"
    + " watermark for rt as ts - interval '1' second"
    + " ) with ("
    + " 'connector.type' = 'filesystem',"
    + " 'connector.path' = '/sensor.txt',"
    + " 'format.type' = 'csv'";
tableEnv.sqlUpdate(sinkDDL); // 需要使用Blink Planner

这里FROM_UNIXTIME是系统内置的时间函数，用来将一个整数（秒数）转换成“YYYY-MM-DD hh:mm:ss”格式（默认，也可以作为第二个String参数传入）的日期时间字符串（date time string）；然后再用TO_TIMESTAMP将其转换成Timestamp。

窗口（Windows）

时间语义，要配合窗口操作才能发挥作用。最主要的用途，当然就是开窗口、根据时间段计算了。
在Table API 和SQL中，主要由两种窗口：Group Windows 和 Over Windows

分组窗口（Group Windows）

分组窗口（Group Windows）会根据时间或行计数间隔，将行聚合到有限的组（Group）中，并对每个组的数据执行一次聚合函数。
Table API中的Group Windows都是使用.window(w:GroupWindow)子句定义的，并且必须由as子句指定一个别名。为了按窗口对表进行分组，窗口的别名必须在group by子句中，像常规的分组字段一样引用。

Table table = input.window([w:GroupWindow] as "w") // 定义窗口，别名w
    .groupBy("w, a") // 以属性a和窗口w作为分组的key
    .select("a, b.sum") // 聚合字段b的值，求和

或者，还可以把窗口的相关信息，作为字段添加到结果集中：

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

Table API提供了一组具有特定语义的预定义window类，这些类会被转换为底层DataStream或DataSet的窗口操作。
Table API支持的窗口定义，和我们熟悉的一样，主要也是三种：滚动（Tumbling）、滑动（Sliding）和会话（Session）。

滚动窗口

滚动窗口（Tumbling windows）要用Tumble类来定义，另外还有三个方法：

over：定义窗口长度
on：用来分组（按时间间隔）或者排序（按行数）的时间字段
as：别名，必须出现在后面的group by中

// Tumbling Event-time Window
.window(Tumble.over("10.minutes").on("rowtime").as("w"))

// Tumbling Processing-time Window
.window(Tumble.over("10.minutes").on("proctime").as("w"))

// Tumbling Row-count Window
.window(Tumble.over("10.rows").on("proctime").as("w"))

滑动窗口

滑动窗口（Sliding Windows）要用Slide类来定义，另外还有四个方法：

over：定义窗口长度
every：定义滑动步长
on：用来分组（按时间间隔）或者排序（按行数）的时间字段
as：别名，必须出现在后面的groupBy中

// 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 Windows）要用Session类来定义，另外还有三个方法：

withGap：会话时间间隔
on：用来分组（按时间间隔）或者排序（按行数）的时间字段
as：别名，必须出现在后面的groupBy中。

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

Over Windows

Over window 聚合是标准SQL中已有的（Over子句），可以再查询的SELECT子句中定义。Over window聚合，会针对每个输入行，计算相邻范围内的聚合。Over windows 使用.window(w:overwindow*)子句定义，并在select() 方法中通过别名来引用。

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

Table API提供了Over类，来配置Over窗口的属性。可以在事件时间或处理时间，以及指定为时间间隔、或行计数的范围，定义Over windows
无界的over window是使用常量指定的。也就是说，时间间隔要指定UNBOUNDED_RANGE，或者行计数间隔要指定UNBOUNDED_ROW。而有界的Over window是用间隔的大小指定。

1）无界的 over window

// 无界的事件时间 over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding(UNBOUNDED_RANGE).as("w"))

// 无界的处理时间 over window
.window(Over.partitionBy("a").orderBy("proctime").preceding(UNBOUNDED_RANGE)).as("w")

// 无界的事件时间 Row-count over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding(UNBOUNDED_ROW).as("w"))

// 无界的处理时间 Row-count over window
.window(Over.partitionBy("a").orderBy("proctime").preceding(UNBOUNDED_ROW).as("w"))

2）有界的 over window

// 有界的事件时间 over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding("1.minutes").as("w"))

// 有的处理时间 over window
.window(Over.partitionBy("a").orderBy("proctime").preceding("1.minutes")).as("w")

// 有的事件时间 Row-count over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding("10.minutes").as("w"))

// 有的处理时间 Row-count over window
.window(Over.partitionBy("a").orderBy("proctime").preceding("10.minutes").as("w"))

SQL中的窗口定义

Group Windows

Group Windows 在SQL查询的Group BY子句中定义。与使用常规Group By子句的查询一样，使用Group By子句的查询会计算每个字的单个结果行。
SQL支持一下Group窗口函数

TUMBLE(time_attr, interval)：定义了一个滚动窗口，第一个参数是时间字段，第二个参数是窗口长度。
HOP(time_attr, interval, interval)：定义了一个滑动窗口，第一个参数时时间字段，第二个参数时窗口的滑动步长，第三个参数时窗口长度。
SESSION(time_attr, interval)：定义了一个会话窗口，第一个参数是时间字段，第二个参数是窗口间隔（Gap）

另外还有一些辅助函数，可以用来选择Group window的开始和结束时间戳，以及时间属性。
这里只写TUMBLE_*，滑动和会话窗口是类似的（HOP_*，SESSION_*）

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

Over Windows

由于Over本来就是SQL内置支持的语法，所以这在SQL中属于基本的聚合操作。所有聚合必须在同一窗口上定义，也就是说，必须是相同的分区、排序和范围，目前仅支持在当前行范围之前的窗口（无边界和有边界）。
注意：Order By必须在单一的时间属性上指定。

select count(amount) over (
  PARTITION by user
  over by proctime
  rows between 2 preceding and corrent row
) from Orders

// 也可以做多个聚合
select count(amount) over w, sum(amount) over w
from Orders
window w as (
    partition by user
  order by proctime
  rows between 2 preceding and corrent row
)

练习代码

package com.zh.apitest.tableapi;

import com.zh.apitest.beans.SensorReading;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * author: zhanghui
 * project: big-data-learning
 * package: com.zh.apitest.tableapi
 * filename: TableTest5_TimeAndWindow
 * date: 2021/12/26 20:11
 * description: time about and window
 */
public class TableTest5_TimeAndWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 1. 读取数据
        DataStreamSource<String> inputStream = env.readTextFile("flink-FlinkTutorial/src/main/resources/sensor.txt");

        // 2. 转换成POJO
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], Long.valueOf(fields[1]), Double.valueOf(fields[2]));
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.minutes(2)) {
            @Override
            public long extractTimestamp(SensorReading element) {
                return element.getTimestamp() * 1000;
            }
        });

        // 3. 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 4. 将流转换成表，定义时间特性
        //Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, pt.proctime");
        Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, rt.rowtime");
        tableEnv.createTemporaryView("sensor", dataTable);

        // 5. 窗口操作
        // 5.1 Group Window
        // table API
        Table resultTable = dataTable.window(Tumble.over("10.seconds").on("rt").as("tw"))
                .groupBy("id, tw")
                .select("id, id.count, temp.avg, tw.end");

        // SQL
        Table resultSqlTable = tableEnv.sqlQuery("select id, count(id) as cnt, avg(temp) as avgTemp, tumble_end(rt, interval '10' second)" +
                " from sensor group by id, tumble(rt, interval '10' second)");

        // 5.2 Over Window
        // Table API
        Table overResult = dataTable.window(Over.partitionBy("id").orderBy("rt").preceding("2.rows").as("ow"))
                .select("id, rt, id.count over ow, temp.avg over ow");

        // SQL
        Table overSqlResult = tableEnv.sqlQuery("select id, rt, count(id) over ow, avg(temp) over ow" +
                " from sensor" +
                " window ow as (partition by id order by rt rows between 2 preceding and current row)");

        dataTable.printSchema();
        //tableEnv.toAppendStream(resultTable, Row.class).print("result");
        //tableEnv.toAppendStream(resultSqlTable, Row.class).print("resultSql");
        tableEnv.toAppendStream(overResult, Row.class).print("result");
        tableEnv.toRetractStream(overSqlResult, Row.class).print("resultSql");

        env.execute();

    }
}

函数（Functions）

系统内置函数

Flink Table API 和 SQL 为用户提供了一组用于数据转换的内置函数。SQL中支持的很多函数，Table API和SQL都已经做了实现，其他还在快速开发扩展中。

函数	SQL	Table API
比较函数	value1 = value2 value1 > value2	ANY1 === ANY2 ANY1 > ANY2
逻辑函数	boolean1 or boolean2 boolean IS FALSE NOTboolean	boolean1 \|\| boolean2 boolean.isFalse !boolean
算数函数	numeric1 + numeric2 POWER(numeric1, numeric2)	numeric1 + numeric2 numeric1.power(numeric2)
字符串函数	string1 \|\| string2 UPPER(string) CHAR_LENGTH(string)	string1 + string2 string.upperCase() string.charLength()
时间函数	DATE String TIMESTAMP string CURRENT_TIME INTERVAL string range	string.toDate string.toTimestamp currentTime() NUMERIC.days NUMERIC.minutes
聚合函数	count(*) `sum([all\|distinct] expression)` rank() ROW_NUMBER()	field.count field.sum0

UDF

用户定义函数（User-defined Functions, UDF）是一个重要的特征，因为他们显著地扩展了查询（Query）的表达能力。一些系统内置函数无法解决的需求，我们可以用UDF来自定义实现。

注册用户自定义函数UDF

在大多数情况下，用户定义的函数必须先注册，然后才能在查询中使用。不需要专门为Scala的Table API注册函数。
函数通过调用registerFunction()方法在TableEnvironment中注册，当用户定义的函数被注册时，它被插入到TableEnvironment的函数目录中，这样Tabel API或SQL解析器就可以识别并正确解释它。

标量函数（Scalar Functions）

用户定义的标量函数，可以将0、1或多个标量值，映射到新的标量值。
为了定义标量函数，必须在org.apache.flink.table.functions中扩展基类Scalar Function，并实现（一个或多个）求值（evaluation，eval）方法。标量函数的行为由求值方法决定，求值方法必须公开声明并命名为eval（直接def声明，没有override）。求值方法的参数类型和返回类型，确定了标量函数的参数和返回类型。

我们自己定义HashCode函数，在TableEnvironment中注册，并在查询中调用它。

package com.zh.apitest.tableapi.udf;

import com.zh.apitest.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;

/**
 * author: zhanghui
 * project: big-data-learning
 * package: com.zh.apitest.tableapi.udf
 * filename: UdfTest1_ScalarFunction
 * date: 2021/12/27 14:20
 * description: udf ScalarFunction
 */
public class UdfTest1_ScalarFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 读取数据
        DataStreamSource<String> inputStream = env.readTextFile("flink-FlinkTutorial/src/main/resources/sensor.txt");

        // 2. 转成POJO
        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], Long.valueOf(fields[1]), Double.valueOf(fields[2]));
        });

        // 3. 将流转换成表
        Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");

        // 4. 自定义标量函数，实现求id的hash值
        // 4.1 table API
        HashCode hashCode = new HashCode(23);
        // 需要在环境中注册UDF
        tableEnv.registerFunction("hashCode", hashCode);
        Table resultTable = sensorTable.select("id, ts, hashCode(id)");

        // 4.2 SQL
        tableEnv.createTemporaryView("sensor", sensorTable);
        Table resultSqlTable = tableEnv.sqlQuery("select id, ts, hashCode(id) from sensor");

        // 打印输出
        tableEnv.toAppendStream(resultTable, Row.class).print("result");
        tableEnv.toAppendStream(resultSqlTable, Row.class).print("sql");

        env.execute();
    }

    // 实现自定义的ScalarFunction
    public static class HashCode extends ScalarFunction {
        private int factor = 13;

        public HashCode(int factor) {
            this.factor = factor;
        }

        public int eval(String str) {
            return str.hashCode() * factor;
        }
    }
}

表函数（Table Functions）

与用户定义的标量函数类似，用户定义的表函数，可以将0、1或多个标量值作为输入参数；与标量函数不同的是，它可以返回任意数量的行作为输出，而不是单个值。
为了定义一个表函数，必须扩展org.apache.flink.table.functions中的基类TableFunction并实现（一个或多个）求值方法。表函数的行为由其求值方法决定，求值方法必须是public的，并命名为eval。求值方法的参数类型，决定表函数的所有有效参数。
返回表的类型由TableFunction的泛型类型确定。求值方法使用protected collect(T)方法发出输出行。
joinLateral算子，会将外部表中的每一行，与表函数（TableFunction，算子的参数是他的表达式）计算得到的所有行连接起来。
而leftOuterJoinLateral算子，则是左外连接，它同样会将外部表中的每一行与表函数计算生成的所有行连接起来；并且，对于表函数返回的是空表的外部行，也要保留下来。
在SQL中，则需要使用Lateral Table(<TableFunction>)，或者带有on true条件的左连接。

package com.zh.apitest.tableapi.udf;

import com.zh.apitest.beans.SensorReading;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * author: zhanghui
 * project: big-data-learning
 * package: com.zh.apitest.tableapi.udf
 * filename: UdfTest2_TableFunction
 * date: 2021/12/27 15:52
 * description: udf TableFunction
 */
public class UdfTest2_TableFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 读取数据
        DataStreamSource<String> inputStream = env.readTextFile("flink-FlinkTutorial/src/main/resources/sensor.txt");

        // 2. 转成POJO
        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], Long.valueOf(fields[1]), Double.valueOf(fields[2]));
        });

        // 3. 将流转换成表
        Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");

        // 4. 自定义表函数，实现求id的拆分，并输出（word,length）
        // 4.1 table API
        Split split = new Split("_");
        // 需要在环境中注册UDF
        tableEnv.registerFunction("split", split);
        Table resultTable = sensorTable
                .joinLateral("split(id) as (word, length)")
                .select("id, ts, word, length");

        // 4.2 SQL
        tableEnv.createTemporaryView("sensor", sensorTable);
        Table resultSqlTable = tableEnv.sqlQuery("select id, ts, word, length" +
                " from sensor, lateral table(split(id)) as splitid(word, length)");

        // 打印输出
        tableEnv.toAppendStream(resultTable, Row.class).print("result");
        tableEnv.toAppendStream(resultSqlTable, Row.class).print("sql");

        env.execute();
    }

    // 实现自定义tableFunction
    public static class Split extends TableFunction<Tuple2<String, Integer>> {
        // 定义私有属性，分隔符
        private String separator = ",";

        public Split(String separator) {
            this.separator = separator;
        }

        // 必须实现eval方法，没有返回值
        public void eval(String str) {
            for (String s : str.split(separator)) {
                collector.collect(new Tuple2<>(s, s.length()));
            }
        }
    }
}

聚合函数（Aggregate Functions）

用户自定义聚合函数（User-Defined Aggregate Functions，UDAGGs）可以把一个表中的数据，聚合成一个标量值。用户定义的聚合函数，是通过继承AggregateFunction抽象类实现的。

上图中显示了一个聚合的例子。
假设现在有一张表，包含了各种饮料的数据。该表由三列（id，name和price）、五行组成数据。现在我们需要找到表中所有饮料的最高价格，即执行max()聚合，结果将是一个数值。

AggregateFunction的工作原理如下。

首先，它需要一个累加器，用来保存聚合中间结果的数据结构（状态）。可以通过调用AggregateFunction的createAccumulator()方法创建空累加器。
随后，对每个输入行调用函数的accumulate()方法来更新累加器。
处理完所有行后，将调用函数的getValue()方法来计算并返回最终结果。

AggregateFunction要求必须实现的方法：

createAccumulator()
accumulate()
getValue()

除了上述方法之外，还有一些可选择实现的方法。其中一些方法，可以让系统执行查询更有效率，而另一些方法，对于这些场景是必须的。例如，如果聚合函数应用在会话窗口session group window的上下文中，则merge()方法时必须的。

retract()
merge()
resetAccumulator()

接下来我们写了一个自定义AggregateFunction，计算一下每个sensor的平均温度值。

package com.zh.apitest.tableapi.udf;

import com.zh.apitest.beans.SensorReading;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;

/**
 * author: zhanghui
 * project: big-data-learning
 * package: com.zh.apitest.tableapi.udf
 * filename: UdfTest3_AggregateFunction
 * date: 2021/12/27 16:12
 * description: udf aggregateFunction
 */
public class UdfTest3_AggregateFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 读取数据
        DataStreamSource<String> inputStream = env.readTextFile("flink-FlinkTutorial/src/main/resources/sensor.txt");

        // 2. 转成POJO
        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], Long.valueOf(fields[1]), Double.valueOf(fields[2]));
        });

        // 3. 将流转换成表
        Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");

        // 4. 自定义集合函数， 求当前传感器的平均温度值
        // 4.1 table API
        AvgTemp avgTemp = new AvgTemp();
        // 需要在环境中注册UDF
        tableEnv.registerFunction("avgTemp", avgTemp);
        Table resultTable = sensorTable
                .groupBy("id")
                .aggregate("avgTemp(temp) as avgtemp")
                .select("id, avgtemp");

        // 4.2 SQL
        tableEnv.createTemporaryView("sensor", sensorTable);
        Table resultSqlTable = tableEnv.sqlQuery("select id, avgTemp(temp) " +
                "from sensor group by id");

        // 打印输出
        tableEnv.toRetractStream(resultTable, Row.class).print("result");
        tableEnv.toRetractStream(resultSqlTable, Row.class).print("sql");

        env.execute();
    }

    // 实现自定义的AggregateFunction
    public static class AvgTemp extends AggregateFunction<Double, Tuple2<Double, Integer>> {

        @Override
        public Double getValue(Tuple2<Double, Integer> accumulator) {
            return accumulator.f0 / accumulator.f1;
        }

        @Override
        public Tuple2<Double, Integer> createAccumulator() {
            return new Tuple2<>(0.0, 0);
        }

        // 必须实现一个accumulate方法，来数据之后更新状态
        public void accumulate(Tuple2<Double, Integer> accumulator, Double temp) {
            accumulator.f0 += temp;
            accumulator.f1 += 1;
        }
    }
}

表聚合函数（Table Aggregate Functions）

用户定义的表聚合函数（User-Defined Table Aggregate Functions, UDTAGGs），可以把一个表中数据，聚合为具有多行和多列的结果表。这跟 AggregateFunction 非常类似，只是之前聚合结果是一个标量值，现在变成了一张表。

比如现在我们需要找到表中所有饮料的前2个最高价格，即执行top2()表聚合。我们需要检查5行中的每一行，得到的结果将是一个具有排序后前2个值的表。用户定义的表聚合函数，是通过继承TableAggregateFunction抽象类来实现的。

TableAggregateFunction的工作原理如下。

首先，它同样需要一个累加器（Accumulate），它是保存聚合中间结果的数据结构。通过调用TableAggregateFunction的createAccumulator()方法可以创建累加器。
随后，对每个输入行调用函数的accumulate()方法来更新累加器。
处理完所有行后，将调用函数的emitValue()方法来计算并返回最终结果。

AggregationFunction要求必须实现的方法：

createAccumulator()
accumulate()

除了上述方法之外，还有一些可选择实现的方法。

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

接下来我们写一个自定义TableAggregateFunction，用来提取每个sensor最高的两个温度值。

package com.zh.apitest.tableapi.udf;

import com.zh.apitest.beans.SensorReading;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

/**
 * author: zhanghui
 * project: big-data-learning
 * package: com.zh.apitest.tableapi.udf
 * filename: UdfTest4_TableAggregateFunction
 * date: 2022/1/7 16:00
 * description: udf TableAggregateFunction
 */
public class UdfTest4_TableAggregateFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 读取数据
        DataStreamSource<String> inputStream = env.readTextFile("flink-FlinkTutorial/src/main/resources/sensor.txt");

        // 2. 转成POJO
        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], Long.valueOf(fields[1]), Double.valueOf(fields[2]));
        });

        // 3.将流转换成表
        Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");

        // 创建一个表聚合函数实例
        Top2Temp top2Temp = new Top2Temp();
        tableEnv.registerFunction("top2Temp", top2Temp);
        Table resultTable = sensorTable
                .groupBy("id")
                .flatAggregate("top2Temp(temp) as (temp, rank)")
                .select("id, temp, rank");

        // 打印输出
        tableEnv.toRetractStream(resultTable, Row.class).print("result");

        env.execute();
    }

    // 定义一个Accumulator
    public static class Top2TempAcc {
        double highestTemp = Double.MIN_VALUE;
        double secondHighestTemp = Double.MIN_VALUE;
    }

    // 自定义表聚合函数
    public static class Top2Temp extends TableAggregateFunction<Tuple2<Double, Integer>, Top2TempAcc> {

        @Override
        public Top2TempAcc createAccumulator() {
            return new Top2TempAcc();
        }

        // 实现计算聚合结果的函数accumulate
        public void accumulate(Top2TempAcc acc, Double temp) {
            if (temp > acc.highestTemp) {
                acc.secondHighestTemp = acc.highestTemp;
                acc.highestTemp = temp;
            } else if (temp > acc.secondHighestTemp) {
                acc.secondHighestTemp = temp;
            }
        }

        // 实现一个输出结果的方法，最终处理完表中所有数据时调用
        public void emitValue(Top2TempAcc acc, Collector<Tuple2<Double, Integer>> out) {
            out.collect(new Tuple2<>(acc.highestTemp, 1));
            out.collect(new Tuple2<>(acc.secondHighestTemp, 2));
        }
    }
}

I-Table API与SQL

整体介绍

什么是Table API和Flink SQL

需要引入的依赖

两种planner(old&blink)的区别

API调用

基本程序结构

创建表环境

在Catalog中注册表

表（Table）

连接到文件系统（Csv格式）

连接到Kafka

表的查询

Table API调用

SQL查询

将DataStream转换成表

代码表达

数据类型与Table Schema

创建临时视图（Temporary View）

输出表

输出到文件

更新模式（Update Mode）

1）追加模式（Append Mode）

2）撤回模式（Retract Mode）

3）Upsert（更新插入）模式

输出到Kafka

输出到ElasticSearch

输出到MySql

将表转换成DataStream

Query的解释和执行

流处理中的特殊概念

流处理和关系代数（表，及SQL）的区别

动态表（Dynamic Tables）

流式持续查询的过程

将流转换为表（Table）

持续查询（Continuous Query）

将动态表转换成流

时间特性

处理时间（Processing Time）

事件时间（Event Time）

窗口（Windows）

分组窗口（Group Windows）

滚动窗口

滑动窗口

会话窗口

Over Windows

SQL中的窗口定义

Group Windows

Over Windows

练习代码

函数（Functions）

系统内置函数

UDF

注册用户自定义函数UDF

标量函数（Scalar Functions）

表函数（Table Functions）

聚合函数（Aggregate Functions）

表聚合函数（Table Aggregate Functions）