概述
Flink 提供了对于“表”处理的支持，这就是更高层级的应用API，在 Flink 中被称为Table API 和 SQL。Table API 顾名思义，就是基于“表”（Table）的一套 API，它是内嵌在 Java、 Scala 等语言中的一种声明式领域特定语言（DSL），也就是专门为处理表而设计的；在此基础上，Flink 还基于Apache Calcite 实现了对 SQL 的支持。这样一来，我们就可以在 Flink 程序中直接写 SQL 来实现处理需求了。
在 Flink 中这两种 API 被集成在一起，SQL 执行的对象也是 Flink 中的表（Table），所以我们一般会认为它们是一体的，本章会放在一起进行介绍。Flink 是批流统一的处理框架，无论是批处理（DataSet API）还是流处理（DataStream API），在上层应用中都可以直接使用 Table API 或者 SQL 来实现；这两种 API 对于一张表执行相同的查询操作，得到的结果是完全一样的。我们主要还是以流处理应用为例进行讲解。
需要说明的是，Table API 和 SQL 最初并不完善，在 Flink 1.9 版本合并阿里巴巴内部版本Blink 之后发生了非常大的改变，此后也一直处在快速开发和完善的过程中，直到 Flink 1.12 版本才基本上做到了功能上的完善。而即使是在目前最新的 1.13 版本中，Table API 和 SQL 也依然不算稳定，接口用法还在不停调整和更新。所以这部分希望大家重在理解原理和基本用法， 具体的 API 调用可以随时关注官网的更新变化。
一、快速上手
1. 需要引入的依赖
我们想要在代码中使用Table API，必须引入相关的依赖。
dependency>
groupId>org.apache.flinkgroupId>
artifactId>flink-table-api-java-bridge_${scala.binary.version}artifactId>
version>${flink.version}version>
dependency>
这里的依赖是一个 Java 的“桥接器”（bridge），主要就是负责 Table API 和下层 DataStream API 的连接支持，按照不同的语言分为 Java 版和 Scala 版。
如果我们希望在本地的集成开发环境（IDE）里运行 Table API 和 SQL，还需要引入以下依赖：
dependency>
groupId>org.apache.flinkgroupId>
artifactId>flink-table-planner-blink_${scala.binary.version}artifactId>
version>${flink.version}version>
dependency>
dependency>
groupId>org.apache.flinkgroupId>
artifactId>flink-streaming-scala_${scala.binary.version}artifactId>
version>${flink.version}version>
dependency>
这里主要添加的依赖是一个“计划器”（planner），它是 Table API 的核心组件，负责提供运行时环境，并生成程序的执行计划。这里我们用到的是新版的 blink planner。由于 Flink 安装包的 lib 目录下会自带planner，所以在生产集群环境中提交的作业不需要打包这个依赖。
而在Table API 的内部实现上，部分相关的代码是用 Scala 实现的，所以还需要额外添加一个 Scala 版流处理的相关依赖。
另外，如果想实现自定义的数据格式来做序列化，可以引入下面的依赖：
dependency>
groupId>org.apache.flinkgroupId>
artifactId>flink-table-commonartifactId>
version>${flink.version}version>
dependency>
2. 一个简单示例
有了基本的依赖，接下来我们就可以尝试在 Flink 代码中使用 Table API 和 SQL 了。比如， 我们可以自定义一些 Event 类型（包含了 user、url 和 timestamp 三个字段，参考 5.2.1 小节中的定义）的用户访问事件，作为输入的数据源；而后从中提取 url 地址和用户名 user 两个字段作为输出。
如果使用DataStream API，我们可以直接读取数据源后，用一个简单转换算子 map 来做字段的提取。而这个需求直接写 SQL 的话，实现会更加简单：
select url, user from EventTable;
这里我们把流中所有数据组成的表叫作 EventTable。在 Flink 代码中直接对这个表执行上面的 SQL，就可以得到想要提取的数据了。
在代码中具体实现如下：
import com.atguigu.chapter05.Event;
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.bridge.java.StreamTableEnvironment;
import static org.apache.flink.table.api.Expressions.$;
public class SimpleTableExample {
public static void main(String[] args) throws Exception {
// 获取流执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 1. 读取数据源
SingleOutputStreamOperatorEvent> eventStream = env
.fromElements(
new Event("Alice", "./home", 1000L),
new Event("Bob", "./cart", 1000L),
new Event("Alice", "./prod?id=1", 5 * 1000L),
new Event("Cary", "./home", 60 * 1000L),
new Event("Bob", "./prod?id=3", 90 * 1000L),
new Event("Alice", "./prod?id=7", 105 * 1000L)
);
// 2. 获取表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 3. 将数据流转换成表
Table eventTable = tableEnv.fromDataStream(eventStream);
// 4. 用执行SQL 的方式提取数据
Table resultTable1 = tableEnv.sqlQuery("select url, user from " + eventTable);
// 5. 基于Table直接转换
Table resultTable2 = eventTable.select($("user"), $("url"))
.where($("user").isEqual("Alice"));
// 6. 将表转换成数据流，打印输出
tableEnv.toDataStream(resultTable1).print("result1");
tableEnv.toDataStream(resultTable2).print("result2");
// 执行程序
env.execute();
}
}
这里我们需要创建一个“表环境”（TableEnvironment），然后将数据流（DataStream）转换成一个表（Table）；之后就可以执行 SQL 在这个表中查询数据了。查询得到的结果依然是一个表，把它重新转换成流就可以打印输出了。
代码执行的结果如下：
+I[./home, Alice]
+I[./cart, Bob]
+I[./prod?id=1, Alice]
+I[./home, Cary]
+I[./prod?id=3, Bob]
+I[./prod?id=7, Alice]
可以看到，我们将原始的 Event 数据转换成了(url，user)这样类似二元组的类型。每行输出前面有一个“+I”标志，这是表示每条数据都是“插入”（Insert）到表中的新增数据。
Table 是 Table API 中的核心接口类，对应着我们熟悉的“表”的概念。基于Table 我们也可以调用一系列查询方法直接进行转换，这就是所谓 Table API 的处理方式：
// 用 Table API 方式提取数据
Table clickTable2 = eventTable.select($("url"), $("user"));
这里的$符号是Table API 中定义的“表达式”类Expressions 中的一个方法，传入一个字段名称，就可以指代数据中对应字段。将得到的表转换成流打印输出，会发现结果与直接执行SQL 完全一样。
二、基本API
1. 程序架构
在 Flink 中，Table API 和 SQL 可以看作联结在一起的一套 API，这套 API 的核心概念就是“表”（Table）。在我们的程序中，输入数据可以定义成一张表；然后对这张表进行查询，就可以得到新的表，这相当于就是流数据的转换操作；最后还可以定义一张用于输出的表，负责将处理结果写入到外部系统。
我们可以看到，程序的整体处理流程与 DataStream API 非常相似，也可以分为读取数据源（Source）、转换（Transform）、输出数据（Sink）三部分；只不过这里的输入输出操作不需要额外定义，只需要将用于输入和输出的表定义出来，然后进行转换查询就可以了。
程序基本架构如下：
// 创建表环境
TableEnvironment tableEnv = ...;
// 创建输入表，连接外部系统读取数据
tableEnv.executeSql("CREATE TEMPORARY TABLE inputTable ... WITH ( 'connector' = ... )");
// 注册一个表，连接到外部系统，用于输出
tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )");
// 执行 SQL 对表进行查询转换，得到一个新的表
Table table1 = tableEnv.sqlQuery("SELECT ... FROM inputTable... ");
// 使用 Table API 对表进行查询转换，得到一个新的表
Table table2 = tableEnv.from("inputTable").select(...);
// 将得到的结果写入输出表
TableResult tableResult = table1.executeInsert("outputTable");
与上一节中不同，这里不是从一个 DataStream 转换成 Table，而是通过执行 DDL 来直接创建一个表。这里执行的 CREATE 语句中用 WITH 指定了外部系统的连接器，于是就可以连接外部系统读取数据了。这其实是更加一般化的程序架构，因为这样我们就可以完全抛开DataStream API，直接用 SQL 语句实现全部的流处理过程。
而后面对于输出表的定义是完全一样的。可以发现，在创建表的过程中，其实并不区分“输入”还是“输出”，只需要将这个表“注册”进来、连接到外部系统就可以了；这里的 inputTable、outputTable 只是注册的表名，并不代表处理逻辑，可以随意更换。至于表的具体作用，则要等到执行后面的查询转换操作时才能明确。我们直接从 inputTable 中查询数据，那么 inputTable 就是输入表；而 outputTable 会接收另外表的结果进行写入，那么就是输出表。
在早期的版本中，有专门的用于输入输出的 TableSource 和TableSink，这与流处理里的概念是一一对应的；不过这种方式与关系型表和 SQL 的使用习惯不符，所以已被弃用，不再区分 Source 和 Sink。
2. 创建表环境
对于 Flink 这样的流处理框架来说，数据流和表在结构上还是有所区别的。所以使用 Table API 和 SQL 需要一个特别的运行时环境，这就是所谓的“表环境”（TableEnvironment）。它主要负责：
（1） 注册Catalog 和表；
（2） 执行 SQL 查询；
（3） 注册用户自定义函数（UDF）；
（4） DataStream 和表之间的转换。
这里的 Catalog 就是“目录”，与标准 SQL 中的概念是一致的，主要用来管理所有数据库
（database）和表（table）的元数据（metadata）。通过 Catalog 可以方便地对数据库和表进行查询的管理，所以可以认为我们所定义的表都会“挂靠”在某个目录下，这样就可以快速检索。在表环境中可以由用户自定义Catalog，并在其中注册表和自定义函数（UDF）。默认的 Catalog就叫作default_catalog。
每个表和 SQL 的执行，都必须绑定在一个表环境（TableEnvironment）中。TableEnvironment是 Table API 中提供的基本接口类，可以通过调用静态的 create()方法来创建一个表环境实例。方法需要传入一个环境的配置参数EnvironmentSettings，它可以指定当前表环境的执行模式和计划器（planner）。执行模式有批处理和流处理两种选择，默认是流处理模式；计划器默认使用 blink planner。
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.inStreamingMode() // 使用流处理模式
.build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
对于流处理场景，其实默认配置就完全够用了。所以我们也可以用另一种更加简单的方式来创建表环境：
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
这里我们引入了一个“ 流式表环境 ”（ StreamTableEnvironment ）， 它是继承自TableEnvironment 的子接口。调用它的 create() 方法， 只需要直接将当前的流执行环境（StreamExecutionEnvironment）传入，就可以创建出对应的流式表环境了。这也正是我们在上一节简单示例中使用的方式。
3. 创建表
表（Table）是我们非常熟悉的一个概念，它是关系型数据库中数据存储的基本形式，也 是SQL 执行的基本对象。Flink 中的表概念也并不特殊，是由多个“行”数据构成的，每个行（Row） 又可以有定义好的多个列（Column）字段；整体来看，表就是固定类型的数据组成的二维矩阵。
为了方便地查询表，表环境中会维护一个目录（Catalog）和表的对应关系。所以表都是通过Catalog 来进行注册创建的。表在环境中有一个唯一的 ID，由三部分组成：目录（catalog） 名，数据库（database）名，以及表名。在默认情况下，目录名为 default_catalog，数据库名为default_database。所以如果我们直接创建一个叫作 MyTable 的表，它的 ID 就是：
default_catalog.default_database.MyTable
具体创建表的方式，有通过连接器（connector）和虚拟表（virtual tables）两种。
3.1 连接器表（Connector Tables）
最直观的创建表的方式，就是通过连接器（connector）连接到一个外部系统，然后定义出对应的表结构。例如我们可以连接到 Kafka 或者文件系统，将存储在这些外部系统的数据以“表” 的形式定义出来，这样对表的读写就可以通过连接器转换成对外部系统的读写了。当我们在表 环境中读取这张表，连接器就会从外部系统读取数据并进行转换；而当我们向这张表写入数据， 连接器就会将数据输出（Sink）到外部系统中。
在代码中，我们可以调用表环境的 executeSql()方法，可以传入一个 DDL 作为参数执行SQL 操作。这里我们传入一个CREATE 语句进行表的创建，并通过 WITH 关键字指定连接到外部系统的连接器：
tableEnv.executeSql("CREATE [TEMPORARY] TABLE MyTable ... WITH ( 'connector' = ... )");
这里的
TEMPORARY关键字可以省略。
这里没有定义 Catalog 和 Database ， 所以都是默认的， 表的完整ID就是default_catalog.default_database.MyTable。如果希望使用自定义的目录名和库名，可以在环境中进行设置：
tEnv.useCatalog("custom_catalog");
tEnv.useDatabase("custom_database");
这样我们创建的表完整 ID 就变成了 custom_catalog.custom_database.MyTable。之后在表环境中创建的所有表，ID 也会都以 custom_catalog.custom_database 作为前缀。
3.2 虚拟表（Virtual Tables）
在环境中注册之后，我们就可以在 SQL 中直接使用这张表进行查询转换了。
这里调用了表环境的 sqlQuery()方法，直接传入一条 SQL 语句作为参数执行查询，得到的结果是一个 Table 对象。Table 是 Table API 中提供的核心接口类，就代表了一个 Java 中定义的表实例。
得到的 newTable 是一个中间转换结果，如果之后又希望直接使用这个表执行 SQL，又该怎么做呢？由于 newTable 是一个 Table 对象，并没有在表环境中注册；所以我们还需要将这个中间结果表注册到环境中，才能在 SQL 中使用：
tableEnv.createTemporaryView("NewTable", newTable);
我们发现，这里的注册其实是创建了一个“虚拟表”（Virtual Table）。这个概念与 SQL 语法中的视图（View）非常类似，所以调用的方法也叫作创建“虚拟视图”（createTemporaryView）。视图之所以是“虚拟”的，是因为我们并不会直接保存这个表的内容，并没有“实体”；只是在用到这张表的时候，会将它对应的查询语句嵌入到 SQL 中。
注册为虚拟表之后，我们就又可以在 SQL 中直接使用 NewTable 进行查询转换了。不难看到，通过虚拟表可以非常方便地让 SQL 分步骤执行得到中间结果，这为代码编写提供了很大的便利。
另外，虚拟表也可以让我们在Table API 和 SQL 之间进行自由切换。一个 Java 中的Table 对象可以直接调用Table API 中定义好的查询转换方法，得到一个中间结果表；这跟对注册好的表直接执行 SQL 结果是一样的。具体我们会在后面继续讲解。
4. 表的查询
Flink 为我们提供了两种查询方式：
SQL和Table API。
4.1 执行 SQL 进行查询
基于表执行SQL 语句，是我们最为熟悉的查询方式。Flink 基于 Apache Calcite 来提供对SQL 的支持，Calcite 是一个为不同的计算平台提供标准 SQL 查询的底层工具，很多大数据框架比如 Apache Hive、Apache Kylin 中的SQL 支持都是通过集成 Calcite 来实现的。
在代码中，我们只要调用表环境的 sqlQuery()方法，传入一个字符串形式的 SQL 查询语句就可以了。执行得到的结果，是一个 Table 对象。
// 创建表环境
TableEnvironment tableEnv = ...;
// 创建表
tableEnv.executeSql("CREATE TABLE EventTable ... WITH ( 'connector' = ... )");
// 查询用户 Alice 的点击事件，并提取表中前两个字段
Table aliceVisitTable = tableEnv.sqlQuery( "SELECT user, url " +
"FROM EventTable " +
"WHERE user = 'Alice' ");
目前 Flink 支持标准 SQL 中的绝大部分用法，并提供了丰富的计算函数。这样我们就可以把已有的技术迁移过来，像在 MySQL、Hive 中那样直接通过编写 SQL 实现自己的处理需求，从而大大降低了 Flink 上手的难度。
例如，我们也可以通过GROUP BY 关键字定义分组聚合，调用COUNT()、SUM()这样的函数来进行统计计算：
Table urlCountTable = tableEnv.sqlQuery( "SELECT user, COUNT(url) " +
"FROM EventTable " + "GROUP BY user "
上面的例子得到的是一个新的 Table 对象，我们可以再次将它注册为虚拟表继续在 SQL 中调用。另外，我们也可以直接将查询的结果写入到已经注册的表中，这需要调用表环境的executeSql()方法来执行DDL，传入的是一个INSERT 语句：
// 注册表
tableEnv.executeSql("CREATE TABLE EventTable ... WITH ( 'connector' = ... )"); tableEnv.executeSql("CREATE TABLE OutputTable ... WITH ( 'connector' = ... )");
// 将查询结果输出到 OutputTable 中
tableEnv.executeSql ( "INSERT INTO OutputTable " +
"SELECT user, url " + "FROM EventTable " + "WHERE user = 'Alice' ");
4.2 调用 Table API 进行查询
另外一种查询方式就是调用 Table API。这是嵌入在 Java 和 Scala 语言内的查询 API，核心就是 Table 接口类，通过一步步链式调用 Table 的方法，就可以定义出所有的查询转换操作。每一步方法调用的返回结果，都是一个Table。
由于Table API 是基于Table 的Java 实例进行调用的，因此我们首先要得到表的Java 对象。基于环境中已注册的表，可以通过表环境的 from()方法非常容易地得到一个Table 对象：
Table eventTable = tableEnv.from("EventTable");
传入的参数就是注册好的表名。注意这里 eventTable 是一个Table 对象，而EventTable 是在环境中注册的表名。得到 Table 对象之后，就可以调用 API 进行各种转换操作了，得到的是一个新的Table 对象：
Table maryClickTable = eventTable
.where($("user").isEqual("Alice"))
.select($("url"), $("user"));
这里每个方法的参数都是一个“表达式”（Expression），用方法调用的形式直观地说明了想要表达的内容；“$”符号用来指定表中的一个字段。上面的代码和直接执行 SQL 是等效的。 Table API 是嵌入编程语言中的DSL，SQL 中的很多特性和功能必须要有对应的实现才可以使用，因此跟直接写 SQL 比起来肯定就要麻烦一些。目前Table API 支持的功能相对更少， 可以预见未来 Flink 社区也会以扩展 SQL 为主，为大家提供更加通用的接口方式；所以我们接下来也会以介绍 SQL 为主，简略地提及 Table API。
4.3 两种 API 的结合使用
可以发现，无论是调用Table API 还是执行 SQL，得到的结果都是一个Table 对象；所以这两种 API 的查询可以很方便地结合在一起。
（1） 无论是那种方式得到的 Table 对象，都可以继续调用 Table API 进行查询转换；
（2） 如果想要对一个表执行 SQL 操作（用 FROM 关键字引用），必须先在环境中对它进行注册。所以我们可以通过创建虚拟表的方式实现两者的转换：
tableEnv.createTemporaryView("MyTable", myTable);
注意：这里的第一个参数”MyTable”是注册的表名，而第二个参数 myTable 是 Java 中的Table 对象。
另外要说明的是，在前面小节的简单示例中，我们并没有将Table 对象注册为虚拟表就直接在 SQL 中使用了：
Table clickTable = tableEnvironment.sqlQuery("select url, user from " + eventTable);
这其实是一种简略的写法，我们将Table 对象名 eventTable 直接以字符串拼接的形式添加到 SQL 语句中，在解析时会自动注册一个同名的虚拟表到环境中，这样就省略了创建虚拟视图的步骤。
两种 API 殊途同归，实际应用中可以按照自己的习惯任意选择。不过由于结合使用容易引起混淆，而 Table API 功能相对较少、通用性较差，所以企业项目中往往会直接选择 SQL 的方式来实现需求。
5. 输出表
表的创建和查询，就对应着流处理中的读取数据源（Source）和转换（Transform）；而最后一个步骤 Sink，也就是将结果数据输出到外部系统，就对应着表的输出操作。
在代码上，输出一张表最直接的方法，就是调用 Table 的方法 executeInsert()方法将一个Table 写入到注册过的表中，方法传入的参数就是注册的表名。
// 注册表，用于输出数据到外部系统
tableEnv.executeSql("CREATE TABLE OutputTable ... WITH ( 'connector' = ... )");
// 经过查询转换，得到结果表
Table result = ...
// 将结果表写入已注册的输出表中
result.executeInsert("OutputTable");
在底层，表的输出是通过将数据写入到TableSink 来实现的。TableSink 是Table API 中提供的一个向外部系统写入数据的通用接口，可以支持不同的文件格式（比如 CSV、Parquet）、存储数据库（比如 JDBC、HBase、Elasticsearch）和消息队列（比如 Kafka）。它有些类似于 DataStream API 中调用addSink()方法时传入的 SinkFunction，有不同的连接器对它进行了实现。关于不同外部系统的连接器，我们会在后面展开介绍。
这里可以发现，我们在环境中注册的“表”，其实在写入数据的时候就对应着一个 TableSink。
6. 表和流的转换
从创建表环境开始，历经表的创建、查询转换和输出，我们已经可以使用 Table API 和 SQL 进行完整的流处理了。不过在应用的开发过程中，我们测试业务逻辑一般不会直接将结果直接写入到外部系统，而是在本地控制台打印输出。对于 DataStream 这非常容易，直接调用 print() 方法就可以看到结果数据流的内容了；但对于 Table 就比较悲剧——它没有提供 print()方法。这该怎么办呢？
在 Flink 中我们可以将 Table 再转换成 DataStream，然后进行打印输出。这就涉及了表和流的转换。
6.1 将表（Table）转换成流（DataStream）
（1）调用 toDataStream()方法
将一个Table 对象转换成DataStream 非常简单，只要直接调用表环境的方法 toDataStream() 就可以了。例如，我们可以将前面小节经查询转换得到的表 maryClickTable 转换成流打印输出，这代表了“Mary 点击的 url 列表”：
Table aliceVisitTable = tableEnv.sqlQuery(
"SELECT user, url " +
"FROM EventTable " +
"WHERE user = 'Alice' "
);
// 将表转换成数据流
tableEnv.toDataStream(aliceVisitTable).print();
这里需要将要转换的Table 对象作为参数传入。
（2） 调用 toChangelogStream()方法
将 maryClickTable 转换成流打印输出是很简单的；然而，如果我们同样希望将“用户点击次数统计”表 urlCountTable 进行打印输出，就会抛出一个TableException 异常：
Exception in thread "main" org.apache.flink.table.api.TableException: Table sink
'default_catalog.default_database.Unregistered_DataStream_Sink_1' doesn't support consuming update changes ...
这表示当前的TableSink 并不支持表的更新（update）操作。这是什么意思呢？
因为 print 本身也可以看作一个 Sink 操作，所以这个异常就是说打印输出的 Sink 操作不支持对数据进行更新。具体来说，urlCountTable 这个表中进行了分组聚合统计，所以表中的每一行是会“更新”的。也就是说，Alice 的第一个点击事件到来，表中会有一行(Alice, 1)； 第二个点击事件到来，这一行就要更新为(Alice, 2)。但之前的(Alice, 1)已经打印输出了，“覆水难收”，我们怎么能对它进行更改呢？所以就会抛出异常。
解决的思路是，对于这样有更新操作的表，我们不要试图直接把它转换成 DataStream 打印输出，而是记录一下它的“更新日志”（change log）。这样一来，对于表的所有更新操作，就变成了一条更新日志的流，我们就可以转换成流打印输出了。
代码中需要调用的是表环境的 toChangelogStream()方法：
Table urlCountTable = tableEnv.sqlQuery( "SELECT user, COUNT(url) " +
"FROM EventTable " + "GROUP BY user "
);
tableEnv.toChangelogStream(urlCountTable).print("count");
与“更新日志流”（Changelog Streams）对应的，是那些只做了简单转换、没有进行聚合统计的表，例如前面提到的 maryClickTable。它们的特点是数据只会插入、不会更新，所以也被叫作“仅插入流”（Insert-Only Streams）。
6.2 将流（DataStream）转换成表（Table）
（1） 调用 fromDataStream()方法
想要将一个DataStream 转换成表也很简单，可以通过调用表环境的 fromDataStream()方法来实现，返回的就是一个 Table 对象。例如，我们可以直接将事件流 eventStream 转换成一个表：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 获取表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 读取数据源
SingleOutputStreamOperatorEvent> eventStream = env.addSource(...)
// 将数据流转换成表
Table eventTable = tableEnv.fromDataStream(eventStream);
由于流中的数据本身就是定义好的 POJO 类型 Event，所以我们将流转换成表之后，每一行数据就对应着一个Event，而表中的列名就对应着Event 中的属性。
另外，我们还可以在 fromDataStream()方法中增加参数，用来指定提取哪些属性作为表中的字段名，并可以任意指定位置：
// 提取 Event 中的 timestamp 和 url 作为表中的列
Table eventTable2 = tableEnv.fromDataStream(eventStream, $("timestamp"),$("url"));
需要注意的是，timestamp 本身是 SQL 中的关键字，所以我们在定义表名、列名时要尽量避免。这时可以通过表达式的 as()方法对字段进行重命名：
// 将 timestamp 字段重命名为 ts
Table eventTable2 = tableEnv.fromDataStream(eventStream, $("timestamp").as("ts"),$("url"));
（2） 调用createTemporaryView()方法
调用 fromDataStream()方法简单直观，可以直接实现DataStream 到 Table 的转换；不过如果我们希望直接在 SQL 中引用这张表，就还需要调用表环境的 createTemporaryView()方法来创建虚拟视图了。
对于这种场景，也有一种更简洁的调用方式。我们可以直接调用 createTemporaryView() 方法创建虚拟表，传入的两个参数，第一个依然是注册的表名，而第二个可以直接就是DataStream。之后仍旧可以传入多个参数，用来指定表中的字段
tableEnv.createTemporaryView("EventTable", eventStream,$("timestamp").as("ts"),$("url"));
这样，我们接下来就可以直接在 SQL 中引用表 EventTable 了。
（3）调用 fromChangelogStream ()方法
表环境还提供了一个方法 fromChangelogStream()，可以将一个更新日志流转换成表。这个方法要求流中的数据类型只能是 Row，而且每一个数据都需要指定当前行的更新类型（RowKind）；所以一般是由连接器帮我们实现的，直接应用比较少见，感兴趣的可以查看官网的文档说明。
6.3 支持的数据类型
前面示例中的 DataStream，流中的数据类型都是定义好的 POJO 类。如果 DataStream 中的类型是简单的基本类型，还可以直接转换成表吗？这就涉及了Table 中支持的数据类型。
整体来看，DataStream 中支持的数据类型，Table 中也是都支持的，只不过在进行转换时需要注意一些细节。
（1）原子类型
在 Flink 中，基础数据类型（Integer、Double、String）和通用数据类型（也就是不可再拆分的数据类型）统一称作“原子类型”。原子类型的 DataStream，转换之后就成了只有一列的Table，列字段（field）的数据类型可以由原子类型推断出。另外，还可以在 fromDataStream() 方法里增加参数，用来重新命名列字段。
StreamTableEnvironment tableEnv = ...; DataStreamLong> stream = ...;
// 将数据流转换成动态表，动态表只有一个字段，重命名为 myLong
Table table = tableEnv.fromDataStream(stream, $("myLong"));
（2）Tuple 类型
当原子类型不做重命名时，默认的字段名就是“f0”，容易想到，这其实就是将原子类型看作了一元组Tuple1 的处理结果。
Table 支持 Flink 中定义的元组类型Tuple，对应在表中字段名默认就是元组中元素的属性名 f0、f1、f2…。所有字段都可以被重新排序，也可以提取其中的一部分字段。字段还可以通过调用表达式的as()方法来进行重命名。
StreamTableEnvironment tableEnv = ...; DataStreamTuple2Long, Integer>> stream = ...;
// 将数据流转换成只包含 f1 字段的表
Table table = tableEnv.fromDataStream(stream, $("f1"));
// 将数据流转换成包含 f0 和 f1 字段的表，在表中 f0 和 f1 位置交换
Table table = tableEnv.fromDataStream(stream, $("f1"), $("f0"));
// 将 f1 字段命名为 myInt，f0 命名为 myLong
Table table = tableEnv.fromDataStream(stream, $("f1").as("myInt"),$("f0").as("myLong"));
（3）POJO 类型
Flink 也支持多种数据类型组合成的“复合类型”，最典型的就是简单 Java 对象（POJO 类型）。由于 POJO 中已经定义好了可读性强的字段名，这种类型的数据流转换成 Table 就显得无比顺畅了。
将 POJO 类型的DataStream 转换成 Table，如果不指定字段名称，就会直接使用原始 POJO类型中的字段名称。POJO 中的字段同样可以被重新排序、提却和重命名，这在之前的例子中已经有过体现。
StreamTableEnvironment tableEnv = ...; DataStreamEvent> stream = ...;
Table table = tableEnv.fromDataStream(stream);
Table table = tableEnv.fromDataStream(stream, $("user"));
Table table = tableEnv.fromDataStream(stream, $("user").as("myUser"),
$("url").as("myUrl"));
（4）Row 类型
Flink 中还定义了一个在关系型表中更加通用的数据类型——行（Row），它是 Table 中数据的基本组织形式。Row 类型也是一种复合类型，它的长度固定，而且无法直接推断出每个字段的类型，所以在使用时必须指明具体的类型信息；我们在创建 Table 时调用的 CREATE 语句就会将所有的字段名称和类型指定，这在 Flink 中被称为表的“模式结构”（Schema）。除此之外，Row 类型还附加了一个属性 RowKind，用来表示当前行在更新操作中的类型。这样， Row 就可以用来表示更新日志流（changelog stream）中的数据，从而架起了 Flink 中流和表的转换桥梁。
所以在更新日志流中，元素的类型必须是 Row，而且需要调用 ofKind()方法来指定更新类型。下面是一个具体的例子：
DataStreamRow> dataStream = env.fromElements(
Row.ofKind(RowKind.INSERT, "Alice", 12),
Row.ofKind(RowKind.INSERT, "Bob", 5),
Row.ofKind(RowKind.UPDATE_BEFORE, "Alice", 12),
Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100));
// 将更新日志流转换为表
Table table = tableEnv.fromChangelogStream(dataStream);
6.4 综合应用示例
现在，我们可以将介绍过的所有 API 整合起来，写出一段完整的代码。同样还是用户的一组点击事件，我们可以查询出某个用户（例如 Alice）点击的 url 列表，也可以统计出每个用户累计的点击次数，这可以用两句 SQL 来分别实现。具体代码如下：
public class TableToStreamExample {
public static void main(String[] args) throws Exception {
// 获取流环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 读取数据源
SingleOutputStreamOperatorEvent> eventStream = env
.fromElements(
new Event("Alice", "./home", 1000L),
new Event("Bob", "./cart", 1000L),
new Event("Alice", "./prod?id=1", 5 * 1000L),
new Event("Cary", "./home", 60 * 1000L),
new Event("Bob", "./prod?id=3", 90 * 1000L),
new Event("Alice", "./prod?id=7", 105 * 1000L)
);
// 获取表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 将数据流转换成表
tableEnv.createTemporaryView("EventTable", eventStream);
// 查询Alice的访问url列表
Table aliceVisitTable = tableEnv.sqlQuery("SELECT url, user FROM EventTable WHERE user = 'Alice'");
// 统计每个用户的点击次数
Table urlCountTable = tableEnv.sqlQuery("SELECT user, COUNT(url) FROM EventTable GROUP BY user");
// 将表转换成数据流，在控制台打印输出
tableEnv.toDataStream(aliceVisitTable).print("alice visit");
tableEnv.toChangelogStream(urlCountTable).print("count");
// 执行程序
env.execute();
}
}
用户 Alice 的点击 url 列表只需要一个简单的条件查询就可以得到，对应的表中只有插入操作，所以我们可以直接调用 toDataStream()将它转换成数据流，然后打印输出。控制台输出的结果如下：
alice visit > +I[./home, Alice]
alice visit > +I[./prod?id=1, Alice]
alice visit > +I[./prod?id=7, Alice]
这里每条数据前缀的+I 就是 RowKind，表示 INSERT（插入）。
而由于统计点击次数时用到了分组聚合，造成结果表中数据会有更新操作，所以在打印输出时需要将表urlCountTable 转换成更新日志流（changelog stream）。控制台输出的结果如下：
count> +I[Alice, 1]
count> +I[Bob, 1]
count> -U[Alice, 1]
count> +U[Alice, 2]
count> +I[Cary, 1]
count> -U[Bob, 1]
count> +U[Bob, 2]
count> -U[Alice, 2]
count> +U[Alice, 3]
这里数据的前缀出现了+I、-U 和+U 三种 RowKind，分别表示 INSERT（插入）、UPDATE_BEFORE（更新前）和 UPDATE_AFTER（更新后）。当收到每个用户的第一次点击事件时，会在表中插入一条数据，例如+I[Alice, 1]、+I[Bob, 1]。而之后每当用户增加一次点击事件，就会带来一次更新操作，更新日志流（changelog stream）中对应会出现两条数据，分别表示之前数据的失效和新数据的生效；例如当Alice 的第二条点击数据到来时，会出现一个-U[Alice, 1]和一个+U[Alice, 2]，表示Alice 的点击个数从 1 变成了 2。
这种表示更新日志的方式，有点像是声明“撤回”了之前的一条数据、再插入一条更新后的数据，所以也叫作“撤回流”（Retract Stream）。关于表到流转换过程的编码方式，我们会在下节进行更深入的讨论。
三、流处理中的表
上一节中介绍了Table API 和SQL 的基本使用方法。我们会发现，在 Flink 中使用表和SQL 基本上跟其它场景是一样的；不过对于表和流的转换，却稍显复杂。当我们将一个 Table 转换成 DataStream 时，有“仅插入流”（Insert-Only Streams）和“更新日志流”（Changelog Streams）两种不同的方式，具体使用哪种方式取决于表中是否存在更新（update）操作。
这种麻烦其实是不可避免的。我们知道，Table API 和 SQL 本质上都是基于关系型表的操作方式；而关系型表（Table）本身是有界的，更适合批处理的场景。所以在 MySQL、Hive 这样的固定数据集中进行查询，使用 SQL 就会显得得心应手。而对于 Flink 这样的流处理框架来说，要处理的是源源不断到来的无界数据流，我们无法等到数据都到齐再做查询，每来一条数据就应该更新一次结果；这时如果一定要使用表和 SQL 进行处理，就会显得有些别扭了， 需要引入一些特殊的概念。
1. 动态表和持续查询
流处理面对的数据是连续不断的，这导致了流处理中的“表”跟我们熟悉的关系型数据库中的表完全不同；而基于表执行的查询操作，也就有了新的含义。
如果我们希望把流数据转换成表的形式，那么这表中的数据就会不断增长；如果进一步基于表执行 SQL 查询，那么得到的结果就不是一成不变的，而是会随着新数据的到来持续更新。
1.1 动态表（Dynamic Tables）
当流中有新数据到来，初始的表中会插入一行；而基于这个表定义的 SQL 查询，就应该在之前的基础上更新结果。这样得到的表就会不断地动态变化，被称为“动态表”（Dynamic Tables）。
动态表是Flink 在Table API 和SQL 中的核心概念，它为流数据处理提供了表和SQL 支持。我们所熟悉的表一般用来做批处理，面向的是固定的数据集，可以认为是“静态表”；而动态表则完全不同，它里面的数据会随时间变化。
其实动态表的概念，我们在传统的关系型数据库中已经有所接触。数据库中的表，其实是一系列 INSERT、UPDATE 和 DELETE 语句执行的结果；在关系型数据库中，我们一般把它称为更新日志流（changelog stream）。如果我们保存了表在某一时刻的快照（snapshot），那么接下来只要读取更新日志流，就可以得到表之后的变化过程和最终结果了。在很多高级关系型数据库（比如 Oracle、DB2）中都有“物化视图”（Materialized Views）的概念，可以用来缓存 SQL 查询的结果；它的更新其实就是不停地处理更新日志流的过程。
Flink 中的动态表，就借鉴了物化视图的思想。
1.2 持续查询（Continuous Query）
动态表可以像静态的批处理表一样进行查询操作。由于数据在不断变化，因此基于它定义的 SQL 查询也不可能执行一次就得到最终结果。这样一来，我们对动态表的查询也就永远不会停止，一直在随着新数据的到来而继续执行。这样的查询就被称作“持续查询”（Continuous Query）。对动态表定义的查询操作，都是持续查询；而持续查询的结果也会是一个动态表。
由于每次数据到来都会触发查询操作，因此可以认为一次查询面对的数据集，就是当前输入动态表中收到的所有数据。这相当于是对输入动态表做了一个“快照”（snapshot），当作有限数据集进行批处理；流式数据的到来会触发连续不断的快照查询，像动画一样连贯起来，就构成了“持续查询”。
如图所示，描述了持续查询的过程。这里我们也可以清晰地看到流、动态表和持续查询的关系：
持续查询的步骤如下：
（1） 流（stream）被转换为动态表（dynamic table）；
（2） 对动态表进行持续查询（continuous query），生成新的动态表；
（3） 生成的动态表被转换成流。
这样，只要API 将流和动态表的转换封装起来，我们就可以直接在数据流上执行 SQL 查询，用处理表的方式来做流处理了。
2. 将流转换成动态表
为了能够使用 SQL 来做流处理，我们必须先把流（stream）转换成动态表。当然，之前在讲解基本 API 时，已经介绍过代码中的 DataStream 和 Table 如何转换；现在我们则要抛开具体的数据类型，从原理上理解流和动态表的转换过程。
如果把流看作一张表，那么流中每个数据的到来，都应该看作是对表的一次插入（Insert） 操作，会在表的末尾添加一行数据。因为流是连续不断的，而且之前的输出结果无法改变、只能在后面追加；所以我们其实是通过一个只有插入操作（insert-only）的更新日志（changelog） 流，来构建一个表。
为了更好地说明流转换成动态表的过程，我们例子来做分析说明。
// 获取流环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1);
// 读取数据源
SingleOutputStreamOperatorEvent> eventStream = env
.fromElements(
new Event("Alice", "./home", 1000L),
new Event("Bob", "./cart", 1000L),
new Event("Alice", "./prod?id=1", 5 * 1000L), new Event("Cary", "./home", 60 * 1000L),
new Event("Bob", "./prod?id=3", 90 * 1000L),
new Event("Alice", "./prod?id=7", 105 * 1000L)
);
// 获取表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 将数据流转换成表
tableEnv.createTemporaryView("EventTable", eventStream, $("user"), $("url"),$("timestamp").as("ts"));
// 统计每个用户的点击次数
Table urlCountTable = tableEnv.sqlQuery("SELECT user, COUNT(url) as cnt FROM EventTable GROUP BY user");
// 将表转换成数据流，在控制台打印输出
tableEnv.toChangelogStream(urlCountTable).print("count");
// 执行程序
env.execute();
我们现在的输入数据，就是用户在网站上的点击访问行为，数据类型被包装为 POJO 类型Event。我们将它转换成一个动态表，注册为EventTable。表中的字段定义如下：
[
user: VARCHAR, // 用户名
url: VARCHAR, // 用户访问的 URL
ts: BIGINT // 时间戳
]
如图所示，当用户点击事件到来时，就对应着动态表中的一次插入（Insert）操作， 每条数据就是表中的一行；随着插入更多的点击事件，得到的动态表将不断增长。
3. 用 SQL 持续查询
3.1 更新（Update）查询
我们在代码中定义了一个 SQL 查询。
Table urlCountTable = tableEnv.sqlQuery("SELECT user, COUNT(url) as cnt FROM EventTable GROUP BY user");
这个查询很简单，主要是分组聚合统计每个用户的点击次数。我们把原始的动态表注册为EventTable，经过查询转换后得到 urlCountTable；这个结果动态表中包含两个字段，具体定义如下：
[
user: VARCHAR, // 用户名
cnt: BIGINT // 用户访问 url 的次数
]
当原始动态表不停地插入新的数据时，查询得到的 urlCountTable 会持续地进行更改。由于 count 数量可能会叠加增长，因此这里的更改操作可以是简单的插入（Insert），也可以是对之前数据的更新（Update）。换句话说，用来定义结果表的更新日志（changelog）流中，包含了 INSERT 和UPDATE 两种操作。这种持续查询被称为更新查询（Update Query），更新查询得到的结果表如果想要转换成DataStream，必须调用 toChangelogStream()方法。
具体步骤解释如下：
（1） 当查询启动时，原始动态表 EventTable 为空；
（2） 当第一行 Alice 的点击数据插入EventTable 表时，查询开始计算结果表，urlCountTable中插入一行数据[Alice，1]。
（3） 当第二行 Bob 点击数据插入 EventTable 表时，查询将更新结果表并插入新行[Bob，1]。
（4） 第三行数据到来，同样是 Alice 的点击事件，这时不会插入新行，而是生成一个针对已有行的更新操作。这样，结果表中第一行[Alice，1]就更新为[Alice，2]。
（5） 当第四行 Cary 的点击数据插入到 EventTable 表时，查询将第三行[Cary，1]插入到结果表中。
3.2 追加（Append）查询
上面的例子中，查询过程用到了分组聚合，结果表中就会产生更新操作。如果我们执行一个简单的条件查询，结果表中就会像原始表EventTable 一样，只有插入（Insert）操作了。
Table aliceVisitTable = tableEnv.sqlQuery("SELECT url, user FROM EventTable WHERE user = 'Cary'");
这样的持续查询，就被称为追加查询（Append Query），它定义的结果表的更新日志（changelog）流中只有 INSERT 操作。追加查询得到的结果表，转换成 DataStream 调用方法没有限制，可以直接用 toDataStream()，也可以像更新查询一样调用 toChangelogStream()。
这样看来，我们似乎可以总结一个规律：只要用到了聚合，在之前的结果上有叠加，就会产生更新操作，就是一个更新查询。但事实上，更新查询的判断标准是结果表中的数据是否会有 UPDATE 操作，如果聚合的结果不再改变，那么同样也不是更新查询。
什么时候聚合的结果会保持不变呢？一个典型的例子就是窗口聚合。
我们考虑开一个滚动窗口，统计每一小时内所有用户的点击次数，并在结果表中增加一个endT 字段，表示当前统计窗口的结束时间。这时结果表的字段定义如下：
[
user: VARCHAR, // 用户名
endT: TIMESTAMP, // 窗口结束时间
cnt: BIGINT // 用户访问 url 的次数
]
与之前的分组聚合一样，当原始动态表不停地插入新的数据时，查询得到的结果 result 会持续地进行更改。比如时间戳在 12:00:00 到 12:59:59 之间的有四条数据，其中 Alice 三次点击、Bob 一次点击；所以当水位线达到 13:00:00 时窗口关闭，输出到结果表中的就是新增两条数据[Alice, 13:00:00, 3]和[Bob, 13:00:00, 1]。同理，当下一小时的窗口关闭时， 也会将统计结果追加到 result 表后面，而不会更新之前的数据。
所以我们发现，由于窗口的统计结果是一次性写入结果表的，所以结果表的更新日志流中只会包含插入 INSERT 操作，而没有更新 UPDATE 操作。所以这里的持续查询，依然是一个追加（Append）查询。结果表 result 如果转换成 DataStream，可以直接调用 toDataStream()方法。
需要注意的是，由于涉及时间窗口，我们还需要为事件时间提取时间戳和生成水位线。完整代码如下：
import com.atguigu.chapter05.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.bridge.java.StreamTableEnvironment;
import static org.apache.flink.table.api.Expressions.$;
public class CumulateWindowExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 读取数据源，并分配时间戳、生成水位线
SingleOutputStreamOperatorEvent> eventStream = env
.fromElements(
new Event("Alice", "./home", 1000L),
new Event("Bob", "./cart", 1000L),
new Event("Alice", "./prod?id=1", 25 * 60 * 1000L),
new Event("Alice", "./prod?id=4", 55 * 60 * 1000L),
new Event("Bob", "./prod?id=5", 3600 * 1000L + 60 * 1000L),
new Event("Cary", "./home", 3600 * 1000L + 30 * 60 * 1000L),
new Event("Cary", "./prod?id=7", 3600 * 1000L + 59 * 60 * 1000L)
)
.assignTimestampsAndWatermarks(
WatermarkStrategy.Event>forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssignerEvent>() {
@Override
public long extractTimestamp(Event element, long recordTimestamp) {
return element.timestamp;
}
})
);
// 创建表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 将数据流转换成表，并指定时间属性
Table eventTable = tableEnv.fromDataStream(
eventStream,
$("user"),
$("url"),
$("timestamp").rowtime().as("ts")
);
// 为方便在SQL中引用，在环境中注册表EventTable
tableEnv.createTemporaryView("EventTable", eventTable);
// 设置累积窗口，执行SQL统计查询
Table result = tableEnv.sqlQuery(
"SELECT " +
"user, " +
"window_end AS endT, " + // 窗口结束时间
"COUNT(url) AS cnt " + // 统计 url 访问次数
"FROM TABLE( " +
"TUMBLE( TABLE EventTable, " + // 1 小时滚动窗口
"DESCRIPTOR(ts), " + "INTERVAL '1' HOUR)) " +
"GROUP BY user, window_start, window_end "
);
tableEnv.toDataStream(result).print();
env.execute();
}
}
运行结果如下：
+I[Alice, 1970-01-01T01:00, 3]
+I[Bob, 1970-01-01T01:00, 1]
+I[Cary, 1970-01-01T02:00, 2]
+I[Bob, 1970-01-01T02:00, 1]
可以看到，所有输出结果都以+I 为前缀，表示都是以 INSERT 操作追加到结果表中的； 这是一个追加查询，所以我们直接使用 toDataStream()转换成流是没有问题的。这里输出的window_end 是一个TIMESTAMP 类型；由于我们直接以一个长整型数作为事件发生的时间戳， 所以可以看到对应的都是 1970 年 1 月 1 日的时间。
关于Table API 和 SQL 中窗口和聚合查询的使用，我们会在后面详细讲解。
3.3 查询限制
在实际应用中，有些持续查询会因为计算代价太高而受到限制。所谓的“代价太高”，可能是由于需要维护的状态持续增长，也可能是由于更新数据的计算太复杂。
- 状态大小
用持续查询做流处理，往往会运行至少几周到几个月；所以持续查询处理的数据总量可能非常大。例如我们之前举的更新查询的例子，需要记录每个用户访问url 的次数。如果随着时间的推移用户数越来越大，那么要维护的状态也将逐渐增长，最终可能会耗尽存储空间导致查询失败。
SELECT user, COUNT(url) FROM clicks
GROUP BY user;
- 更新计算
对于有些查询来说，更新计算的复杂度可能很高。每来一条新的数据，更新结果的时候可能需要全部重新计算，并且对很多已经输出的行进行更新。一个典型的例子就是 RANK()函数， 它会基于一组数据计算当前值的排名。例如下面的 SQL 查询，会根据用户最后一次点击的时间为每个用户计算一个排名。当我们收到一个新的数据，用户的最后一次点击时间（lastAction） 就会更新，进而所有用户必须重新排序计算一个新的排名。当一个用户的排名发生改变时，被他超过的那些用户的排名也会改变；这样的更新操作无疑代价巨大，而且还会随着用户的增多越来越严重。
SELECT user, RANK() OVER (ORDER BY lastAction) FROM (
SELECT user, MAX(ts) AS lastAction FROM EventTable GROUP BY user
);
这样的查询操作，就不太适合作为连续查询在流处理中执行。这里 RANK()的使用要配合一个OVER 子句，这是所谓的“开窗聚合”，我们会在后面展开介绍。
4. 将动态表转换为流
与关系型数据库中的表一样，动态表也可以通过插入（Insert）、更新（Update）和删除（Delete）操作，进行持续的更改。将动态表转换为流或将其写入外部系统时，就需要对这些更改操作进行编码，通过发送编码消息的方式告诉外部系统要执行的操作。在 Flink 中，Table API 和 SQL 支持三种编码方式：
- 仅追加（Append-only）流
仅通过插入（Insert）更改来修改的动态表，可以直接转换为“仅追加”流。这个流中发出的数据，其实就是动态表中新增的每一行。 - 撤回（Retract）流
撤回流是包含两类消息的流，添加（add）消息和撤回（retract）消息。
具体的编码规则是：INSERT 插入操作编码为 add 消息；DELETE 删除操作编码为 retract 消息；而 UPDATE 更新操作则编码为被更改行的 retract 消息，和更新后行（新行）的 add 消息。这样，我们可以通过编码后的消息指明所有的增删改操作，一个动态表就可以转换为撤回流了。
可以看到，更新操作对于撤回流来说，对应着两个消息：之前数据的撤回（删除）和新数据的插入。
这里我们用+代表 add 消息（对应插入 INSERT 操作），用-代表 retract 消息（对应删除 DELETE 操作）；当 Alice 的第一个点击事件到来时，结果表新增一条数据[Alice, 1]；而当 Alice 的第二个点击事件到来时，结果表会将[Alice, 1]更新为[Alice, 2]，对应的编码就是删除[Alice, 1]、插入[Alice, 2]。这样当一个外部系统收到这样的两条消息时，就知道是要对Alice 的点击统计次数进行更新了。
- 更新插入（Upsert）流
更新插入流中只包含两种类型的消息：更新插入（upsert）消息和删除（delete）消息。所谓的“upsert”其实是“update”和“insert”的合成词，所以对于更新插入流来说，INSERT 插入操作和 UPDATE 更新操作，统一被编码为upsert 消息；而DELETE 删除操作则被编码为delete消息。
既然更新插入流中不区分插入（insert）和更新（update），那我们自然会想到一个问题：如果希望更新一行数据时，怎么保证最后做的操作不是插入呢？
这就需要动态表中必须有唯一的键（key）。通过这个 key 进行查询，如果存在对应的数据就做更新（update），如果不存在就直接插入（insert）。这是一个动态表可以转换为更新插入流的必要条件。当然，收到这条流中数据的外部系统，也需要知道这唯一的键（key），这样才能正确地处理消息。
可以看到，更新插入流跟撤回流的主要区别在于，更新（update）操作由于有 key 的存在， 只需要用单条消息编码就可以，因此效率更高。
需要注意的是，在代码里将动态表转换为 DataStream 时，只支持仅追加（append-only） 和撤回（retract）流，我们调用 toChangelogStream()得到的其实就是撤回流；这也很好理解， DataStream 中并没有 key 的定义，所以只能通过两条消息一减一增来表示更新操作。而连接到外部系统时，则可以支持不同的编码方法，这取决于外部系统本身的特性。
参考资料
Word版：https://download.csdn.net/download/mengxianglong123/85035166
PDF版：https://download.csdn.net/download/mengxianglong123/85035172
原文链接：Flink Table API和SQL（上）本文转载自落花雨时，原文链接：https://blog.csdn.net/mengxianglong123/article/details/124041915。
