Flink (十一) --------- Table API 和 SQL

ゞ浴缸里的玫瑰 2024-03-24 16:44 23阅读 0赞

#### 目录 ####

*  一、快速上手
 *   *  1. 需要引入的依赖
     *  2. 一个简单示例
 *  二、基本 API
 *   *  1. 程序架构
     *  2. 创建表环境
     *  3. 创建表
     *  4. 表的查询
     *  5. 输出表
     *  6. 表和流的转换
 *  三、流处理中的表
 *   *  1. 动态表和持续查询
     *  2. 将流转换成动态表
     *  3. 用 SQL 持续查询
     *  3. 将动态表转换为流
 *  四、时间属性和窗口
 *   *  1. 事件时间
     *  2. 处理时间
     *  3. 窗口（Window）
 *  五、聚合 (Aggregation) 查询
 *   *  1. 分组聚合
     *  2. 窗口聚合
     *  3. 开窗（Over）聚合
     *  4. 应用实例 —— Top N
 *  六、联结（Join）查询
 *   *  1. 常规联结查询
     *  2. 间隔联结查询
 *  七、函数
 *   *  1. 系统函数
     *  2. 自定义函数（UDF）
 *  八、SQL 客户端
 *  九、连接到外部系统
 *   *  1. Kafka
     *  2. 文件系统
     *  3. JDBC
     *  4. Elasticsearch
     *  5. HBase
     *  6. Hive
 *  十、总结

--------------------

如图 11-1 所示，在 Flink 提供的多层级 API 中，核心是 DataStream API，这是我们开发流处理应用的基本途径；底层则是所谓的处理函数（process function），可以访问事件的时间信息、注册定时器、自定义状态，进行有状态的流处理。DataStream API 和处理函数比较通用，有了这些 API，理论上我们就可以实现所有场景的需求了。

不过在企业实际应用中，往往会面对大量类似的处理逻辑，所以一般会将底层 API 包装成更加具体的应用级接口。怎样的接口风格最容易让大家接收呢？作为大数据工程师，我们最为熟悉的数据统计方式，当然就是写 SQL 了。

SQL 是结构化查询语言（Structured Query Language）的缩写，是我们对关系型数据库进行查询和修改的通用编程语言。在关系型数据库中，数据是以表（table）的形式组织起来的，所以也可以认为 SQL 是用来对表进行处理的工具语言。无论是传统架构中进行数据存储的MySQL、PostgreSQL，还是大数据应用中的 Hive，都少不了 SQL 的身影；而 Spark 作为大数据处理引擎，为了更好地支持在 Hive 中的 SQL 查询，也提供了 Spark SQL 作为入口。

Flink 同样提供了对于“表”处理的支持，这就是更高层级的应用 API，在 Flink 中被称为Table API 和 SQL。Table API 顾名思义，就是基于“表”（Table）的一套 API，它是内嵌在 Java、Scala 等语言中的一种声明式领域特定语言（DSL），也就是专门为处理表而设计的；在此基础上，Flink 还基于 Apache Calcite 实现了对 SQL 的支持。这样一来，我们就可以在 Flink 程序中直接写 SQL 来实现处理需求了。

![在这里插入图片描述][326fc23db2aa4ebabfc3f28397626a65.png]  
在 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 调用可以随时关注官网的更新变化。

--------------------

## 一、快速上手 ##

如果我们对关系型数据库和 SQL 非常熟悉，那么 Table API 和 SQL 的使用其实非常简单：只要得到一个“表”(Table) ，然后对它调用 Table API，或者直接写 SQL 就可以了。接下来我们就以一个非常简单的例子上手，初步了解一下这种高层级 API 的使用方法。

### 1. 需要引入的依赖 ###

我们想要在代码中使用 Table API，必须引入相关的依赖。

<dependency>
    	<groupId>org.apache.flink</groupId>
    	<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.flink</groupId>
    	<artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
    	<version>${flink.version}</version>
    </dependency>
    <dependency>
    	<groupId>org.apache.flink</groupId>
    	<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.flink</groupId>
    	<artifactId>flink-table-common</artifactId>
    	<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，就可以得到想要提取的数据了。

在代码中具体实现如下：

public class TableExample {
        
    	public static void main(String[] args) throws Exception {
        
    		// 获取流执行环境
    		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    		env.setParallelism(1);
    		// 读取数据源
    		SingleOutputStreamOperator<Event> 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);
    		// 将数据流转换成表
    		Table eventTable = tableEnv.fromDataStream(eventStream);
    		// 用执行 SQL 的方式提取数据
    		Table visitTable = tableEnv.sqlQuery("select url, user from " + eventTable);
    		// 将表转换成数据流，打印输出
    		tableEnv.toDataStream(visitTable).print();
    		// 执行程序
    		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 ##

通过上节中的简单示例，我们已经对 Table API 和 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）两种。

**1. 连接器表（Connector Tables）**

最直观的创建表的方式，就是通过连接器（connector）连接到一个外部系统，然后定义出对应的表结构。例如我们可以连接到 Kafka 或者文件系统，将存储在这些外部系统的数据以“表”的形式定义出来，这样对表的读写就可以通过连接器转换成对外部系统的读写了。当我们在表环境中读取这张表，连接器就会从外部系统读取数据并进行转换；而当我们向这张表写入数据，连接器就会将数据输出（Sink）到外部系统中。

在代码中，我们可以调用表环境的 executeSql()方法，可以传入一个 DDL 作为参数执行 SQL 操作。这里我们传入一个 CREATE 语句进行表的创建，并通过 WITH 关键字指定连接到外部系统的连接器：

tableEnv.executeSql("CREATE [TEMPORARY] TABLE MyTable ... WITH ( 'connector' = ... )");

这里的 TEMPORARY 关键字可以省略。关于连接器的具体定义，我们会在 11.8 节中展开讲解。

这里没有定义 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 作为前缀。

**2. 虚拟表（Virtual Tables）**

在环境中注册之后，我们就可以在 SQL 中直接使用这张表进行查询转换了。

Table newTable = tableEnv.sqlQuery("SELECT ... FROM MyTable... ");

这里调用了表环境的 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. 表的查询 ###

创建好了表，接下来自然就是对表进行查询转换了。对一个表的查询（Query）操作，就对应着流数据的转换(Transform) 处理。  
Flink 为我们提供了两种查询方式：SQL 和 Table API。

**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' "
    );

**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。

**3. 两种 API 的结合使用**

可以发现，无论是调用 Table API 还是执行 SQL，得到的结果都是一个 Table 对象；所以这两种 API 的查询可以很方便地结合在一起。

（1）无论是那种方式得到的 Table 对象，都可以继续调用 Table API 进行查询转换；  
（2）如果想要对一个表执行 SQL 操作（用 FROM 关键字引用），必须先在环境中对它进行注册。所以我们可以通过创建虚拟表的方式实现两者的转换：

tableEnv.createTemporaryView("MyTable", myTable);

注意：这里的第一个参数"MyTable"是注册的表名，而第二个参数 myTable 是 Java 中的Table 对象。  
另外要说明的是，在 11.1.2 小节的简单示例中，我们并没有将 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，有不同的连接器对它进行了实现。  
关于不同外部系统的连接器，我们会在 11.8 节展开介绍。  
这里可以发现，我们在环境中注册的“表”，其实在写入数据的时候就对应着一个 TableSink。

### 6. 表和流的转换 ###

从创建表环境开始，历经表的创建、查询转换和输出，我们已经可以使用 Table API 和 SQL 进行完整的流处理了。不过在应用的开发过程中，我们测试业务逻辑一般不会直接将结果直接写入到外部系统，而是在本地控制台打印输出。对于 DataStream 这非常容易，直接调用 print() 方法就可以看到结果数据流的内容了；但对于 Table 就比较悲剧——它没有提供 print()方法。这该怎么办呢？  
在 Flink 中我们可以将 Table 再转换成 DataStream，然后进行打印输出。这就涉及了表和流的转换。

**1. 将表（Table）转换成流（DataStream）**

(1) 调用 toDataStream()方法  
将一个 Table 对象转换成 DataStream 非常简单，只要直接调用表环境的方法 toDataStream()就可以了。例如，我们可以将 11.2.4 小节经查询转换得到的表 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.toDataStream(urlCountTable).print();

与“更新日志流”（Changelog Streams）对应的，是那些只做了简单转换、没有进行聚合统计的表，例如前面提到的 maryClickTable。它们的特点是数据只会插入、不会更新，所以也被叫作“仅插入流”（Insert-Only Streams）。

**2. 将流（DataStream）转换成表（Table）**

(1) 调用 fromDataStream()方法

想要将一个 DataStream 转换成表也很简单，可以通过调用表环境的 fromDataStream()方法来实现，返回的就是一个 Table 对象。例如，我们可以直接将事件流 eventStream 转换成一个表：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 获取表环境
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    // 读取数据源
    SingleOutputStreamOperator<Event> 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）；所以一般是由连接器帮我们实现的，直接应用比较少见，感兴趣的读者可以查看官网的文档说明。

**3. 支持的数据类型**

前面示例中的 DataStream，流中的数据类型都是定义好的 POJO 类。如果 DataStream 中的类型是简单的基本类型，还可以直接转换成表吗？这就涉及了 Table 中支持的数据类型。整体来看，DataStream 中支持的数据类型，Table 中也是都支持的，只不过在进行转换时需要注意一些细节。

(1) 原子类型

在 Flink 中，基础数据类型（Integer、Double、String）和通用数据类型（也就是不可再拆分的数据类型）统一称作“原子类型”。原子类型的 DataStream，转换之后就成了只有一列的Table，列字段（field）的数据类型可以由原子类型推断出。另外，还可以在 fromDataStream() 方法里增加参数，用来重新命名列字段。

StreamTableEnvironment tableEnv = ...;
    DataStream<Long> stream = ...;
    // 将数据流转换成动态表，动态表只有一个字段，重命名为 myLong
    Table table = tableEnv.fromDataStream(stream, $("myLong"));

(2) Tuple 类型

当原子类型不做重命名时，默认的字段名就是“f0”，容易想到，这其实就是将原子类型看作了一元组 Tuple1 的处理结果。

Table 支持 Flink 中定义的元组类型 Tuple，对应在表中字段名默认就是元组中元素的属性名 f0、f1、f2…。所有字段都可以被重新排序，也可以提取其中的一部分字段。字段还可以通过调用表达式的 as()方法来进行重命名。

StreamTableEnvironment tableEnv = ...;
    DataStream<Tuple2<Long, 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 = ...;
    DataStream<Event> stream = ...;
    Table table = tableEnv.fromDataStream(stream);
    Table table = tableEnv.fromDataStream(stream, $("user"));
    Table table = tableEnv.fromDataStream(stream, $("user").as("myUser"), $("url").as("myUrl"));

(5) Row 类型

Flink 中还定义了一个在关系型表中更加通用的数据类型——行（Row），它是 Table 中数据的基本组织形式。Row 类型也是一种复合类型，它的长度固定，而且无法直接推断出每个字段的类型，所以在使用时必须指明具体的类型信息；我们在创建 Table 时调用的 CREATE 语句就会将所有的字段名称和类型指定，这在 Flink 中被称为表的“模式结构”（Schema）。除此之外，Row 类型还附加了一个属性 RowKind，用来表示当前行在更新操作中的类型。这样，Row 就可以用来表示更新日志流（changelog stream）中的数据，从而架起了 Flink 中流和表的  
转换桥梁。

所以在更新日志流中，元素的类型必须是 Row，而且需要调用 ofKind()方法来指定更新类型。下面是一个具体的例子：

DataStream<Row> 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);

**4. 综合应用示例**

现在，我们可以将介绍过的所有 API 整合起来，写出一段完整的代码。同样还是用户的一组点击事件，我们可以查询出某个用户（例如 Alice）点击的 url 列表，也可以统计出每个用户累计的点击次数，这可以用两句 SQL 来分别实现。具体代码如下：

public class TableToStreamExample {
        
    	public static void main(String[] args) throws Exception {
        
    		// 获取流环境
    		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    		env.setParallelism(1);
    		// 读取数据源
    		SingleOutputStreamOperator<Event> 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 进行处理，就会显得有些别扭了，需要引入一些特殊的概念。

我们可以将关系型表/SQL 与流处理做一个对比，如表 11-1 所示。

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

可以看到，其实关系型表和 SQL，主要就是针对批处理设计的，这和流处理有着天生的隔阂。既然“八字不合”，那 Flink 中的 Table API 和 SQL 又是怎样做流处理的呢？接下来我们就来深入探讨一下流处理中表的概念。

### 1. 动态表和持续查询 ###

流处理面对的数据是连续不断的，这导致了流处理中的“表”跟我们熟悉的关系型数据库中的表完全不同；而基于表执行的查询操作，也就有了新的含义。

如果我们希望把流数据转换成表的形式，那么这表中的数据就会不断增长；如果进一步基于表执行 SQL 查询，那么得到的结果就不是一成不变的，而是会随着新数据的到来持续更新。

**1. 动态表（Dynamic Tables）**

当流中有新数据到来，初始的表中会插入一行；而基于这个表定义的 SQL 查询，就应该在之前的基础上更新结果。这样得到的表就会不断地动态变化，被称为“动态表”（Dynamic Tables）。

动态表是Flink在Table API和SQL中的核心概念，它为流数据处理提供了表和SQL支持。我们所熟悉的表一般用来做批处理，面向的是固定的数据集，可以认为是“静态表”；而动态表则完全不同，它里面的数据会随时间变化。其实动态表的概念，我们在传统的关系型数据库中已经有所接触。数据库中的表，其实是一系列 INSERT、UPDATE 和 DELETE 语句执行的结果；在关系型数据库中，我们一般把它称为更新日志流（changelog stream）。如果我们保存了表在某一时刻的快照（snapshot），那么接下来只要读取更新日志流，就可以得到表之后的变化过程和最终结果了。在很多高级关系型数据库（比如 Oracle、DB2）中都有“物化视图”（Materialized Views）的概念，可以用来缓存 SQL 查询的结果；它的更新其实就是不停地处理更新日志流的过程。Flink 中的动态表，就借鉴了物化视图的思想。

**2. 持续查询（Continuous Query）**

动态表可以像静态的批处理表一样进行查询操作。由于数据在不断变化，因此基于它定义的 SQL 查询也不可能执行一次就得到最终结果。这样一来，我们对动态表的查询也就永远不会停止，一直在随着新数据的到来而继续执行。这样的查询就被称作“持续查询”（Continuous Query）。对动态表定义的查询操作，都是持续查询；而持续查询的结果也会是一个动态表。

由于每次数据到来都会触发查询操作，因此可以认为一次查询面对的数据集，就是当前输入动态表中收到的所有数据。这相当于是对输入动态表做了一个“快照”（snapshot），当作有限数据集进行批处理；流式数据的到来会触发连续不断的快照查询，像动画一样连贯起来，就构成了“持续查询”。

如图 11-2 所示，描述了持续查询的过程。这里我们也可以清晰地看到流、动态表和持续查询的关系：

![在这里插入图片描述][156cb6d99824400692300ea032f2ffd7.png]  
持续查询的步骤如下：  
（1）流（stream）被转换为动态表（dynamic table）；  
（2）对动态表进行持续查询（continuous query），生成新的动态表；  
（3）生成的动态表被转换成流。

这样，只要 API 将流和动态表的转换封装起来，我们就可以直接在数据流上执行 SQL 查询，用处理表的方式来做流处理了。

### 2. 将流转换成动态表 ###

为了能够使用 SQL 来做流处理，我们必须先把流（stream）转换成动态表。当然，之前在讲解基本 API 时，已经介绍过代码中的 DataStream 和 Table 如何转换；现在我们则要抛开具体的数据类型，从原理上理解流和动态表的转换过程。

如果把流看作一张表，那么流中每个数据的到来，都应该看作是对表的一次插入（Insert）操作，会在表的末尾添加一行数据。因为流是连续不断的，而且之前的输出结果无法改变、只能在后面追加；所以我们其实是通过一个只有插入操作（insert-only）的更新日志（changelog）流，来构建一个表。

为了更好地说明流转换成动态表的过程，我们还是用 11.2 节中举的例子来做分析说明。

// 获取流环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    // 读取数据源
    SingleOutputStreamOperator<Event> 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 FROMEventTable GROUP BY user");
    // 将表转换成数据流，在控制台打印输出
    tableEnv.toChangelogStream(urlCountTable).print("count"); 
    // 执行程序
    env.execute();

我们现在的输入数据，就是用户在网站上的点击访问行为，数据类型被包装为 POJO 类型 Event。我们将它转换成一个动态表，注册为 EventTable。表中的字段定义如下：

[
    	user: VARCHAR, // 用户名
    	url: VARCHAR, // 用户访问的 URL
    	ts: BIGINT // 时间戳
    ]

如图 11-3 所示，当用户点击事件到来时，就对应着动态表中的一次插入（Insert）操作，每条数据就是表中的一行；随着插入更多的点击事件，得到的动态表将不断增长。

![在这里插入图片描述][ea260af2da38450aa9af18e680e51a9e.png]

### 3. 用 SQL 持续查询 ###

**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 的次数
    ]

如图 11-4 所示，当原始动态表不停地插入新的数据时，查询得到的 urlCountTable 会持续地进行更改。由于 count 数量可能会叠加增长，因此这里的更改操作可以是简单的插入（Insert），也可以是对之前数据的更新（Update）。换句话说，用来定义结果表的更新日志（changelog）流中，包含了 INSERT 和 UPDATE 两种操作。这种持续查询被称为更新查询（Update Query），更新查询得到的结果表如果想要转换成 DataStream，必须调用 toChangelogStream()方法。

![在这里插入图片描述][4b7a9db60cb74b5fb7beaf0be7b3d397.png]  
具体步骤解释如下：  
（1）当查询启动时，原始动态表 EventTable 为空；  
（2）当第一行 Alice 的点击数据插入 EventTable 表时，查询开始计算结果表，urlCountTable中插入一行数据\[Alice，1\]。  
（3）当第二行 Bob 点击数据插入 EventTable 表时，查询将更新结果表并插入新行\[Bob，1\]。  
（4）第三行数据到来，同样是 Alice 的点击事件，这时不会插入新行，而是生成一个针对已有行的更新操作。这样，结果表中第一行\[Alice，1\]就更新为\[Alice，2\]。  
（5）当第四行 Cary 的点击数据插入到 EventTable 表时，查询将第三行\[Cary，1\]插入到结果表中。

**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 的次数
    ]

如图 11-5 所示，与之前的分组聚合一样，当原始动态表不停地插入新的数据时，查询得到的结果 result 会持续地进行更改。比如时间戳在 12:00:00 到 12:59:59 之间的有四条数据，其中 Alice 三次点击、Bob 一次点击；所以当水位线达到 13:00:00 时窗口关闭，输出到结果表中的就是新增两条数据\[Alice, 13:00:00, 3\]和\[Bob, 13:00:00, 1\]。同理，当下一小时的窗口关闭时，也会将统计结果追加到 result 表后面，而不会更新之前的数据。

![在这里插入图片描述][70ade69d48174394808c35cd5f375d5a.png]  
所以我们发现，由于窗口的统计结果是一次性写入结果表的，所以结果表的更新日志流中只会包含插入 INSERT 操作，而没有更新 UPDATE 操作。所以这里的持续查询，依然是一个追加（Append）查询。结果表 result 如果转换成 DataStream，可以直接调用 toDataStream()方法。

需要注意的是，由于涉及时间窗口，我们还需要为事件时间提取时间戳和生成水位线。完整代码如下：

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 AppendQueryExample {
        
    	public static void main(String[] args) throws Exception {
        
    		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    		env.setParallelism(1);
    		// 读取数据源，并分配时间戳、生成水位线
    		SingleOutputStreamOperator<Event> 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 SerializableTimestampAssigner<Event>() {
        
    							@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") 
    			// 将 timestamp 指定为事件时间，并命名为 ts
    		);
    		// 为方便在 SQL 中引用，在环境中注册表 EventTable
    		tableEnv.createTemporaryView("EventTable", eventTable);
    		// 设置 1 小时滚动窗口，执行 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. 查询限制**

在实际应用中，有些持续查询会因为计算代价太高而受到限制。所谓的“代价太高”，可能是由于需要维护的状态持续增长，也可能是由于更新数据的计算太复杂。

① 状态大小  
用持续查询做流处理，往往会运行至少几周到几个月；所以持续查询处理的数据总量可能非常大。例如我们之前举的更新查询的例子，需要记录每个用户访问 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 子句，这是所谓的“开窗聚合”，我们会在 11.5 节展开介绍。

### 3. 将动态表转换为流 ###

与关系型数据库中的表一样，动态表也可以通过插入（Insert）、更新（Update）和删除（Delete）操作，进行持续的更改。将动态表转换为流或将其写入外部系统时，就需要对这些更改操作进行编码，通过发送编码消息的方式告诉外部系统要执行的操作。在 Flink 中，Table API 和 SQL  
支持三种编码方式：

*  仅追加（Append-only）流  
    仅通过插入（Insert）更改来修改的动态表，可以直接转换为“仅追加”流。这个流中发  
    出的数据，其实就是动态表中新增的每一行。
 *  撤回（Retract）流  
    撤回流是包含两类消息的流，添加（add）消息和撤回（retract）消息。具体的编码规则是：INSERT 插入操作编码为 add 消息；DELETE 删除操作编码为 retract消息；而 UPDATE 更新操作则编码为被更改行的 retract 消息，和更新后行（新行）的 add 消息。这样，我们可以通过编码后的消息指明所有的增删改操作，一个动态表就可以转换为撤回流了。

可以看到，更新操作对于撤回流来说，对应着两个消息：之前数据的撤回（删除）和新数据的插入。

如图 11-6 所示，显示了将动态表转换为撤回流的过程。

![在这里插入图片描述][63a594dee3254179aa07c400147a9f77.png]  
这里我们用+代表 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），这样才能正确地处理消息。

如图 11-7 所示，显示了将动态表转换为更新插入流的过程。

![在这里插入图片描述][df4f8d6c24b74bb9a5f9e1006e982d52.png]  
可以看到，更新插入流跟撤回流的主要区别在于，更新（update）操作由于有 key 的存在，只需要用单条消息编码就可以，因此效率更高。

需要注意的是，在代码里将动态表转换为 DataStream 时，只支持仅追加（append-only）和撤回（retract）流，我们调用 toChangelogStream()得到的其实就是撤回流；这也很好理解，DataStream 中并没有 key 的定义，所以只能通过两条消息一减一增来表示更新操作。而连接到外部系统时，则可以支持不同的编码方法，这取决于外部系统本身的特性。

## 四、时间属性和窗口 ##

基于时间的操作（比如时间窗口），需要定义相关的时间语义和时间数据来源的信息。在Table API 和 SQL 中，会给表单独提供一个逻辑上的时间字段，专门用来在表处理程序中指示时间。

所以所谓的时间属性（time attributes），其实就是每个表模式结构（schema）的一部分。它可以在创建表的 DDL 里直接定义为一个字段，也可以在 DataStream 转换成表时定义。一旦定义了时间属性，它就可以作为一个普通字段引用，并且可以在基于时间的操作中使用。时间属性的数据类型为 TIMESTAMP，它的行为类似于常规时间戳，可以直接访问并且进行计算。

按照时间语义的不同，我们可以把时间属性的定义分成事件时间（event time）和处理时间（processing time）两种情况。

### 1. 事件时间 ###

我们在实际应用中，最常用的就是事件时间。在事件时间语义下，允许表处理程序根据每个数据中包含的时间戳（也就是事件发生的时间）来生成结果。  
事件时间语义最大的用途就是处理乱序事件或者延迟事件的场景。我们通过设置水位线（watermark）来表示事件时间的进展，而水位线可以根据数据的最大时间戳设置一个延迟时间。这样即使在出现乱序的情况下，对数据的处理也可以获得正确的结果。为了处理无序事件，并区分流中的迟到事件。Flink 需要从事件数据中提取时间戳，并生成水位线，用来推进事件时间的进展。事件时间属性可以在创建表 DDL 中定义，也可以在数据流和表的转换中定义。

**1. 在创建表的 DDL 中定义**  
在创建表的 DDL（CREATE TABLE 语句）中，可以增加一个字段，通过 WATERMARK  
语句来定义事件时间属性。WATERMARK 语句主要用来定义水位线（watermark）的生成表达  
式，这个表达式会将带有事件时间戳的字段标记为事件时间属性，并在它基础上给出水位线的  
延迟时间。具体定义方式如下：

CREATE TABLE EventTable(
    	user STRING,
    	url STRING,
    	ts TIMESTAMP(3),
    	WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
    ) WITH (
     ...
    );

这里我们把 ts 字段定义为事件时间属性，而且基于 ts 设置了 5 秒的水位线延迟。这里的“5 秒”是以“时间间隔”的形式定义的，格式是 INTERVAL <数值> <时间单位>：INTERVAL ‘5’ SECOND 这里的数值必须用单引号引起来，而单位用 SECOND 和 SECONDS 是等效的。

Flink 中支持的事件时间属性数据类型必须为 TIMESTAMP 或者 TIMESTAMP\_LTZ。这里TIMESTAMP\_LTZ 是指带有本地时区信息的时间戳（TIMESTAMP WITH LOCAL TIME ZONE）；一般情况下如果数据中的时间戳是“年-月-日-时-分-秒”的形式，那就是不带时区信息的，可以将事件时间属性定义为 TIMESTAMP 类型。而如果原始的时间戳就是一个长整型的毫秒数，这时就需要另外定义一个字段来表示事件时间属性，类型定义为  
TIMESTAMP\_LTZ 会更方便：

CREATE TABLE events (
    	user STRING,
    	url STRING,
    	ts BIGINT,
    	ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
    	WATERMARK FOR ts_ltz AS time_ltz - INTERVAL '5' SECOND
    ) WITH (
    	...
    );

这里我们另外定义了一个字段 ts\_ltz，是把长整型的 ts 转换为 TIMESTAMP\_LTZ 得到的；进而使用 WATERMARK 语句将它设为事件时间属性，并设置 5 秒的水位线延迟。

**2. 在数据流转换为表时定义**

事件时间属性也可以在将 DataStream 转换为表的时候来定义。我们调用 fromDataStream() 方法创建表时，可以追加参数来定义表中的字段结构；这时可以给某个字段加上.rowtime() 后缀，就表示将当前字段指定为事件时间属性。这个字段可以是数据中本不存在、额外追加上去的“逻辑字段”，就像之前 DDL 中定义的第二种情况；也可以是本身固有的字段，那么这个字段就会被事件时间属性所覆盖，类型也会被转换为 TIMESTAMP。不论那种方式，时间属性字段中保存的都是事件的时间戳（TIMESTAMP 类型）。

需要注意的是，这种方式只负责指定时间属性，而时间戳的提取和水位线的生成应该之前就在 DataStream 上定义好了。由于 DataStream 中没有时区概念，因此 Flink 会将事件时间属性解析成不带时区的 TIMESTAMP 类型，所有的时间值都被当作 UTC 标准时间。在代码中的定义方式如下：

// 方法一:
    // 流中数据类型为二元组 Tuple2，包含两个字段；需要自定义提取时间戳并生成水位线
    DataStream<Tuple2<String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);
    // 声明一个额外的逻辑字段作为事件时间属性
    Table table = tEnv.fromDataStream(stream, $("user"), $("url"), $("ts").rowtime());
    
    // 方法二:
    // 流中数据类型为三元组 Tuple3，最后一个字段就是事件时间戳
    DataStream<Tuple3<String, String, Long>> stream = inputStream.assignTimestampsAndWatermarks(...);
    // 不再声明额外字段，直接用最后一个字段作为事件时间属性
    Table table = tEnv.fromDataStream(stream, $("user"), $("url"), $("ts").rowtime());

### 2. 处理时间 ###

相比之下处理时间就比较简单了，它就是我们的系统时间，使用时不需要提取时间戳（timestamp）和生成水位线(watermark)。因此在定义处理时间属性时，必须要额外声明一个字段，专门用来保存当前的处理时间。类似地，处理时间属性的定义也有两种方式：创建表 DDL 中定义，或者在数据流转换成表时定义。

**1. 在创建表的 DDL 中定义**

在创建表的 DDL（CREATE TABLE 语句）中，可以增加一个额外的字段，通过调用系统内置的 PROCTIME()函数来指定当前的处理时间属性，返回的类型是 TIMESTAMP\_LTZ。

CREATE TABLE EventTable(
    	user STRING,
    	url STRING,
    	ts AS PROCTIME()
    ) WITH (
    	...
    );

这里的时间属性，其实是以“计算列”（computed column）的形式定义出来的。所谓的计算列是 Flink SQL 中引入的特殊概念，可以用一个 AS 语句来在表中产生数据中不存在的列，并且可以利用原有的列、各种运算符及内置函数。在前面事件时间属性的定义中，将 ts 字段转换成 TIMESTAMP\_LTZ 类型的 ts\_ltz，也是计算列的定义方式。

**2. 在数据流转换为表时定义**

处理时间属性同样可以在将 DataStream 转换为表的时候来定义 。我们调用fromDataStream()方法创建表时，可以用.proctime()后缀来指定处理时间属性字段。由于处理时间是系统时间，原始数据中并没有这个字段，所以处理时间属性一定不能定义在一个已有字段上，只能定义在表结构所有字段的最后，作为额外的逻辑字段出现代码中定义处理时间属性的方法如下：

DataStream<Tuple2<String, String>> stream = ...;
    // 声明一个额外的字段作为处理时间属性字段
    Table table = tEnv.fromDataStream(stream, $("user"), $("url"), $("ts").proctime());

### 3. 窗口（Window） ###

有了时间属性，接下来就可以定义窗口进行计算了。我们知道，窗口可以将无界流切割成大小有限的“桶” (bucket)来做计算，通过截取有限数据集来处理无限的流数据。在 DataStream API 中提供了对不同类型的窗口进行定义和处理的接口，而在 Table API 和 SQL 中，类似的功能也都可以实现。

**1. 分组窗口（Group Window，老版本）**

在 Flink 1.12 之前的版本中，Table API 和 SQL 提供了一组“分组窗口”（Group Window）函数，常用的时间窗口如滚动窗口、滑动窗口、会话窗口都有对应的实现；具体在 SQL 中就是调用 TUMBLE()、HOP()、SESSION()，传入时间属性字段、窗口大小等参数就可以了。以滚动窗口为例：

TUMBLE(ts, INTERVAL '1' HOUR)

这里的 ts 是定义好的时间属性字段，窗口大小用“时间间隔”INTERVAL 来定义。在进行窗口计算时，分组窗口是将窗口本身当作一个字段对数据进行分组的，可以对组内的数据进行聚合。基本使用方式如下：

Table result = tableEnv.sqlQuery(
     "SELECT " + "user, " + "TUMBLE_END(ts, INTERVAL '1' HOUR) as endT, " +
     "COUNT(url) AS cnt " +"FROM EventTable " + "GROUP BY " + // 使用窗口和用户名进行分组
     "user, " +"TUMBLE(ts, INTERVAL '1' HOUR)" // 定义 1 小时滚动窗口
    );

这里定义了 1 小时的滚动窗口，将窗口和用户 user 一起作为分组的字段。用聚合函数COUNT()对分组数据的个数进行了聚合统计，并将结果字段重命名为cnt；用TUPMBLE\_END()函数获取滚动窗口的结束时间，重命名为 endT 提取出来。分组窗口的功能比较有限，只支持窗口聚合，所以目前已经处于弃用（deprecated）的状态。

**2. 窗口表值函数（Windowing TVFs，新版本）**

从 1.13 版本开始，Flink 开始使用窗口表值函数（Windowing table-valued functions，Windowing TVFs）来定义窗口。窗口表值函数是 Flink 定义的多态表函数（PTF），可以将表进行扩展后返回。表函数（table function）可以看作是返回一个表的函数，关于这部分内容，我们会在 11.6 节进行介绍。

目前 Flink 提供了以下几个窗口 TVF：

*  滚动窗口（Tumbling Windows）；
 *  滑动窗口（Hop Windows，跳跃窗口）；
 *  累积窗口（Cumulate Windows）；
 *  会话窗口（Session Windows，目前尚未完全支持）。

窗口表值函数可以完全替代传统的分组窗口函数。窗口 TVF 更符合 SQL 标准，性能得到了优化，拥有更强大的功能；可以支持基于窗口的复杂计算，例如窗口 Top-N、窗口联结（window join）等等。当然，目前窗口 TVF 的功能还不完善，会话窗口和很多高级功能还不支持，不过正在快速地更新完善。可以预见在未来的版本中，窗口 TVF 将越来越强大，将会是窗口处理的唯一入口。

在窗口 TVF 的返回值中，除去原始表中的所有列，还增加了用来描述窗口的额外 3 个列：“窗口起始点”(window\_start) 、“窗口结束点” (window\_end) 、“窗口时间” (window\_time)。起始点和结束点比较好理解，这里的“窗口时间”指的是窗口中的时间属性，它的值等于window\_end - 1ms，所以相当于是窗口中能够包含数据的最大时间戳。在 SQL 中的声明方式，与以前的分组窗口是类似的，直接调用 TUMBLE()、HOP()、CUMULATE()就可以实现滚动、滑动和累积窗口，不过传入的参数会有所不同。下面我们就分别对这几种窗口 TVF 进行介绍。

(1) 滚动窗口（TUMBLE）

滚动窗口在 SQL 中的概念与 DataStream API 中的定义完全一样，是长度固定、时间对齐、无重叠的窗口，一般用于周期性的统计计算。在 SQL 中通过调用 TUMBLE()函数就可以声明一个滚动窗口，只有一个核心参数就是窗  
口大小（size）。在 SQL 中不考虑计数窗口，所以滚动窗口就是滚动时间窗口，参数中还需要将当前的时间属性字段传入；另外，窗口 TVF 本质上是表函数，可以对表进行扩展，所以还应该把当前查询的表作为参数整体传入。具体声明如下：

TUMBLE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOUR)

这里基于时间字段 ts，对表 EventTable 中的数据开了大小为 1 小时的滚动窗口。窗口会将表中的每一行数据，按照它们 ts 的值分配到一个指定的窗口中。

(2) 滑动窗口（HOP）

滑动窗口的使用与滚动窗口类似，可以通过设置滑动步长来控制统计输出的频率。在 SQL中通过调用 HOP()来声明滑动窗口；除了也要传入表名、时间属性外，还需要传入窗口大小（size）和滑动步长（slide）两个参数。

HOP(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '1' HOURS));

这里我们基于时间属性 ts，在表 EventTable 上创建了大小为 1 小时的滑动窗口，每 5 分钟滑动一次。需要注意的是，紧跟在时间属性字段后面的第三个参数是步长（slide），第四个参数才是窗口大小（size）。

(3) 累积窗口（CUMULATE）

滚动窗口和滑动窗口，可以用来计算大多数周期性的统计指标。不过在实际应用中还会遇到这样一类需求：我们的统计周期可能较长，因此希望中间每隔一段时间就输出一次当前的统计值；与滑动窗口不同的是，在一个统计周期内，我们会多次输出统计值，它们应该是不断叠加累积的。

例如，我们按天来统计网站的 PV（Page View，页面浏览量），如果用 1 天的滚动窗口，那需要到每天 24 点才会计算一次，输出频率太低；如果用滑动窗口，计算频率可以更高，但统计的就变成了“过去 24 小时的 PV”。所以我们真正希望的是，还是按照自然日统计每天的PV，不过需要每隔 1 小时就输出一次当天到目前为止的 PV 值。这种特殊的窗口就叫作“累积窗口”（Cumulate Window）。

![在这里插入图片描述][7c2679cef5914d1397ebe7c90e4eb98b.png]  
累积窗口是窗口 TVF 中新增的窗口功能，它会在一定的统计周期内进行累积计算。累积窗口中有两个核心的参数：最大窗口长度（max window size）和累积步长（step）。所谓的最大窗口长度其实就是我们所说的“统计周期”，最终目的就是统计这段时间内的数据。如图 11-8所示，开始时，创建的第一个窗口大小就是步长 step；之后的每个窗口都会在之前的基础上再扩展 step 的长度，直到达到最大窗口长度。在 SQL 中可以用 CUMULATE()函数来定义，具体如下：

CUMULATE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOURS, INTERVAL '1' DAYS))

这里我们基于时间属性 ts，在表 EventTable 上定义了一个统计周期为 1 天、累积步长为 1小时的累积窗口。注意第三个参数为步长 step，第四个参数则是最大窗口长度。上面所有的语句只是定义了窗口，类似于 DataStream API 中的窗口分配器；在 SQL 中窗口的完整调用，还需要配合聚合操作和其它操作。我们会在下一节详细讲解窗口的聚合。

## 五、聚合 (Aggregation) 查询 ##

在 SQL 中，一个很常见的功能就是对某一列的多条数据做一个合并统计，得到一个或多个结果值；比如求和、最大最小值、平均值等等，这种操作叫作聚合（Aggregation）查询。Flink 中的 SQL 是流处理与标准 SQL 结合的产物，所以聚合查询也可以分成两种：流处理中特有的聚合（主要指窗口聚合），以及 SQL 原生的聚合查询方式。

### 1. 分组聚合 ###

SQL 中一般所说的聚合我们都很熟悉，主要是通过内置的一些聚合函数来实现的，比如 SUM()、MAX()、MIN()、AVG() 以及 COUNT()。它们的特点是对多条输入数据进行计算，得到一个唯一的值，属于“多对一”的转换。比如我们可以通过下面的代码计算输入数据的个数：

Table eventCountTable = tableEnv.sqlQuery("select COUNT(*) from EventTable");

而更多的情况下，我们可以通过 GROUP BY 子句来指定分组的键（key），从而对数据按照某个字段做一个分组统计。例如之前我们举的例子，可以按照用户名进行分组，统计每个用户点击 url 的次数：

SELECT user, COUNT(url) as cnt FROM EventTable GROUP BY user

这种聚合方式，就叫作“分组聚合”（group aggregation）。从概念上讲，SQL 中的分组聚合可以对应 DataStream API 中 keyBy 之后的聚合转换，它们都是按照某个 key 对数据进行了划分，各自维护状态来进行聚合统计的。在流处理中，分组聚合同样是一个持续查询，而且是一个更新查询，得到的是一个动态表；每当流中有一个新的数据到来时，都会导致结果表的更新操作。因此，想要将结果表转换成流或输出到外部系统，必须采用撤回流（retract stream）或更新插入流（upsert stream）的编码方式；如果在代码中直接转换成 DataStream 打印输出，  
需要调用 toChangelogStream()。

另外，在持续查询的过程中，由于用于分组的 key 可能会不断增加，因此计算结果所需要维护的状态也会持续增长。为了防止状态无限增长耗尽资源，Flink Table API 和 SQL 可以在表环境中配置状态的生存时间（TTL）：

TableEnvironment tableEnv = ...
    // 获取表环境的配置
    TableConfig tableConfig = tableEnv.getConfig();
    // 配置状态保持时间
    tableConfig.setIdleStateRetention(Duration.ofMinutes(60));
    或者也可以直接设置配置项 table.exec.state.ttl：
    TableEnvironment tableEnv = ...
    Configuration configuration = tableEnv.getConfig().getConfiguration();
    configuration.setString("table.exec.state.ttl", "60 min");

这两种方式是等效的。需要注意，配置 TTL 有可能会导致统计结果不准确，这其实是以牺牲正确性为代价换取了资源的释放。

此外，在 Flink SQL 的分组聚合中同样可以使用 DISTINCT 进行去重的聚合处理；可以使用 HAVING 对聚合结果进行条件筛选；还可以使用 GROUPING SETS（分组集）设置多个分组情况分别统计。这些语法跟标准 SQL 中的用法一致，这里就不再详细展开了。  
可以看到，分组聚合既是 SQL 原生的聚合查询，也是流处理中的聚合操作，这是实际应用中最常见的聚合方式。当然，使用的聚合函数一般都是系统内置的，如果希望实现特殊需求也可以进行自定义。关于自定义函数（UDF），我们会在 11.7 节中详细介绍。

### 2. 窗口聚合 ###

在流处理中，往往需要将无限数据流划分成有界数据集，这就是所谓的“窗口”。在 11.4.3小节中已经介绍了窗口的声明方式，这相当于 DataStream API 中的窗口分配器（window assigner），只是明确了窗口的形式以及数据如何分配；而窗口具体的计算处理操作，在DataStream API 中还需要窗口函数（window function）来进行定义。

在 Flink 的 Table API 和 SQL 中，窗口的计算是通过“窗口聚合”（window aggregation）来实现的。与分组聚合类似，窗口聚合也需要调用 SUM()、MAX()、MIN()、COUNT()一类的聚合函数，通过 GROUP BY 子句来指定分组的字段。只不过窗口聚合时，需要将窗口信息作为分组 key 的一部分定义出来。在 Flink 1.12 版本之前，是直接把窗口自身作为分组 key 放在GROUP BY 之后的，所以也叫“分组窗口聚合”（参见 11.4.3 小节）；而 1.13 版本开始使用了“窗口表值函数”（Windowing TVF），窗口本身返回的是就是一个表，所以窗口会出现在 FROM后面，GROUP BY 后面的则是窗口新增的字段 window\_start 和 window\_end。  
比如，我们将 11.4.3 中分组窗口的聚合，用窗口 TVF 重新实现一下：

Table result = tableEnv.sqlQuery(
    	"SELECT " +
    	 "user, " +
    	"window_end AS endT, " +
    	"COUNT(url) AS cnt " +
    	"FROM TABLE( " +
    	"TUMBLE( TABLE EventTable, " +
    	"DESCRIPTOR(ts), " +
    	"INTERVAL '1' HOUR)) " +
    	"GROUP BY user, window_start, window_end "
    );

这里我们以 ts 作为时间属性字段、基于 EventTable 定义了 1 小时的滚动窗口，希望统计出每小时每个用户点击 url 的次数。用来分组的字段是用户名 user，以及表示窗口的,window\_start 和 window\_end；而 TUMBLE()是表值函数，所以得到的是一个表（Table），我们的聚合查询就是在这个 Table 中进行的。这就是 11.3.2 小节中窗口聚合的实现方式。

Flink SQL 目前提供了滚动窗口 TUMBLE()、滑动窗口 HOP()和累积窗口（CUMULATE）三种表值函数 (TVF)。在具体应用中，我们还需要提前定义好时间属性。下面是一段窗口聚合的完整代码，以累积窗口为例：

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);
    		// 读取数据源，并分配时间戳、生成水位线
    		SingleOutputStreamOperator<Event> 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 SerializableTimestampAssigner<Event>() {
        
    						@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 " +
    				"FROM TABLE( " +
    				"CUMULATE( TABLE EventTable, " + // 定义累积窗口
    				"DESCRIPTOR(ts), " +
    				"INTERVAL '30' MINUTE, " +
    				"INTERVAL '1' HOUR)) " +
    				"GROUP BY user, window_start, window_end "
    			);
    		tableEnv.toDataStream(result).print();
    		env.execute();
    	}
    }

这里我们使用了统计周期为 1 小时、累积间隔为 30 分钟的累积窗口。可以看到，代码的架构和处理逻辑与 11.3.2 小节中的实现完全一致，只是将滚动窗口 TUMBLE()换成了累积窗口  
CUMULATE()。代码执行结果如下：

+I[Alice, 1970-01-01T00:30, 2]
    +I[Bob, 1970-01-01T00:30, 1]
    +I[Alice, 1970-01-01T01:00, 3]
    +I[Bob, 1970-01-01T01:00, 1]
    +I[Bob, 1970-01-01T01:30, 1]
    +I[Cary, 1970-01-01T02:00, 2]
    +I[Bob, 1970-01-01T02:00, 1]

与分组聚合不同，窗口聚合不会将中间聚合的状态输出，只会最后输出一个结果。我们可以看到，所有数据都是以 INSERT 操作追加到结果动态表中的，因此输出每行前面都有+I 的前缀。所以窗口聚合查询都属于追加查询，没有更新操作，代码中可以直接用 toDataStream()将结果表转换成流。

具体来看，上面代码输入的前三条数据属于第一个半小时的累积窗口，其中 Alice 的访问数据有两条，Bob 的访问数据有 1 条，所以输出了两条结果\[Alice, 1970-01-01T00:30, 2\]和\[Bob, 1970-01-01T00:30, 1\]；而之后又到来的一条 Alice 访问数据属于第二个半小时范围，同时也属于第一个 1 小时的统计周期 ，所以会在之前两条的基础上进行叠加，输出 \[Alice, 1970-01-01T00:30, 3\]，而 Bob 没有新的访问数据，因此依然输出\[Bob, 1970-01-01T00:30, 1\]。

从第二个小时起，数据属于新的统计周期，就全部从零开始重新计数了。相比之前的分组窗口聚合，Flink 1.13 版本的窗口表值函数（TVF）聚合有更强大的功能。除了应用简单的聚合函数、提取窗口开始时间（window\_start）和结束时间(window\_end)之外，窗口 TVF 还提供了一个 window\_time 字段，用于表示窗口中的时间属性；这样就可以方便地进行窗口的级联（cascading window）和计算了。另外，窗口 TVF 还支持 GROUPING SETS，极大地扩展了窗口的应用范围。

基于窗口的聚合，是流处理中聚合统计的一个特色，也是与标准 SQL 最大的不同之处。在实际项目中，很多统计指标其实都是基于时间窗口来进行计算的，所以窗口聚合是 Flink SQL中非常重要的功能；基于窗口 TVF 的聚合未来也会有更多功能的扩展支持，比如窗口 Top N、会话窗口、窗口联结等等。

### 3. 开窗（Over）聚合 ###

在标准 SQL 中还有另外一类比较特殊的聚合方式，可以针对每一行计算一个聚合值。比如说，我们可以以每一行数据为基准，计算它之前 1 小时内所有数据的平均值；也可以计算它之前 10 个数的平均值。就好像是在每一行上打开了一扇窗户、收集数据进行统计一样，这就是所谓的“开窗函数”。开窗函数的聚合与之前两种聚合有本质的不同：分组聚合、窗口 TVF聚合都是“多对一”的关系，将数据分组之后每组只会得到一个聚合结果；而开窗函数是对每行都要做一次开窗聚合，因此聚合之后表中的行数不会有任何减少，是一个“多对多”的关系。与标准 SQL 中一致，Flink SQL 中的开窗函数也是通过 OVER 子句来实现的，所以有时开窗聚合也叫作“OVER 聚合”（Over Aggregation）。基本语法如下：

SELECT
     <聚合函数> OVER (
     [PARTITION BY <字段 1>[, <字段 2>, ...]]
     ORDER BY <时间属性字段>
     <开窗范围>),
     ...
    FROM ...

这里 OVER 关键字前面是一个聚合函数，它会应用在后面 OVER 定义的窗口上。在 OVER子句中主要有以下几个部分：

*  PARTITION BY（可选）用来指定分区的键（key），类似于 GROUP BY 的分组，这部分是可选的；
 *  ORDER BY  
    OVER 窗口是基于当前行扩展出的一段数据范围，选择的标准可以基于时间也可以基于数量。不论那种定义，数据都应该是以某种顺序排列好的；而表中的数据本身是无序的。所以在 OVER 子句中必须用 ORDER BY 明确地指出数据基于那个字段排序。在 Flink 的流处理中，目前只支持按照时间属性的升序排列，所以这里 ORDER BY 后面的字段必须是定义好的时间属性。
 *  开窗范围  
    对于开窗函数而言，还有一个必须要指定的就是开窗的范围，也就是到底要扩展多少行来做聚合。这个范围是由 BETWEEN <下界> AND <上界> 来定义的，也就是“从下界到上界”的范围。目前支持的上界只能是 CURRENT ROW，也就是定义一个“从之前某一行到当前行”的范围，所以一般的形式为：  
    BETWEEN … PRECEDING AND CURRENT ROW

前面我们提到，开窗选择的范围可以基于时间，也可以基于数据的数量。所以开窗范围还应该在两种模式之间做出选择：范围间隔（RANGE intervals）和行间隔（ROW intervals）。

*  范围间隔  
    范围间隔以 RANGE 为前缀，就是基于 ORDER BY 指定的时间字段去选取一个范围，一般就是当前行时间戳之前的一段时间。例如开窗范围选择当前行之前 1 小时的数据：  
    RANGE BETWEEN INTERVAL ‘1’ HOUR PRECEDING AND CURRENT ROW
 *  行间隔  
    行间隔以 ROWS 为前缀，就是直接确定要选多少行，由当前行出发向前选取就可以了。例如开窗范围选择当前行之前的 5 行数据（最终聚合会包括当前行，所以一共 6 条数据）：  
    ROWS BETWEEN 5 PRECEDING AND CURRENT ROW

下面是一个具体示例：

SELECT user, ts,
     COUNT(url) OVER (
     PARTITION BY user
     ORDER BY ts
     RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
    ) AS cnt
    FROM EventTable

这里我们以 ts 作为时间属性字段，对 EventTable 中的每行数据都选取它之前 1 小时的所有数据进行聚合，统计每个用户访问 url 的总次数，并重命名为 cnt。最终将表中每行的 user，ts 以及扩展出 cnt 提取出来。

可以看到，整个开窗聚合的结果，是对每一行数据都有一个对应的聚合值，因此就像将表中扩展出了一个新的列一样。由于聚合范围上界只能到当前行，新到的数据一般不会影响之前数据的聚合结果，所以结果表只需要不断插入（INSERT）就可以了。执行上面 SQL 得到的结果表，可以用 toDataStream()直接转换成流打印输出。开窗聚合与窗口聚合（窗口 TVF 聚合）本质上不同，不过也还是有一些相似之处的：它们都是在无界的数据流上划定了一个范围，截取出有限数据集进行聚合统计；这其实都是“窗口”的思路。事实上，在 Table API 中确实就定义了两类窗口：分组窗口（GroupWindow）和开窗窗口（OverWindow）；而在 SQL 中，也可以用 WINDOW 子句来在 SELECT 外部单独定义一个 OVER 窗口：

SELECT user, ts,
     COUNT(url) OVER w AS cnt,
     MAX(CHAR_LENGTH(url)) OVER w AS max_url
    FROM EventTable
    WINDOW w AS (
     PARTITION BY user
     ORDER BY ts
     ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
    )

上面的 SQL 中定义了一个选取之前 2 行数据的 OVER 窗口，并重命名为 w；接下来就可以基于它调用多个聚合函数，扩展出更多的列提取出来。比如这里除统计 url 的个数外，还统计了 url 的最大长度：首先用 CHAR\_LENGTH()函数计算出 url 的长度，再调用聚合函数 MAX()进行聚合统计。这样，我们就可以方便重复引用定义好的 OVER 窗口了，大大增强了代码的可读性。

### 4. 应用实例 —— Top N ###

灵活使用各种类型的窗口以及聚合函数，可以实现不同的需求。一般的聚合函数，比如SUM()、MAX()、MIN()、COUNT()等，往往只是针对一组数据聚合得到一个唯一的值；所谓OVER 聚合的“多对多”模式，也是针对每行数据都进行一次聚合才得到了多行的结果，对于每次聚合计算实际上得到的还是唯一的值。而有时我们可能不仅仅需要统计数据中的最大/最小值，还希望得到前 N 个最大/最小值；这时每次聚合的结果就不是一行，而是 N 行了。这就是经典的“Top N”应用场景。

Top N 聚合字面意思是“最大 N 个”，这只是一个泛称，它不仅包括查询最大的 N 个值、也包括了查询最小的 N 个值的场景。理想的状态下，我们应该有一个 TOPN()聚合函数，调用它对表进行聚合就可以得到想要选取的前 N 个值了。不过仔细一想就会发现，这个聚合函数并不容易实现：对于每一次聚合计算，都应该都有多行数据输入，并得到 N 行结果输出，这是一个真正意义上的“多对多”转换。这种函数相当于把一个表聚合成了另一个表，所以叫作“表聚合函数”（Table Aggregate Function）。表聚合函数的抽象比较困难，目前只有窗口 TVF 有能力提供直接的 Top N 聚合，不过也尚未实现。

所以目前在 Flink SQL 中没有能够直接调用的 Top N 函数，而是提供了稍微复杂些的变通实现方法。

**1. 普通 Top N**

在 Flink SQL 中，是通过 OVER 聚合和一个条件筛选来实现 Top N 的。具体来说，是通过将一个特殊的聚合函数ROW\_NUMBER()应用到OVER窗口上，统计出每一行排序后的行号，作为一个字段提取出来；然后再用 WHERE 子句筛选行号小于等于 N 的那些行返回。

基本语法如下：

SELECT ...
    FROM (
     SELECT ...,
     ROW_NUMBER() OVER (
    [PARTITION BY <字段 1>[, <字段 1>...]]
     ORDER BY <排序字段 1> [asc|desc][, <排序字段 2> [asc|desc]...]
    ) AS row_num
    FROM ...)
    WHERE row_num <= N [AND <其它条件>]

这里的 OVER 窗口定义与之前的介绍基本一致，目的就是利用 ROW\_NUMBER()函数为每一行数据聚合得到一个排序之后的行号。行号重命名为 row\_num，并在外层的查询中以 row\_num <= N 作为条件进行筛选，就可以得到根据排序字段统计的 Top N 结果了。  
需要对关键字额外做一些说明：

*  WHERE  
    用来指定 Top N 选取的条件，这里必须通过 row\_num <= N 或者 row\_num < N + 1 指定一个“排名结束点”（rank end），以保证结果有界。
 *  PARTITION BY  
    可选的，用来指定分区的字段，这样我们就可以针对不同的分组分别统计 Top N 了。
 *  ORDER BY  
    指定了排序的字段，因为只有排序之后，才能进行前 N 个最大/最小的选取。每个排序字段后可以用 asc 或者 desc 来指定排序规则：asc 为升序排列，取出的就是最小的 N 个值；desc为降序排序，对应的就是最大的 N 个值。默认情况下为升序，asc 可以省略。

细心的读者可能会发现，之前介绍的 OVER 窗口不是说了，目前 ORDER BY 后面只能跟时间字段、并且只支持升序吗？这里怎么又可以任意指定字段进行排序了呢？这是因为 OVER 窗口目前并不完善，不过针对 Top N 这样一个经典应用场景，Flink SQL专门用 OVER 聚合做了优化实现。所以只有在 Top N 的应用场景中，OVER 窗口 ORDER BY后才可以指定其它排序字段；而要想实现 Top N，就必须按照上面的格式进行定义，否则 Flink SQL 的优化器将无法正常解析。而且，目前 Table API 中并不支持 ROW\_NUMBER()函数，所以也只有 SQL 中这一种通用的 Top N 实现方式。

另外要注意，Top N 的实现必须写成上面的嵌套查询形式。这是因为行号 row\_num 是内部子查询聚合的结果，不可能在内部作为筛选条件，只能放在外层的 WHERE 子句中。下面是一个具体的示例，我们统计每个用户的访问事件中，按照字符长度排序的前两个

url：
    SELECT user, url, ts, row_num
    FROM (
     SELECT *,
     ROW_NUMBER() OVER (
    PARTITION BY user
     ORDER BY CHAR_LENGTH(url) desc 
    ) AS row_num FROM EventTable) WHERE row_num <= 2

这里我们以用户来分组，以访问 url 的字符长度作为排序的字段，降序排列后用聚合统计出每一行的行号，这样就相当于在 EventTable 基础上扩展出了一列 row\_num。而后筛选出行号小于等于 2 的所有数据，就得到了每个用户访问的长度最长的两个 url。

需要特别说明的是，这里的 Top N 聚合是一个更新查询。新数据到来后，可能会改变之前数据的排名，所以会有更新（UPDATE）操作。这是 ROW\_NUMBER()聚合函数的特性决定的。因此，如果执行上面的 SQL 得到结果表，需要调用 toChangelogStream()才能转换成流打印输出。

**2. 窗口 Top N**

除了直接对数据进行 Top N 的选取，我们也可以针对窗口来做 Top N。  
例如电商行业，实际应用中往往有这样的需求：统计一段时间内的热门商品。这就需要先开窗口，在窗口中统计每个商品的点击量；然后将统计数据收集起来，按窗口进行分组，并按点击量大小降序排序，选取前 N 个作为结果返回。

我们已经知道，Top N 聚合本质上是一个表聚合函数，这和窗口表值函数（TVF）有天然的联系。尽管如此，想要基于窗口 TVF 实现一个通用的 Top N 聚合函数还是比较麻烦的，目前Flink SQL尚不支持。不过我们同样可以借鉴之前的思路，使用OVER窗口统计行号来实现。

具体来说，可以先做一个窗口聚合，将窗口信息 window\_start、window\_end 连同每个商品的点击量一并返回，这样就得到了聚合的结果表，包含了窗口信息、商品和统计的点击量。接下来就可以像一般的 Top N 那样定义 OVER 窗口了，按窗口分组，按点击量排序，用ROW\_NUMBER()统计行号并筛选前 N 行就可以得到结果。所以窗口 Top N 的实现就是窗口聚合与 OVER 聚合的结合使用。

下面是一个具体案例的代码实现。由于用户访问事件 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 WindowTopNExample {
        
    	public static void main(String[] args) throws Exception {
        
    		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    		env.setParallelism(1);
    		// 读取数据源，并分配时间戳、生成水位线
    		SingleOutputStreamOperator<Event> 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  SerializableTimestampAssigner<Event>() {
        
    								@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") 
    			// 将 timestamp 指定为事件时间，并命名为 ts
    		);
    		// 为方便在 SQL 中引用，在环境中注册表 EventTable
    		tableEnv.createTemporaryView("EventTable", eventTable);
    		// 定义子查询，进行窗口聚合，得到包含窗口信息、用户以及访问次数的结果表
    		String subQuery =	"SELECT window_start, window_end, user, COUNT(url) as cnt " +	"FROM TABLE ( " +"TUMBLE( TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOUR )) 
    		" +"GROUP BY window_start, window_end, user ";
    		// 定义 Top N 的外层查询
    		String topNQuery =
    			"SELECT * " +
    			"FROM (" +
    			"SELECT *, " +
    			"ROW_NUMBER() OVER ( " +
    			"PARTITION BY window_start, window_end " +
    			"ORDER BY cnt desc " +
    			") AS row_num " +
    			"FROM (" + subQuery + ")) " +
    			"WHERE row_num <= 2";
    		// 执行 SQL 得到结果表
    		Table result = tableEnv.sqlQuery(topNQuery);
    		tableEnv.toDataStream(result).print();
    		env.execute();
    	}
    }

这里为了更好的代码可读性，我们将 SQL 拆分成了用来做窗口聚合的内部子查询，和套用 Top N 模板的外层查询。

(1) 首先基于 ts 时间字段定义 1 小时滚动窗口，统计 EventTable 中每个用户的访问次数，重命名为 cnt；为了方便后面做排序，我们将窗口信息 window\_start 和 window\_end 也提取出来，与 user 和 cnt 一起作为聚合结果表中的字段。

(2) 然后套用 Top N 模板，对窗口聚合的结果表中每一行数据进行 OVER 聚合统计行号。

这里以窗口信息进行分组，按访问次数 cnt 进行排序，并筛选行号小于等于 2 的数据，就可以得到每个窗口内访问次数最多的前两个用户了。  
运行结果如下：

+I[1970-01-01T00:00, 1970-01-01T01:00, Alice, 3, 1]
    +I[1970-01-01T00:00, 1970-01-01T01:00, Bob, 1, 2]
    +I[1970-01-01T01:00, 1970-01-01T02:00, Cary, 2, 1]
    +I[1970-01-01T01:00, 1970-01-01T02:00, Bob, 1, 2]

可以看到，第一个 1 小时窗口中，Alice 有 3 次访问排名第一，Bob 有 1 次访问排名第二；而第二小时内，Cary 以 2 次访问占据活跃榜首，Bob 仍以 1 次访问排名第二。由于窗口的统计结果只会最终输出一次，所以排名也是确定的，这里结果表中只有插入（INSERT）操作。也就是说，窗口 Top N 是追加查询，可以直接用 toDataStream()将结果表转换成流打印输出。

## 六、联结（Join）查询 ##

按照数据库理论，关系型表的设计往往至少需要满足第三范式（3NF），表中的列都直接依赖于主键，这样就可以避免数据冗余和更新异常。例如商品的订单信息，我们会保存在一个“订单表”中，而这个表中只有商品 ID，详情则需要到“商品表”按照 ID 去查询；这样的好处是当商品信息发生变化时，只要更新商品表即可，而不需要在订单表中对所有这个商品的所有订单进行修改。不过这样一来，我们就无法从一个单独的表中提取所有想要的数据了。

在标准 SQL 中，可以将多个表连接合并起来，从中查询出想要的信息；这种操作就是表的联结（Join）。在 Flink SQL 中，同样支持各种灵活的联结（Join）查询，操作的对象是动态表。

在流处理中，动态表的 Join 对应着两条数据流的 Join 操作。与上一节的聚合查询类似，Flink SQL 中的联结查询大体上也可以分为两类：SQL 原生的联结查询方式，和流处理中特有的联结查询。

### 1. 常规联结查询 ###

常规联结（Regular Join）是 SQL 中原生定义的 Join 方式，是最通用的一类联结操作。它的具体语法与标准 SQL 的联结完全相同，通过关键字 JOIN 来联结两个表，后面用关键字 ON来指明联结条件。按照习惯，我们一般以“左侧”和“右侧”来区分联结操作的两个表。

在两个动态表的联结中，任何一侧表的插入（INSERT）或更改（UPDATE）操作都会让联结的结果表发生改变。例如，如果左侧有新数据到来，那么它会与右侧表中所有之前的数据进行联结合并，右侧表之后到来的新数据也会与这条数据连接合并。所以，常规联结查询一般是更新（Update）查询。

与标准 SQL 一致，Flink SQL 的常规联结也可以分为内联结（INNER JOIN）和外联结（OUTER JOIN），区别在于结果中是否包含不符合联结条件的行。目前仅支持“等值条件”作为联结条件，也就是关键字 ON 后面必须是判断两表中字段相等的逻辑表达式。

**1. 等值内联结（INNER Equi-JOIN）**

内联结用 INNER JOIN 来定义，会返回两表中符合联接条件的所有行的组合，也就是所谓的笛卡尔积（Cartesian product）。目前仅支持等值联结条件。例如之前提到的“订单表”（定义为 Order）和“商品表”（定义为 Product）的联结查询，就可以用以下 SQL 实现：

SELECT *
    FROM Order
    INNER JOIN Product
    ON Order.product_id = Product.id

这里是一个内联结，联结条件是订单数据的 product\_id 和商品数据的 id 相等。由于订单表中出现的商品id一定会在商品表中出现，因此这样得到的联结结果表，就包含了订单表Order中所有订单数据对应的详细信息。

**2. 等值外联结（OUTER Equi-JOIN）**

与内联结类似，外联结也会返回符合联结条件的所有行的笛卡尔积；另外，还可以将某一侧表中找不到任何匹配的行也单独返回。Flink SQL 支持左外（LEFT JOIN）、右外（RIGHT JOIN）和全外（FULL OUTER JOIN），分别表示会将左侧表、右侧表以及双侧表中没有任何匹配的行返回。例如，订单表中未必包含了商品表中的所有 ID，为了将哪些没有任何订单的商品信息也查询出来，我们就可以使用右外联结（RIGHT JOIN）。当然，外联结查询目前也仅支持等值

联结条件。具体用法如下：

SELECT *
    FROM Order
    LEFT JOIN Product
    ON Order.product_id = Product.id
    SELECT *
    FROM Order
    RIGHT JOIN Product
    ON Order.product_id = Product.id
    SELECT *
    FROM Order
    FULL OUTER JOIN Product
    ON Order.product_id = Product.id

这部分知识与标准 SQL 中是完全一样的，这里不再赘述。

### 2. 间隔联结查询 ###

在 8.3 节中，我们曾经学习过 DataStream API 中的双流 Join，包括窗口联结（window join）和间隔联结 (interval join）。两条流的 Join 就对应着 SQL 中两个表的 Join，这是流处理中特有的联结方式。目前 Flink SQL 还不支持窗口联结，而间隔联结则已经实现。

间隔联结（Interval Join）返回的，同样是符合约束条件的两条中数据的笛卡尔积。只不过这里的“约束条件”除了常规的联结条件外，还多了一个时间间隔的限制。具体语法有以下要点：

*  两表的联结  
    间隔联结不需要用 JOIN 关键字，直接在 FROM 后将要联结的两表列出来就可以，用逗号分隔。这与标准 SQL 中的语法一致，表示一个“交叉联结”（Cross Join），会返回两表中所有行的笛卡尔积。
 *  联结条件  
    联结条件用 WHERE 子句来定义，用一个等值表达式描述。交叉联结之后再用 WHERE进行条件筛选，效果跟内联结 INNER JOIN … ON …非常类似。\\
 *  时间间隔限制  
    我们可以在 WHERE 子句中，联结条件后用 AND 追加一个时间间隔的限制条件；做法是提取左右两侧表中的时间字段，然后用一个表达式来指明两者需要满足的间隔限制。具体定义方式有下面三种，这里分别用 ltime 和 rtime 表示左右表中的时间字段：  
    （1）ltime = rtime  
    （2）ltime >= rtime AND ltime < rtime + INTERVAL ‘10’ MINUTE  
    （3）ltime BETWEEN rtime - INTERVAL ‘10’ SECOND AND rtime + INTERVAL ‘5’ SECOND

判断两者相等，这是最强的时间约束，要求两表中数据的时间必须完全一致才能匹配；一般情况下，我们还是会放宽一些，给出一个间隔。间隔的定义可以用<，<=，>=，>这一类的关系不等式，也可以用 BETWEEN … AND …这样的表达式。

例如，我们现在除了订单表 Order 外，还有一个“发货表”Shipment，要求在收到订单后四个小时内发货。那么我们就可以用一个间隔联结查询，把所有订单与它对应的发货信息连接合并在一起返回。

SELECT *
    FROM Order o, Shipment s
    WHERE o.id = s.order_id
    AND o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time

在流处理中，间隔联结查询只支持具有时间属性的“仅追加”（Append-only）表。那对于有更新操作的表，又怎么办呢？除了间隔联结之外，Flink SQL 还支持时间联结（Temporal Join），这主要是针对“版本表”（versioned table）而言的。所谓版本表，就是记录了数据随着时间推移版本变化的表，可以理解成一个“更新日志”（change log），它就是具有时间属性、还会进行更新操作的表。当我们联结某个版本表时，并不是把当前的数据连接合并  
起来就行了，而是希望能够根据数据发生的时间，找到当时的“版本”；这种根据更新时间提取当时的值进行联结的操作，就叫作“时间联结”（Temporal Join）。这部分内容由于涉及版本表的定义，我们就不详细展开了，感兴趣的读者可以查阅官网资料。

## 七、函数 ##

在 SQL 中，我们可以把一些数据的转换操作包装起来，嵌入到 SQL 查询中统一调用，这就是“函数(functions)。Flink 的 Table API 和 SQL 同样提供了函数的功能。两者在调用时略有不同：Table API 中的函数是通过数据对象的方法调用来实现的；而 SQL 则是直接引用函数名称，传入数据作为参数。例如，要把一个字符串 str 转换成全大写的形式，Table API 的写法是调用 str 这个 String对象的 upperCase()方法：

str.upperCase();

而 SQL 中的写法就是直接引用 UPPER()函数，将 str 作为参数传入：UPPER(str) 由于 Table API 是内嵌在 Java 语言中的，很多方法需要在类中额外添加，因此扩展功能比较麻烦，目前支持的函数比较少；而且 Table API 也不如 SQL 的通用性强，所以一般情况下较少使用。下面我们主要介绍 Flink SQL 中函数的使用。Flink SQL 中的函数可以分为两类：一类是 SQL 中内置的系统函数，直接通过函数名调用就可以，能够实现一些常用的转换操作，比如之前我们用到的 COUNT()、CHAR\_LENGTH()、UPPER()等等；而另一类函数则是用户自定义的函数(UDF)，需要在表环境中注册才能使用。接下来我们就对这两类函数分别进行介绍。

### 1. 系统函数 ###

系统函数（System Functions）也叫内置函数（Built-in Functions），是在系统中预先实现好的功能模块。我们可以通过固定的函数名直接调用，实现想要的转换操作。Flink SQL 提供了大量的系统函数，几乎支持所有的标准 SQL 中的操作，这为我们使用 SQL 编写流处理程序提供了极大的方便。

Flink SQL 中的系统函数又主要可以分为两大类：标量函数（Scalar Functions）和聚合函数（Aggregate Functions）。

**1. 标量函数（Scalar Functions）**

所谓的“标量”，是指只有数值大小、没有方向的量；所以标量函数指的就是只对输入数据做转换操作、返回一个值的函数。这里的输入数据对应在表中，一般就是一行数据中 1 个或多个字段，因此这种操作有点像流处理转换算子中的 map。另外，对于一些没有输入参数、直接可以得到唯一结果的函数，也属于标量函数。标量函数是最常见、也最简单的一类系统函数，数量非常庞大，很多在标准 SQL 中也有定义。所以我们这里只对一些常见类型列举部分函数，做一个简单概述，具体应用可以查看官网的完整函数列表。

*  比较函数（Comparison Functions）  
    比较函数其实就是一个比较表达式，用来判断两个值之间的关系，返回一个布尔类型的值。  
    这个比较表达式可以是用 <、>、= 等符号连接两个值，也可以是用关键字定义的某种判断。  
    例如：  
    （1）value1 = value2 判断两个值相等；  
    （2）value1 <> value2 判断两个值不相等  
    （3）value IS NOT NULL 判断 value 不为空
 *  逻辑函数（Logical Functions）  
    逻辑函数就是一个逻辑表达式，也就是用与（AND）、或（OR）、非（NOT）将布尔类型的值连接起来，也可以用判断语句（IS、IS NOT）进行真值判断；返回的还是一个布尔类型的值。  
    例如：  
    （1）boolean1 OR boolean2 布尔值 boolean1 与布尔值 boolean2 取逻辑或  
    （2）boolean IS FALSE 判断布尔值 boolean 是否为 false  
    （3）NOT boolean 布尔值 boolean 取逻辑非
 *  算术函数（Arithmetic Functions）  
    进行算术计算的函数，包括用算术符号连接的运算，和复杂的数学运算。  
    例如：  
    （1）numeric1 + numeric2 两数相加  
    （2）POWER(numeric1, numeric2) 幂运算，取数 numeric1 的 numeric2 次方  
    （3）RAND() 返回（0.0, 1.0）区间内的一个 double 类型的伪随机数
 *  字符串函数（String Functions）  
    进行字符串处理的函数。  
    例如：  
    （1）string1 || string2 两个字符串的连接  
    （2）UPPER(string) 将字符串 string 转为全部大写  
    （3）CHAR\_LENGTH(string) 计算字符串 string 的长度
 *  时间函数（Temporal Functions）  
    进行与时间相关操作的函数。例如：  
    （1）DATE string 按格式"yyyy-MM-dd"解析字符串 string，返回类型为 SQL Date  
    （2）TIMESTAMP string 按格式"yyyy-MM-dd HH:mm:ss\[.SSS\]"解析，返回类型为 SQL timestamp  
    （3）CURRENT\_TIME 返回本地时区的当前时间，类型为 SQL time（与 LOCALTIME等价）  
    （4）INTERVAL string range 返回一个时间间隔。string 表示数值；range 可以是 DAY，MINUTE，DAT TO HOUR 等单位，也可以是 YEAR TO MONTH 这样的复合单位。如“2 年10 个月”可以写成：INTERVAL ‘2-10’ YEAR TO MONTH样的复合单位。如“2 年10 个月”可以写成：INTERVAL ‘2-10’ YEAR TO MONTH

**2. 聚合函数（Aggregate Functions）**

聚合函数是以表中多个行作为输入，提取字段进行聚合操作的函数，会将唯一的聚合值作为结果返回。聚合函数应用非常广泛，不论分组聚合、窗口聚合还是开窗（Over）聚合，对数据的聚合操作都可以用相同的函数来定义。标准 SQL 中常见的聚合函数 Flink SQL 都是支持的，目前也在不断扩展，为流处理应用提供更强大的功能。例如：

*  COUNT(\*) 返回所有行的数量，统计个数
 *  SUM(\[ ALL | DISTINCT \] expression) 对某个字段进行求和操作。默认情况下省略了关键字 ALL，表示对所有行求和；如果指定 DISTINCT，则会对数据进行去重，每个值只叠加一次。
 *  RANK() 返回当前值在一组值中的排名
 *  ROW\_NUMBER() 对一组值排序后，返回当前值的行号。与 RANK()的功能相似

其中，RANK()和 ROW\_NUMBER()一般用在 OVER 窗口中，在之前 11.5.4 小节实现 Top N 的过程中起到了非常重要的作用。

### 2. 自定义函数（UDF） ###

系统函数尽管庞大，也不可能涵盖所有的功能；如果有系统函数不支持的需求，我们就需要用自定义函数（User Defined Functions，UDF）来实现了。事实上，系统内置函数仍然在不断扩充，如果我们认为自己实现的自定义函数足够通用、应用非常广泛，也可以在项目跟踪工具 JIRA 上向 Flink 开发团队提出“议题”（issue），请求将新的函数添加到系统函数中。  
Flink 的 Table API 和 SQL 提供了多种自定义函数的接口，以抽象类的形式定义。当前 UDF主要有以下几类：

*  标量函数（Scalar Functions）：将输入的标量值转换成一个新的标量值；
 *  表函数（Table Functions）：将标量值转换成一个或多个新的行数据，也就是扩展成一个表；
 *  聚合函数（Aggregate Functions）：将多行数据里的标量值转换成一个新的标量值；
 *  表聚合函数（Table Aggregate Functions）：将多行数据里的标量值转换成一个或多个新的行数据。

**1. 整体调用流程**

要想在代码中使用自定义的函数，我们需要首先自定义对应 UDF 抽象类的实现，并在表环境中注册这个函数，然后就可以在 Table API 和 SQL 中调用了。

(1) 注册函数  
注册函数时需要调用表环境的 createTemporarySystemFunction()方法，传入注册的函数名以及 UDF 类的 Class 对象：

// 注册函数
    tableEnv.createTemporarySystemFunction("MyFunction", MyFunction.class);

我们自定义的 UDF 类叫作 MyFunction，它应该是上面四种 UDF 抽象类中某一个的具体实现；在环境中将它注册为名叫 MyFunction 的函数。这里 createTemporarySystemFunction()方法的意思是创建了一个“临时系统函数”，所以MyFunction 函 数 名 是 全 局 的 ， 可 以 当 作 系 统 函 数 来 使 用 ； 我 们 也 可 以 用createTemporaryFunction()方法，注册的函数就依赖于当前的数据库（database）和目录（catalog）了，所以这就不是系统函数，而是“目录函数”（catalog function），它的完整名称应该包括所属的 database 和 catalog。一般情况下，我们直接用 createTemporarySystemFunction()方法将 UDF 注册为系统函数就可以了。

(2) 使用 Table API 调用函数

在 Table API 中，需要使用 call()方法来调用自定义函数：

tableEnv.from("MyTable").select(call("MyFunction", $("myField")));

这里 call()方法有两个参数，一个是注册好的函数名 MyFunction，另一个则是函数调用时本身的参数。这里我们定义 MyFunction 在调用时，需要传入的参数是 myField 字段。此外，在 Table API 中也可以不注册函数，直接用“内联”（inline）的方式调用 UDF：

tableEnv.from("MyTable").select(call(SubstringFunction.class, $("myField")));

区别只是在于 call()方法第一个参数不再是注册好的函数名，而直接就是函数类的 Class对象了。

(3) 在 SQL 中调用函数

当我们将函数注册为系统函数之后，在 SQL 中的调用就与内置系统函数完全一样了：

tableEnv.sqlQuery("SELECT MyFunction(myField) FROM MyTable");

可见，SQL 的调用方式更加方便，我们后续依然会以 SQL 为例介绍 UDF 的用法。  
接下来我们就对不同类型的 UDF 进行展开介绍。

**2. 标量函数（Scalar Functions）**

自定义标量函数可以把 0 个、 1 个或多个标量值转换成一个标量值，它对应的输入是一行数据中的字段，输出则是唯一的值。所以从输入和输出表中行数据的对应关系看，标量函数是“一对一”的转换。

想要实现自定义的标量函数，我们需要自定义一个类来继承抽象类 ScalarFunction，并实现叫作 eval() 的求值方法。标量函数的行为就取决于求值方法的定义，它必须是公有的（public），而且名字必须是 eval。求值方法 eval 可以重载多次，任何数据类型都可作为求值方法的参数和返回值类型。

这里需要特别说明的是，ScalarFunction 抽象类中并没有定义 eval()方法，所以我们不能直接在代码中重写（override）；但 Table API 的框架底层又要求了求值方法必须名字为 eval()。这是 Table API 和 SQL 目前还显得不够完善的地方，未来的版本应该会有所改进。

ScalarFunction 以及其它所有的 UDF 接口，都在 org.apache.flink.table.functions 中。下面我们来看一个具体的例子。我们实现一个自定义的哈希（hash）函数 HashFunction，用来求传入对象的哈希值。

public static class HashFunction extends ScalarFunction {
        
    	// 接受任意类型输入，返回 INT 型输出
    	public int eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
        
    		return o.hashCode();
    	}
    }
    // 注册函数
    tableEnv.createTemporarySystemFunction("HashFunction", HashFunction.class);
    // 在 SQL 里调用注册好的函数
    tableEnv.sqlQuery("SELECT HashFunction(myField) FROM MyTable");

这里我们自定义了一个 ScalarFunction，实现了 eval()求值方法，将任意类型的对象传入，得到一个 Int 类型的哈希值返回。当然，具体的求哈希操作就省略了，直接调用对象的 hashCode()方法即可。

另外注意，由于 Table API 在对函数进行解析时需要提取求值方法参数的类型引用，所以我们用 DataTypeHint(inputGroup = InputGroup.ANY)对输入参数的类型做了标注，表示 eval 的参数可以是任意类型。

**3. 表函数（Table Functions）**

跟标量函数一样，表函数的输入参数也可以是 0 个、1 个或多个标量值；不同的是，它可以返回任意多行数据。“多行数据”事实上就构成了一个表，所以“表函数”可以认为就是返回一个表的函数，这是一个“一对多”的转换关系。之前我们介绍过的窗口 TVF，本质上就是表函数。

类似地，要实现自定义的表函数，需要自定义类来继承抽象类 TableFunction，内部必须要实现的也是一个名为 eval 的求值方法。与标量函数不同的是，TableFunction 类本身是有一个泛型参数T 的，这就是表函数返回数据的类型；而 eval()方法没有返回类型，内部也没有 return语句，是通过调用 collect()方法来发送想要输出的行数据的。多么熟悉的感觉——回忆一下DataStream API 中的 FlatMapFunction 和 ProcessFunction，它们的 flatMap 和 processElement 方法也没有返回值，也是通过 out.collect()来向下游发送数据的。

我们使用表函数，可以对一行数据得到一个表，这和 Hive 中的 UDTF 非常相似。那对于原先输入的整张表来说，又该得到什么呢？一个简单的想法是，就让输入表中的每一行，与它转换得到的表进行联结（join），然后再拼成一个完整的大表，这就相当于对原来的表进行了扩展。在 Hive 的 SQL 语法中，提供了“侧向视图”（lateral view，也叫横向视图）的功能，可以将表中的一行数据拆分成多行；Flink SQL 也有类似的功能，是用 LATERAL TABLE 语法来实现的。

在 SQL 中调用表函数，需要使用 LATERAL TABLE()来生成扩展的“侧向表”，然后与原始表进行联结（Join）。这里的 Join 操作可以是直接做交叉联结（cross join），在 FROM 后用逗号分隔两个表就可以；也可以是以 ON TRUE 为条件的左联结（LEFT JOIN）。

下面是表函数的一个具体示例。我们实现了一个分隔字符串的函数 SplitFunction，可以将一个字符串转换成（字符串，长度）的二元组。

// 注意这里的类型标注，输出是 Row 类型，Row 中包含两个字段：word 和 length。
    @FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
    public static class SplitFunction extends TableFunction<Row> {
        
    	public void eval(String str) {
        
    		for (String s : str.split(" ")) {
        
    			// 使用 collect()方法发送一行数据
    			collect(Row.of(s, s.length()));
    		}
    	}
    }
    // 注册函数
    tableEnv.createTemporarySystemFunction("SplitFunction", SplitFunction.class);
    // 在 SQL 里调用注册好的函数
    // 1. 交叉联结
    tableEnv.sqlQuery(
     "SELECT myField, word, length " +
     "FROM MyTable, LATERAL TABLE(SplitFunction(myField))");
    // 2. 带 ON TRUE 条件的左联结
    tableEnv.sqlQuery(
     "SELECT myField, word, length " +
     "FROM MyTable " +
     "LEFT JOIN LATERAL TABLE(SplitFunction(myField)) ON TRUE");
    // 重命名侧向表中的字段
    tableEnv.sqlQuery(
     "SELECT myField, newWord, newLength " +
     "FROM MyTable " +
     "LEFT JOIN LATERAL TABLE(SplitFunction(myField)) AS T(newWord, newLength) ON TRUE");

这里我们直接将表函数的输出类型定义成了 ROW，这就是得到的侧向表中的数据类型；每行数据转换后也只有一行。我们分别用交叉联结和左联结两种方式在 SQL 中进行了调用，还可以对侧向表的中字段进行重命名。

**4. 聚合函数（Aggregate Functions）**

用户自定义聚合函数（User Defined AGGregate function，UDAGG）会把一行或多行数据（也就是一个表）聚合成一个标量值。这是一个标准的“多对一”的转换。

聚合函数的概念我们之前已经接触过多次，如 SUM()、MAX()、MIN()、AVG()、COUNT()都是常见的系统内置聚合函数。而如果有些需求无法直接调用系统函数解决，我们就必须自定义聚合函数来实现功能了。自定义聚合函数需要继承抽象类 AggregateFunction。AggregateFunction 有两个泛型参数<T, ACC>，T 表示聚合输出的结果类型，ACC 则表示聚合的中间状态类型。  
Flink SQL 中的聚合函数的工作原理如下：  
（1）首先，它需要创建一个累加器（accumulator），用来存储聚合的中间结果。这与DataStream API 中的 AggregateFunction 非常类似，累加器就可以看作是一个聚合状态。调用createAccumulator()方法可以创建一个空的累加器。  
（2）对于输入的每一行数据，都会调用 accumulate()方法来更新累加器，这是聚合的核心过程。  
（3）当所有的数据都处理完之后，通过调用 getValue()方法来计算并返回最终的结果。

所以，每个 AggregateFunction 都必须实现以下几个方法：

*  createAccumulator()  
    这是创建累加器的方法。没有输入参数，返回类型为累加器类型 ACC。
 *  accumulate()  
    这是进行聚合计算的核心方法，每来一行数据都会调用。它的第一个参数是确定的，就是当前的累加器，类型为 ACC，表示当前聚合的中间状态；后面的参数则是聚合函数调用时传入的参数，可以有多个，类型也可以不同。这个方法主要是更新聚合状态，所以没有返回类型。需要注意的是，accumulate()与之前的求值方法 eval()类似，也是底层架构要求的，必须为 public，方法名必须为 accumulate，且无法直接 override、只能手动实现。
 *  getValue()  
    这是得到最终返回结果的方法。输入参数是 ACC 类型的累加器，输出类型为 T。

在遇到复杂类型时，Flink 的类型推导可能会无法得到正确的结果。所以AggregateFunction也可以专门对累加器和返回结果的类型进行声明，这是通过 getAccumulatorType()和getResultType()两个方法来指定的。

除了上面的方法，还有几个方法是可选的。这些方法有些可以让查询更加高效，有些是在某些特定场景下必须要实现的。比如，如果是对会话窗口进行聚合，merge()方法就是必须要实现的，它会定义累加器的合并操作，而且这个方法对一些场景的优化也很有用；而如果聚合函数用在 OVER 窗口聚合中，就必须实现 retract()方法，保证数据可以进行撤回操作；resetAccumulator()方法则是重置累加器，这在一些批处理场景中会比较有用。AggregateFunction 的所有方法都必须是 公有的（public），不能是静态的（static），而且名字必须跟上面写的完全一样。 createAccumulator 、 getValue 、 getResultType 以 及getAccumulatorType 这几个方法是在抽象类 AggregateFunction 中定义的，可以 override；而其他则都是底层架构约定的方法。

下面举一个具体的示例。在常用的系统内置聚合函数里，可以用 AVG()来计算平均值；如果我们现在希望计算的是某个字段的“加权平均值”，又该怎么做呢？系统函数里没有现成的实现，所以只能自定义一个聚合函数 WeightedAvg 来计算了。比如我们要从学生的分数表 ScoreTable 中计算每个学生的加权平均分。为了计算加权平均值，应该从输入的每行数据中提取两个值作为参数：要计算的分数值 score，以及它的权重weight。而在聚合过程中，累加器（accumulator）需要存储当前的加权总和 sum，以及目前数据的个数 count。这可以用一个二元组来表示，也可以单独定义一个类 WeightedAvgAccum，里面包含 sum 和 count 两个属性，用它的对象实例来作为聚合的累加器。

具体代码如下：

// 累加器类型定义
    public static class WeightedAvgAccumulator {
        
    	public long sum = 0; // 加权和
    	public int count = 0; // 数据个数
    }

// 自定义聚合函数，输出为长整型的平均值，累加器类型为 WeightedAvgAccumulator
    public static class WeightedAvg extends AggregateFunction<Long, WeightedAvgAccumulator> {
        
    	@Override
    	public WeightedAvgAccumulator createAccumulator() {
        
    		return new WeightedAvgAccumulator(); // 创建累加器
    	}
    	@Override
    	public Long getValue(WeightedAvgAccumulator acc) {
        
    		if (acc.count == 0) {
        
    			return null; // 防止除数为 0
    		} else {
        
    			return acc.sum / acc.count; // 计算平均值并返回
    		}
    	}
    	// 累加计算方法，每来一行数据都会调用
    	public void accumulate(WeightedAvgAccumulator acc, Long iValue, Integer iWeight) {
        
    		acc.sum += iValue * iWeight;
    		acc.count += iWeight;
    	}
    }
    // 注册自定义聚合函数
    tableEnv.createTemporarySystemFunction("WeightedAvg", WeightedAvg.class);
    // 调用函数计算加权平均值
    Table result = tableEnv.sqlQuery(
    	"SELECT student, WeightedAvg(score, weight) FROM ScoreTable GROUP BY student"
    );

聚合函数的 accumulate()方法有三个输入参数。第一个是 WeightedAvgAccum 类型的累加器；另外两个则是函数调用时输入的字段：要计算的值 ivalue 和 对应的权重 iweight。这里我们并不考虑其它方法的实现，只要有必须的三个方法就可以了。

**5. 表聚合函数（Table Aggregate Functions）**

用户自定义表聚合函数（UDTAGG）可以把一行或多行数据（也就是一个表）聚合成另一张表，结果表中可以有多行多列。很明显，这就像表函数和聚合函数的结合体，是一个“多对多”的转换。

自定义表聚合函数需要继承抽象类 TableAggregateFunction。TableAggregateFunction 的结构和原理与 AggregateFunction 非常类似，同样有两个泛型参数<T, ACC>，用一个 ACC 类型的累加器（accumulator）来存储聚合的中间结果。聚合函数中必须实现的三个方法，在TableAggregateFunction 中也必须对应实现：

*  createAccumulator()  
    创建累加器的方法，与 AggregateFunction 中用法相同。
 *  accumulate()  
    聚合计算的核心方法，与 AggregateFunction 中用法相同。
 *  emitValue()  
    所有输入行处理完成后，输出最终计算结果的方法。这个方法对应着 AggregateFunction中的 getValue()方法；区别在于 emitValue 没有输出类型，而输入参数有两个：第一个是 ACC类型的累加器，第二个则是用于输出数据的“收集器”out，它的类型为 Collect。所以很明显，表聚合函数输出数据不是直接 return，而是调用 out.collect()方法，调用多次就可以输出多行数据了；这一点与表函数非常相似。另外，emitValue()在抽象类中也没有定义，无法 override，必须手动实现。

表聚合函数得到的是一张表；在流处理中做持续查询，应该每次都会把这个表重新计算输出。如果输入一条数据后，只是对结果表里一行或几行进行了更新（Update），这时我们重新计算整个表、全部输出显然就不够高效了。为了提高处理效率，TableAggregateFunction 还提供了一个 emitUpdateWithRetract()方法，它可以在结果表发生变化时，以“撤回”（retract）老数据、发送新数据的方式增量地进行更新。如果同时定义了 emitValue()和 emitUpdateWithRetract()两个方法，在进行更新操作时会优先调用 emitUpdateWithRetract()。表聚合函数相对比较复杂，它的一个典型应用场景就是 Top N 查询。比如我们希望选出一组数据排序后的前两名，这就是最简单的 TOP-2 查询。没有线程的系统函数，那么我们就可以自定义一个表聚合函数来实现这个功能。在累加器中应该能够保存当前最大的两个值，每当来一条新数据就在 accumulate()方法中进行比较更新，最终在 emitValue()中调用两次out.collect()将前两名数据输出。

具体代码如下：

// 聚合累加器的类型定义，包含最大的第一和第二两个数据
    public static class Top2Accumulator {
        
    	public Integer first;
    	public Integer second;
    }
    // 自定义表聚合函数，查询一组数中最大的两个，返回值为(数值，排名)的二元组
    public static class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accumulator> {
        
    	@Override
    	public Top2Accumulator createAccumulator() {
        
    		Top2Accumulator acc = new Top2Accumulator();
    		acc.first = Integer.MIN_VALUE; // 为方便比较，初始值给最小值
    		acc.second = Integer.MIN_VALUE;
    		return acc;
    	}
    	// 每来一个数据调用一次，判断是否更新累加器
    	public void accumulate(Top2Accumulator acc, Integer value) {
        
    		if (value > acc.first) {
        
    			acc.second = acc.first;
    			acc.first = value;
    		} else if (value > acc.second) {
        
    			acc.second = value;
    		}
    	}
    	// 输出(数值，排名)的二元组，输出两行数据
    	public void emitValue(Top2Accumulator acc, Collector<Tuple2<Integer, Integer>> out) {
        
    		if (acc.first != Integer.MIN_VALUE) {
        
    			out.collect(Tuple2.of(acc.first, 1));
    		}
    		if (acc.second != Integer.MIN_VALUE) {
        
    			out.collect(Tuple2.of(acc.second, 2));
    		}
    	}
    }

目前 SQL 中没有直接使用表聚合函数的方式，所以需要使用 Table API 的方式来调用：

// 注册表聚合函数函数
    tableEnv.createTemporarySystemFunction("Top2", Top2.class);
    // 在 Table API 中调用函数
    tableEnv.from("MyTable")
    .groupBy($("myField"))
    .flatAggregate(call("Top2", $("value")).as("value", "rank"))
    .select($("myField"), $("value"), $("rank"));

这里使用了 flatAggregate()方法，它就是专门用来调用表聚合函数的接口。对 MyTable 中数据按 myField 字段进行分组聚合，统计 value 值最大的两个；并将聚合结果的两个字段重命名为 value 和 rank，之后就可以使用 select()将它们提取出来了。

## 八、SQL 客户端 ##

有了 Table API 和 SQL，我们就可以使用熟悉的 SQL 来编写查询语句进行流处理了。不过，这种方式还是将 SQL 语句嵌入到 Java/Scala 代码中进行的；另外，写完的代码后想要提交作业还需要使用工具进行打包。这都给 Flink 的使用设置了门槛，如果不是 Java/Scala 程序员，即使是非常熟悉 SQL 的工程师恐怕也会望而生畏了。

基于这样的考虑，Flink 为我们提供了一个工具来进行 Flink 程序的编写、测试和提交，这工具叫作“SQL 客户端”。SQL 客户端提供了一个命令行交互界面（CLI），我们可以在里面非常容易地编写 SQL 进行查询，就像使用 MySQL 一样；整个 Flink 应用编写、提交的过程全变成了写 SQL，不需要写一行 Java/Scala 代码。

具体使用流程如下：

(1) 首先启动本地集群

./bin/start-cluster.sh

(2) 启动 Flink SQL 客户端

./bin/sql-client.sh

SQL 客户端的启动脚本同样位于 Flink 的 bin 目录下。默认的启动模式是 embedded，也就是说客户端是一个嵌入在本地的进程，这是目前唯一支持的模式。未来会支持连接到远程 SQL客户端的模式。

(3) 设置运行模式

启动客户端后，就进入了命令行界面，这时就可以开始写 SQL 了。一般我们会在开始之前对环境做一些设置，比较重要的就是运行模式。

首先是表环境的运行时模式，有流处理和批处理两个选项。默认为流处理：

Flink SQL> SET 'execution.runtime-mode' = 'streaming';

其次是 SQL 客户端的“执行结果模式”，主要有 table、changelog、tableau 三种，默认为table 模式：

Flink SQL> SET 'sql-client.execution.result-mode' = 'table';

table 模式就是最普通的表处理模式，结果会以逗号分隔每个字段；changelog 则是更新日志模式，会在数据前加上“+”（表示插入）或“-”（表示撤回）的前缀；而 tableau 则是经典的可视化表模式，结果会是一个虚线框的表格。

此外我们还可以做一些其它可选的设置，比如之前提到的空闲状态生存时间（TTL）：

Flink SQL> SET 'table.exec.state.ttl' = '1000';

除了在命令行进行设置，我们也可以直接在 SQL 客户端的配置文件 sql-cli-defaults.yaml中进行各种配置，甚至还可以在这个 yaml 文件里预定义表、函数和 catalog。关于配置文件的更多用法，大家可以查阅官网的详细说明。

(4) 执行 SQL 查询

接下来就可以愉快的编写 SQL 语句了，这跟操作 MySQL、Oracle 等关系型数据库没什么区别。  
我们可以尝试把一开始举的简单聚合例子写一下：

Flink SQL> CREATE TABLE EventTable(
    > 	user STRING,
    >	 url STRING,
    > 	`timestamp` BIGINT
    > ) WITH (
    > 	'connector' = 'filesystem',
    > 	'path' = 'events.csv',
    > 	'format' = 'csv'
    > );
    Flink SQL> CREATE TABLE ResultTable (
    >	 user STRING,
    > 	 cnt BIGINT
    > ) WITH (
    > 	'connector' = 'print'
    > );
    Flink SQL> INSERT INTO ResultTable SELECT user, COUNT(url) as cnt FROM EventTable GROUP BY user;

这里我们直接用 DDL 创建两张表，注意需要有 WITH 定义的外部连接。一张表叫作EventTable，是从外部文件 events.csv 中读取数据的，这是输入数据表；另一张叫作 ResultTable，连接器为“print”，其实就是标准控制台打印，当然就是输出表了。所以接下来就可以直接执行 SQL 查询，并将查询结果 INSERT 写入结果表中了。

在 SQL 客户端中，每定义一个 SQL 查询，就会把它作为一个 Flink 作业提交到集群上执行。所以通过这种方式，我们可以快速地对流处理程序进行开发测试。

## 九、连接到外部系统 ##

在 Table API 和 SQL 编写的 Flink 程序中，可以在创建表的时候用 WITH 子句指定连接器（connector），这样就可以连接到外部系统进行数据交互了。架构中的 TableSource 负责从外部系统中读取数据并转换成表，TableSink 则负责将结果表写入外部系统。在 Flink 1.13 的 API 调用中，已经不去区分 TableSource 和 TableSink，我们只要建立到外部系统的连接并创建表就可以，Flink 自动会从程序的处理逻辑中解析出它们的用途。

Flink 的 Table API 和 SQL 支持了各种不同的连接器。当然，最简单的其实就是上一节中提到的连接到控制台打印输出：

CREATE TABLE ResultTable (
    	user STRING,
    	cnt BIGINT
    WITH (
    	'connector' = 'print'
    );

这里只需要在 WITH 中定义 connector 为 print 就可以了。而对于其它的外部系统，则需要增加一些配置项。下面我们就分别进行讲解。

### 1. Kafka ###

Kafka 的 SQL 连接器可以从 Kafka 的主题（topic）读取数据转换成表，也可以将表数据写入 Kafka 的主题。换句话说，创建表的时候指定连接器为 Kafka，则这个表既可以作为输入表，也可以作为输出表。

**1. 引入依赖**

想要在 Flink 程序中使用 Kafka 连接器，需要引入如下依赖：

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
     <version>${flink.version}</version>
    </dependency>

这里我们引入的 Flink 和 Kafka 的连接器，与之前 DataStream API 中引入的连接器是一样的。如果想在 SQL 客户端里使用 Kafka 连接器，还需要下载对应的 jar 包放到 lib 目录下。

另外，Flink 为各种连接器提供了一系列的“表格式”（table formats），比如 CSV、JSON、Avro、Parquet 等等。这些表格式定义了底层存储的二进制数据和表的列之间的转换方式，相当于表的序列化工具。对于 Kafka 而言，CSV、JSON、Avro 等主要格式都是支持的，根据 Kafka 连接器中配置的格式，我们可能需要引入对应的依赖支持。以 CSV 为例：

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-csv</artifactId>
     <version>${flink.version}</version>
    </dependency>

由于 SQL 客户端中已经内置了 CSV、JSON 的支持，因此使用时无需专门引入；而对于没有内置支持的格式（比如 Avro），则仍然要下载相应的 jar 包。关于连接器的格式细节详见官网说明，我们后面就不再讨论了。

**2. 创建连接到 Kafka 的表**

创建一个连接到 Kafka 表，需要在 CREATE TABLE 的 DDL 中在 WITH 子句里指定连接器为 Kafka，并定义必要的配置参数。

下面是一个具体示例：

CREATE TABLE KafkaTable (
    `user` STRING,
     `url` STRING,
     `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
    ) WITH (
     'connector' = 'kafka',
     'topic' = 'events',
     'properties.bootstrap.servers' = 'localhost:9092',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'earliest-offset',
     'format' = 'csv'
    )

这里定义了 Kafka 连接器对应的主题（topic），Kafka 服务器，消费者组 ID，消费者起始模式以及表格式。需要特别说明的是，在 KafkaTable 的字段中有一个 ts，它的声明中用到了METADATA FROM，这是表示一个“元数据列”（metadata column），它是由 Kafka 连接器的元数据“timestamp”生成的。这里的 timestamp 其实就是 Kafka 中数据自带的时间戳，我们把它直接作为元数据提取出来，转换成一个新的字段 ts。

**3. Upsert Kafka**

正常情况下，Kafka 作为保持数据顺序的消息队列，读取和写入都应该是流式的数据，对应在表中就是仅追加（append-only）模式。如果我们想要将有更新操作（比如分组聚合）的结果表写入 Kafka，就会因为 Kafka 无法识别撤回（retract）或更新插入（upsert）消息而导致异常。为了解决这个问题，Flink 专门增加了一个“更新插入 Kafka”（Upsert Kafka）连接器。这个连接器支持以更新插入（UPSERT）的方式向 Kafka 的 topic 中读写数据。

具体来说，Upsert Kafka 连接器处理的是更新日志（changlog）流。如果作为 TableSource，连接器会将读取到的 topic中的数据（key, value），解释为对当前 key 的数据值的更新（UPDATE），也就是查找动态表中 key 对应的一行数据，将 value 更新为最新的值；因为是 Upsert 操作，所以如果没有 key 对应的行，那么也会执行插入（INSERT）操作。另外，如果遇到 value 为空（null），连接器就把这条数据理解为对相应 key 那一行的删除（DELETE）操作。

如果作为 TableSink，Upsert Kafka 连接器会将有更新操作的结果表，转换成更新日志（changelog）流。如果遇到插入（INSERT）或者更新后（UPDATE\_AFTER）的数据，对应的是一个添加（add）消息，那么就直接正常写入 Kafka 主题；如果是删除（DELETE）或者更新前的数据，对应是一个撤回（retract）消息，那么就把 value 为空（null）的数据写入 Kafka。由于 Flink 是根据键（key）的值对数据进行分区的，这样就可以保证同一个 key 上的更新和删除消息都会落到同一个分区中。

下面是一个创建和使用 Upsert Kafka 表的例子：

CREATE TABLE pageviews_per_region (
     user_region STRING,
     pv BIGINT,
     uv BIGINT,
     PRIMARY KEY (user_region) NOT ENFORCED
    ) WITH (
     'connector' = 'upsert-kafka',
     'topic' = 'pageviews_per_region',
     'properties.bootstrap.servers' = '...',
     'key.format' = 'avro',
     'value.format' = 'avro'
    );
    CREATE TABLE pageviews (
     user_id BIGINT,
     page_id BIGINT,
     viewtime TIMESTAMP,
     user_region STRING,
     WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
     ) WITH (
     'connector' = 'kafka',
     'topic' = 'pageviews',
     'properties.bootstrap.servers' = '...',
     'format' = 'json'
    );
    -- 计算 pv、uv 并插入到 upsert-kafka 表中
    INSERT INTO pageviews_per_region
    SELECT
     user_region,
     COUNT(*),
     COUNT(DISTINCT user_id)
    FROM pageviews
    GROUP BY user_region;

这里我们从 Kafka 表 pageviews 中读取数据，统计每个区域的 PV（全部浏览量）和 UV（对用户去重），这是一个分组聚合的更新查询，得到的结果表会不停地更新数据。为了将结果表写入 Kafka 的 pageviews\_per\_region 主题，我们定义了一个 Upsert Kafka 表，它的字段中需要用PRIMARY KEY来指定主键，并且在WITH子句中分别指定key和value的序列化格式。

### 2. 文件系统 ###

另一类非常常见的外部系统就是文件系统（File System）了。Flink 提供了文件系统的连接器，支持从本地或者分布式的文件系统中读写数据。这个连接器是内置在 Flink 中的，所以使用它并不需要额外引入依赖。

下面是一个连接到文件系统的示例：

CREATE TABLE MyTable (
     column_name1 INT,
     column_name2 STRING,
     ...
     part_name1 INT,
     part_name2 STRING
    ) PARTITIONED BY (part_name1, part_name2) WITH (
     'connector' = 'filesystem', -- 连接器类型
     'path' = '...', -- 文件路径
     'format' = '...' -- 文件格式
    )

这里在 WITH 前使用了 PARTITIONED BY 对数据进行了分区操作。文件系统连接器支持对分区文件的访问。

### 3. JDBC ###

关系型数据表本身就是 SQL 最初应用的地方，所以我们也会希望能直接向关系型数据库中读写表数据。Flink 提供的 JDBC 连接器可以通过 JDBC 驱动程序（driver）向任意的关系型数据库读写数据，比如 MySQL、PostgreSQL、Derby 等。

作为 TableSink 向数据库写入数据时，运行的模式取决于创建表的 DDL 是否定义了主键（primary key）。如果有主键，那么 JDBC 连接器就将以更新插入（Upsert）模式运行，可以向外部数据库发送按照指定键（key）的更新（UPDATE）和删除（DELETE）操作；如果没有定义主键，那么就将在追加（Append）模式下运行，不支持更新和删除操作。  
**1. 引入依赖**

想要在 Flink 程序中使用 JDBC 连接器，需要引入如下依赖：

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
     <version>${flink.version}</version>
    </dependency>

此外，为了连接到特定的数据库，我们还用引入相关的驱动器依赖，比如 MySQL：

<dependency>
     <groupId>mysql</groupId>
     <artifactId>mysql-connector-java</artifactId>
     <version>5.1.38</version>
    </dependency>

这里引入的驱动器版本是 5.1.38，读者可以依据自己的 MySQL 版本来进行选择。

**2. 创建 JDBC 表**

创建 JDBC 表的方法与前面 Upsert Kafka 大同小异。下面是一个具体示例：

-- 创建一张连接到 MySQL 的 表
    CREATE TABLE MyTable (
     id BIGINT,
     name STRING,
     age INT,
     status BOOLEAN,
     PRIMARY KEY (id) NOT ENFORCED
    ) WITH (
     'connector' = 'jdbc',
     'url' = 'jdbc:mysql://localhost:3306/mydatabase',
     'table-name' = 'users'
    );
    -- 将另一张表 T 的数据写入到 MyTable 表中
    INSERT INTO MyTable
    SELECT id, name, age, status FROM T;

这里创建表的 DDL 中定义了主键，所以数据会以 Upsert 模式写入到 MySQL 表中；而到MySQL 的连接，是通过 WITH 子句中的 url 定义的。要注意写入 MySQL 中真正的表名称是users，而 MyTable 是注册在 Flink 表环境中的表。

### 4. Elasticsearch ###

Elasticsearch 作为分布式搜索分析引擎，在大数据应用中有非常多的场景。Flink 提供的  
Elasticsearch的SQL连接器只能作为TableSink，可以将表数据写入Elasticsearch的索引（index）。  
Elasticsearch 连接器的使用与 JDBC 连接器非常相似，写入数据的模式同样是由创建表的 DDL  
中是否有主键定义决定的。

**1. 引入依赖**

想要在 Flink 程序中使用 Elasticsearch 连接器，需要引入对应的依赖。具体的依赖与Elasticsearch 服务器的版本有关，对于 6.x 版本引入依赖如下：

<dependency>
    	<groupId>org.apache.flink</groupId> 
    	<artifactId>flink-connector-elasticsearch6_${scala.binary.version}</artifactId>
    	<version>${flink.version}</version>
    </dependency>

对于 Elasticsearch 7 以上的版本，引入的依赖则是：

<dependency>
    	<groupId>org.apache.flink</groupId> 
    	<artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
    	<version>${flink.version}</version>
    </dependency>

**2. 创建连接到 Elasticsearch 的表**

创建 Elasticsearch 表的方法与 JDBC 表基本一致。下面是一个具体示例：

-- 创建一张连接到 Elasticsearch 的 表
    CREATE TABLE MyTable (
     user_id STRING,
     user_name STRING
     uv BIGINT,
     pv BIGINT,
     PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
     'connector' = 'elasticsearch-7',
     'hosts' = 'http://localhost:9200',
     'index' = 'users'
    );

这里定义了主键，所以会以更新插入（Upsert）模式向 Elasticsearch 写入数据。

### 5. HBase ###

作为高性能、可伸缩的分布式列存储数据库，HBase 在大数据分析中是一个非常重要的工具。Flink 提供的 HBase 连接器支持面向 HBase 集群的读写操作。

在流处理场景下，连接器作为 TableSink 向 HBase 写入数据时，采用的始终是更新插入（Upsert）模式。也就是说，HBase 要求连接器必须通过定义的主键（primary key）来发送更新日志（changelog）。所以在创建表的 DDL 中，我们必须要定义行键（rowkey）字段，并将它声明为主键；如果没有用 PRIMARY KEY 子句声明主键，连接器会默认把 rowkey 作为主键。

**1. 引入依赖**  
想要在 Flink 程序中使用 HBase 连接器，需要引入对应的依赖。目前 Flink 只对 HBase 的1.4.x 和 2.2.x 版本提供了连接器支持，而引入的依赖也应该与具体的 HBase 版本有关。对于1.4 版本引入依赖如下：

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-connector-hbase-1.4_${scala.binary.version}</artifactId>
     <version>${flink.version}</version>
    </dependency>

对于 HBase 2.2 版本，引入的依赖则是：

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-connector-hbase-2.2_${scala.binary.version}</artifactId>
     <version>${flink.version}</version>
    </dependency>

**2. 创建连接到 HBase 的表**

由于 HBase 并不是关系型数据库，因此转换为 Flink SQL 中的表会稍有一些麻烦。在 DDL创建出的 HBase 表中，所有的列族（column family）都必须声明为 ROW 类型，在表中占据一个字段；而每个 family 中的列（column qualifier）则对应着 ROW 里的嵌套字段。我们不需要将 HBase 中所有的 family 和 qualifier 都在 Flink SQL 的表中声明出来，只要把那些在查询中用到的声明出来就可以了。

除了所有 ROW 类型的字段（对应着 HBase 中的 family），表中还应有一个原子类型的字段，它就会被识别为 HBase 的 rowkey。在表中这个字段可以任意取名，不一定非要叫 rowkey。

下面是一个具体示例：

-- 创建一张连接到 HBase 的 表
    CREATE TABLE MyTable (
    rowkey INT,
    family1 ROW<q1 INT>,
    family2 ROW<q2 STRING, q3 BIGINT>,
    family3 ROW<q4 DOUBLE, q5 BOOLEAN, q6 STRING>,
    PRIMARY KEY (rowkey) NOT ENFORCED
    ) WITH (
    'connector' = 'hbase-1.4',
    'table-name' = 'mytable',
    'zookeeper.quorum' = 'localhost:2181'
    );
    -- 假设表 T 的字段结构是 [rowkey, f1q1, f2q2, f2q3, f3q4, f3q5, f3q6]
    INSERT INTO MyTable
    SELECT rowkey, ROW(f1q1), ROW(f2q2, f2q3), ROW(f3q4, f3q5, f3q6) FROM T;

我们将另一张 T 中的数据提取出来，并用 ROW()函数来构造出对应的 column family，最终写入 HBase 中名为 mytable 的表。

### 6. Hive ###

Apache Hive 作为一个基于 Hadoop 的数据仓库基础框架，可以说已经成为了进行海量数据分析的核心组件。Hive 支持类 SQL 的查询语言，可以用来方便对数据进行处理和统计分析，而且基于 HDFS 的数据存储有非常好的可扩展性，是存储分析超大量数据集的唯一选择。Hive的主要缺点在于查询的延迟很高，几乎成了离线分析的代言人。而 Flink 的特点就是实时性强，所以 Flink SQL 与 Hive 的结合势在必行。

Flink 与 Hive 的集成比较特别。Flink 提供了“Hive 目录”（HiveCatalog）功能，允许使用Hive 的“元存储”(Metastore) 来管理 Flink 的元数据。这带来的好处体现在两个方面：

(1) Metastore 可以作为一个持久化的目录，因此使用 HiveCatalog 可以跨会话存储 Flink特定的元数据。这样一来，我们在 HiveCatalog 中执行执行创建 Kafka 表或者 ElasticSearch 表，就可以把它们的元数据持久化存储在 Hive 的 Metastore 中；对于不同的作业会话就不需要重复创建了，直接在 SQL 查询中重用就可以。

(2) 使用 HiveCatalog，Flink 可以作为读写 Hive 表的替代分析引擎。这样一来，在 Hive中进行批处理会更加高效；与此同时，也有了连续在 Hive 中读写数据、进行流处理的能力，这也使得“实时数仓”（real-time data warehouse）成为了可能。HiveCatalog 被设计为“开箱即用”，与现有的 Hive 配置完全兼容，我们不需要做任何的  
修改与调整就可以直接使用。注意只有 Blink 的计划器（planner）提供了 Hive 集成的支持，所以需要在使用 Flink SQL时选择Blink planner。下面我们就来看以下与Hive 集成的具体步骤。

**1. 引入依赖**

Hive 各版本特性变化比较大，所以使用时需要注意版本的兼容性。目前 Flink 支持的 Hive版本包括：

*  Hive 1.x：1.0.0~1.2.2；
 *  Hive 2.x：2.0.02.2.0，2.3.02.3.6；
 *  Hive 3.x：3.0.0~3.1.2；

目前 Flink 与 Hive 的集成程度在持续加强，支持的版本信息也会不停变化和调整，大家可以随着关注官网的更新信息。

由于 Hive 是基于 Hadoop 的组件，因此我们首先需要提供 Hadoop 的相关支持，在环境变量中设置

HADOOP_CLASSPATH：
    export HADOOP_CLASSPATH=`hadoop classpath`

在 Flink 程序中可以引入以下依赖：

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
     <version>${flink.version}</version>
    </dependency>
    
    <dependency>
     <groupId>org.apache.hive</groupId>
     <artifactId>hive-exec</artifactId>
     <version>${hive.version}</version>
    </dependency>

建议不要把这些依赖打包到结果 jar 文件中，而是在运行时的集群环境中为不同的 Hive版本添加不同的依赖支持。具体版本对应的依赖关系，可以查询官网说明。

**2. 连接到 Hive**

在 Flink 中连接 Hive，是通过在表环境中配置 HiveCatalog 来实现的。需要说明的是，配置 HiveCatalog 本身并不需要限定使用哪个 planner，不过对 Hive 表的读写操作只有 Blink 的planner 才支持。所以一般我们需要将表环境的 planner 设置为 Blink。

下面是代码中配置 Catalog 的示例：

EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build();
    TableEnvironment tableEnv = TableEnvironment.create(settings);
    String name = "myhive";
    String defaultDatabase = "mydatabase";
    String hiveConfDir = "/opt/hive-conf";
    // 创建一个 HiveCatalog，并在表环境中注册
    HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
    tableEnv.registerCatalog("myhive", hive);
    // 使用 HiveCatalog 作为当前会话的 catalog
    tableEnv.useCatalog("myhive");

当然，我们也可以直接启动 SQL 客户端，用 CREATE CATALOG 语句直接创建 HiveCatalog：

Flink SQL> create catalog myhive with ('type' = 'hive', 'hive-conf-dir' = '/opt/hive-conf');
    [INFO] Execute statement succeed.
    Flink SQL> use catalog myhive;
    [INFO] Execute statement succeed.

**3. 设置 SQL 方言**

我们知道，Hive内部提供了类SQL的查询语言，不过语法细节与标准SQL会有一些出入，相当于是 SQL 的一种“方言”（dialect）。为了提高与 Hive 集成时的兼容性，Flink SQL 提供了一个非常有趣而强大的功能：可以使用方言来编写 SQL 语句。换句话说，我们可以直接在 Flink中写 Hive SQL 来操作 Hive 表，这无疑给我们的读写处理带来了极大的方便。

Flink 目前支持两种 SQL 方言的配置：default 和 hive。所谓的 default 就是 Flink SQL 默认的 SQL 语法了。我们需要先切换到 hive 方言，然后才能使用 Hive SQL 的语法。具体设置可以分为 SQL 和 Table API 两种方式。

(1) SQL 中设置

我们可以通过配置 table.sql-dialect 属性来设置 SQL 方言：

set table.sql-dialect=hive;

当然，我们可以在代码中执行上面的 SET 语句，也可以直接启动 SQL 客户端来运行。如果使用 SQL 客户端，我们还可以在配置文件 sql-cli-defaults.yaml 中通过“configuration”模块来设置：

execution:
     planner: blink
     type: batch
     result-mode: table
    configuration:
     table.sql-dialect: hive

(2) Table API 中设置  
另外一种方式就是在代码中，直接使用 Table API 获取表环境的配置项来进行设置：

// 配置 hive 方言
    tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
    // 配置 default 方言
    tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);

**4. 读写 Hive 表**

有了 SQL 方言的设置，我们就可以很方便的在 Flink 中创建 Hive 表并进行读写操作了。Flink 支持以批处理和流处理模式向 Hive 中读写数据。在批处理模式下，Flink 会在执行查询语句时对 Hive 表进行一次性读取，在作业完成时将结果数据向 Hive 表进行一次性写入；而在流处理模式下，Flink 会持续监控 Hive 表，在新数据可用时增量读取，也可以持续写入新数据并增量式地让它们可见。

更灵活的是，我们可以随时切换 SQL 方言，从其它数据源（例如 Kafka）读取数据、经转换后再写入Hive。下面是以纯SQL形式编写的一个示例，我们可以启动SQL客户端来运行：

-- 设置 SQL 方言为 hive，创建 Hive 表
    SET table.sql-dialect=hive;
    CREATE TABLE hive_table (
     user_id STRING,
     order_amount DOUBLE
    ) PARTITIONED BY (dt STRING, hr STRING) STORED AS parquet TBLPROPERTIES (
     'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00',
     'sink.partition-commit.trigger'='partition-time',
     'sink.partition-commit.delay'='1 h',
     'sink.partition-commit.policy.kind'='metastore,success-file'
    );
    -- 设置 SQL 方言为 default，创建 Kafka 表
    SET table.sql-dialect=default;
    CREATE TABLE kafka_table (
     user_id STRING,
     order_amount DOUBLE,
     log_ts TIMESTAMP(3),
     WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND – 定义水位线
    ) WITH (...);
    -- 将 Kafka 中读取的数据经转换后写入 Hive 
    INSERT INTO TABLE hive_table 
    SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), 
    DATE_FORMAT(log_ts, 'HH')
    FROM kafka_table;

这里我们创建 Hive 表时设置了通过分区时间来触发提交的策略。将 Kafka 中读取的数据经转换后写入 Hive，这是一个流处理的 Flink SQL 程序。

## 十、总结 ##

在本章中，我们从一个简单示例入手，由浅入深地介绍了 Flink Table API 和 SQL 的用法。由于这两套 API 底层原理一致，而 Table API 功能不够完善、应用不够方便，实际项目开发往往写 SQL 居多；因此本章内容是以 Flink SQL 的各种功能特性为主线贯穿始终，对 Table API 只做原理性讲解。

11.2 节主要介绍 Table API 和 SQL 的基本用法，有了这部分知识，就可以写出完整的 Flink SQL 程序了。11.3 节深入讲解了表和 SQL 在流处理中的一些核心概念，比如动态表和持续查询，更新查询和追加查询等；这些知识或许对于应用逻辑没有太大帮助，然而却是深入理解流式处理架构的关键，也是从程序员向着架构师迈进的路上必须跨越的门槛。11.4~11.6 节主要介绍 Flink SQL 中的高级特性：窗口、聚合查询和联结查询，在这一部分中标准 SQL 语法和Flink 的 DataStream API 彼此渗透融合，在流处理中使用 SQL 查询的特色体现得淋漓尽致；另一方面，这几节也是对 SQL 和 DataStream API 知识的一个总结。11.7 节详细讲解了函数的用法，这部分主要是一个知识扩展，实际应用的场景较少，一般只需要知道系统函数的用法就够了。11.8、11.9 两节介绍了 SQL 客户端工具和外部系统的连接器，内容相对比较简单，主要侧重于实际应用场景。

本章内容较多，如果仅以快速应用为目的，读者可以主要浏览 11.1、11.2、11.4、11.5、11.9 这五节内容；当然如果时间精力充沛，还是建议完整通读，并在官网详细浏览相关资料。

Table API 和 SQL 是 Flink 最上层的应用接口，目前尚不完善，但发展非常迅速，每个小版本都会有底层优化和功能扩展。可以想到不久的将来，Flink SQL 将会是最为高效、最为普遍的开发手段，我们应该时刻保持跟进，随着框架的发展完善不断提升自己的技术能力。

[326fc23db2aa4ebabfc3f28397626a65.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/c9e6764c403649dda60b2984a1097cd7.png
[156cb6d99824400692300ea032f2ffd7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/947ef7283f7c4abfbaa29eb92551de8e.png
[ea260af2da38450aa9af18e680e51a9e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/acfc0485c0f44bb097d257f49ad9e3dd.png
[4b7a9db60cb74b5fb7beaf0be7b3d397.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/a911835c065e4b538dc7b8408b68472e.png
[70ade69d48174394808c35cd5f375d5a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/370092f3690c41fa9a47058e5491abfb.png
[63a594dee3254179aa07c400147a9f77.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/acbdde6eb8ad4ca6b9713d226e2456d9.png
[df4f8d6c24b74bb9a5f9e1006e982d52.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/5fcd69e2d7c049eaa55540aab05a0d0d.png
[7c2679cef5914d1397ebe7c90e4eb98b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/dc200ea6a5f64606a2f4469d817a9727.png