Flink Table API和SQL(上)

概述

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

Flink Table API和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 为我们提供了两种查询方式:SQLTable 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),当作有限数据集进行批处理;流式数据的到来会触发连续不断的快照查询,像动画一样连贯起来,就构成了“持续查询”。
  如图所示,描述了持续查询的过程。这里我们也可以清晰地看到流、动态表和持续查询的关系:

Flink Table API和SQL(上)
持续查询的步骤如下:
(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)操作, 每条数据就是表中的一行;随着插入更多的点击事件,得到的动态表将不断增长。

Flink Table API和SQL(上)

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()方法。

Flink Table API和SQL(上)
具体步骤解释如下:
(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 表后面,而不会更新之前的数据。

Flink Table API和SQL(上)
  所以我们发现,由于窗口的统计结果是一次性写入结果表的,所以结果表的更新日志流中只会包含插入 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 消息。这样,我们可以通过编码后的消息指明所有的增删改操作,一个动态表就可以转换为撤回流了。
  可以看到,更新操作对于撤回流来说,对应着两个消息:之前数据的撤回(删除)和新数据的插入。

Flink Table API和SQL(上)
  这里我们用+代表 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),这样才能正确地处理消息。
    Flink Table API和SQL(上)
      可以看到,更新插入流跟撤回流的主要区别在于,更新(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(上)
0 0 投票数
文章评分

本文转载自落花雨时,原文链接:https://blog.csdn.net/mengxianglong123/article/details/124041915。

(0)
上一篇 2022-07-06 23:14
下一篇 2022-07-07 09:42

相关推荐

订阅评论
提醒
guest
0 评论
内联反馈
查看所有评论
0
希望看到您的想法,请您发表评论x