day6_FlinkSQL实战

FlinkSQL实战

今日课程内容大纲

• Flink SQL Gateway
• Flink SQL 基础概念
• Flink SQL 编程
• 常用 Connector 读写
• sql-client 使用 savepoint
• CateLog

到现在我们学完了底层API（也就是 process）和核心 API（这里由于Flink现在已经流批一体，所以我们只学习 DataStream就好了），然后就是剩下的 Table API（类似于 Spark 中的 DataFrame 和 DataSet）和SQL（类似于Spark SQL）。显然最上层的都是高级 API ，它们的底层还是我们学的这些 DataStream 和 process 算子。不过毕竟是高级 API ，它对 SQL 语句都进行了优化，一般能用 SQL 肯定没人愿意用繁琐的代码去实现，大大降低了开发 Flink 程序的难度，但是一些 SQL 实现不了的东西当然还是得底层核心 API 来实现，就像 Spark 中的 RDD 编程一样。

Flink SQL Gateway

Flink SQL Gateway简介

SQL Gateway 服务支持并发执行从多个client提交的 SQL。它提供了一种简单的方法来提交 Flink 作业、查找元数据和在线分析数据。

SQL Gateway 由插件化的 endpoint 和 SqlGatewayService 组成。多个 endpoint 可以复用 SqlGatewayService 处理请求。endpoint 是用户连接的入口。 用户可以使用不同的工具连接不同类型的 endpoint。

Flink SQL Gateway使用

SQL Gateway 脚本也在 Flink 二进制包的目录中。用户通过以下命令启动：

$ ./bin/sql-gateway.sh start -Dsql-gateway.endpoint.rest.address=node1

这个命令启动 SQL Gateway 和 REST Endpoint，监听 localhost:8083 地址。你可以使用 curl 命令检查 REST Endpoint 是否存活。

$ curl http://node1:8083/v1/info
{"productName":"Apache Flink","version":"1.20.0"}

执行 SQL 查询

你可以通过以下步骤来验证集群配置和连接。

$ curl --request POST http://node1:8083/v1/sessions
{"sessionHandle":"..."}

SQL Gateway 返回结果中的sessionHandle用来唯一标识每个活跃用户。

$ curl --request POST http://node1:8083/v1/sessions/${sessionHandle}/statements/ --data '{"statement": "SELECT 1"}'
{"operationHandle":"..."}

SQL Gateway 返回结果中的 operationHandle 用来唯一标识提交的 SQL。

通过上述sessionHandle和operationHandle，你能获取相应的结果。

$ curl --request GET http://node1:8083/v1/sessions/${sessionHandle}/operations/${operationHandle}/result/0
{
  "results": {
    "columns": [
      {
        "name": "EXPR$0",
        "logicalType": {
          "type": "INTEGER",
          "nullable": false
        }
      }
    ],
    "data": [
      {
        "kind": "INSERT",
        "fields": [
          1
        ]
      }
    ]
  },
  "resultType": "PAYLOAD",
  "nextResultUri": "..."
}

结果中的nextResultUri不是null时，用于获取下一批结果。

$ curl --request GET ${nextResultUri}

配置

目前 SQL Gateway 有以下可选命令，它们将在下文详细讨论。

$ ./bin/sql-gateway.sh --help

Usage: sql-gateway.sh [start|start-foreground|stop|stop-all] [args]
  commands:
    start               - Run a SQL Gateway as a daemon
    start-foreground    - Run a SQL Gateway as a console application
    stop                - Stop the SQL Gateway daemon
    stop-all            - Stop all the SQL Gateway daemons
    -h | --help         - Show this help message

“start” 或者 “start-foreground” 命令可以使你在 CLI 中配置 SQL Gateway。

$ ./bin/sql-gateway.sh start --help

Start the Flink SQL Gateway as a daemon to submit Flink SQL.

  Syntax: start [OPTIONS]
     -D <property=value>   Use value for given property
     -h,--help             Show the help message with descriptions of all
                           options.

SQL Gateway 配置
你可以通过以下方式在启动时配置 SQL Gateway，或者任意合法的 Flink configuration 配置：

$ ./sql-gateway -Dkey=value

Key	Default	Type	Description
sql-gateway.session.check-interval	1 min	Duration	定时检查空闲 session 是否超时的间隔时间，设置为 0 时关闭检查。
sql-gateway.session.idle-timeout	10 min	Duration	session 超时时间，在这个时间区间内没有被访问过的 session 会被关闭。如果设置为 0，session 将不会被关闭。
sql-gateway.session.max-num	1000000	Integer	SQL Gateway 服务中存活 session 的最大数量。
sql-gateway.session.plan-cache.enabled	false	Boolean	设置为 true 的时候，SQL Gateway 会在一个 session 内部缓存并复用 plan。
sql-gateway.session.plan-cache.size	100	Integer	Plan cache 的大小, 当且仅当 table.optimizer.plan-cache.enabled 为 true 的时候生效。
sql-gateway.session.plan-cache.ttl	1 hour	Duration	Plan cache 的 TTL, 控制 cache 在写入之后多久过期, 当且仅当 table.optimizer.plan-cache.enabled 为 true 的时候生效。
sql-gateway.worker.keepalive-time	5 min	Duration	空闲工作线程的存活时间。当工作线程数量超过了配置的最小值，超过存活时间的多余空闲工作线程会被杀掉。
sql-gateway.worker.threads.max	500	Integer	SQL Gateway 服务中工作线程的最大数量。
sql-gateway.worker.threads.min	5	Integer	SQL Gateway 服务中工作线程的最小数量。

如何使用SQL-Client连接Flink

Flink 的 Table & SQL API 可以处理 SQL 语言编写的查询语句，但是这些查询需要嵌入用 Java 或 Scala 编写的表程序中。此外，这些程序在提交到集群前需要用构建工具打包。这或多或少限制了 Java/Scala 程序员对 Flink 的使用。

SQL 客户端 的目的是提供一种简单的方式来编写、调试和提交表程序到 Flink 集群上，而无需写一行 Java 或 Scala 代码。SQL 客户端命令行界面（CLI） 能够在命令行中检索和可视化分布式应用中实时产生的结果。

启动 SQL 客户端命令行界面

SQL Client 脚本也位于 Flink 的 bin 目录中。

#启动flink sql客户端

./bin/sql-client.sh gateway --endpoint node1:8083

#注意：需要使用部署flink集群的用户启动

[root@node1 flink]# ./bin/sql-client.sh gateway --endpoint node1:8083

                                   ▒▓██▓██▒
                               ▓████▒▒█▓▒▓███▓▒
                            ▓███▓░░        ▒▒▒▓██▒  ▒
                          ░██▒   ▒▒▓▓█▓▓▒░      ▒████
                          ██▒         ░▒▓███▒    ▒█▒█▒
                            ░▓█            ███   ▓░▒██
                              ▓█       ▒▒▒▒▒▓██▓░▒░▓▓█
                            █░ █   ▒▒░       ███▓▓█ ▒█▒▒▒
                            ████░   ▒▓█▓      ██▒▒▒ ▓███▒
                         ░▒█▓▓██       ▓█▒    ▓█▒▓██▓ ░█░
                   ▓░▒▓████▒ ██         ▒█    █▓░▒█▒░▒█▒
                  ███▓░██▓  ▓█           █   █▓ ▒▓█▓▓█▒
                ░██▓  ░█░            █  █▒ ▒█████▓▒ ██▓░▒
               ███░ ░ █░          ▓ ░█ █████▒░░    ░█░▓  ▓░
              ██▓█ ▒▒▓▒          ▓███████▓░       ▒█▒ ▒▓ ▓██▓
           ▒██▓ ▓█ █▓█       ░▒█████▓▓▒░         ██▒▒  █ ▒  ▓█▒
           ▓█▓  ▓█ ██▓ ░▓▓▓▓▓▓▓▒              ▒██▓           ░█▒
           ▓█    █ ▓███▓▒░              ░▓▓▓███▓          ░▒░ ▓█
           ██▓    ██▒    ░▒▓▓███▓▓▓▓▓██████▓▒            ▓███  █
          ▓███▒ ███   ░▓▓▒░░   ░▓████▓░                  ░▒▓▒  █▓
          █▓▒▒▓▓██  ░▒▒░░░▒▒▒▒▓██▓░                            █▓
          ██ ▓░▒█   ▓▓▓▓▒░░  ▒█▓       ▒▓▓██▓    ▓▒          ▒▒▓
          ▓█▓ ▓▒█  █▓░  ░▒▓▓██▒            ░▓█▒   ▒▒▒░▒▒▓█████▒
           ██░ ▓█▒█▒  ▒▓▓▒  ▓█                █░      ░░░░   ░█▒
           ▓█   ▒█▓   ░     █░                ▒█              █▓
            █▓   ██         █░                 ▓▓        ▒█▓▓▓▒█░
             █▓ ░▓██░       ▓▒                  ▓█▓▒░░░▒▓█░    ▒█
              ██   ▓█▓░      ▒                    ░▒█▒██▒      ▓▓
               ▓█▒   ▒█▓▒░                         ▒▒ █▒█▓▒▒░░▒██
                ░██▒    ▒▓▓▒                     ▓██▓▒█▒ ░▓▓▓▓▒█▓
                  ░▓██▒                          ▓░  ▒█▓█  ░░▒▒▒
                      ▒▓▓▓▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░▓▓  ▓░▒█░
          
    ______ _ _       _       _____  ____  _         _____ _ _            _  BETA   
   |  ____| (_)     | |     / ____|/ __ \| |       / ____| (_)          | |  
   | |__  | |_ _ __ | | __ | (___ | |  | | |      | |    | |_  ___ _ __ | |_ 
   |  __| | | | '_ \| |/ /  \___ \| |  | | |      | |    | | |/ _ \ '_ \| __|
   | |    | | | | | |   <   ____) | |__| | |____  | |____| | |  __/ | | | |_ 
   |_|    |_|_|_| |_|_|\_\ |_____/ \___\_\______|  \_____|_|_|\___|_| |_|\__|
          
        Welcome! Enter 'HELP;' to list all available commands. 'QUIT;' to exit.

Command history file path: /root/.flink-sql-history

Flink SQL> 

入门案例

#入门案例
select 'Hello World'; 
select 1;

常用配置

Flink-SQL结果可以显示为三种模式：

• table（默认）表模式

set sql-client.execution.result-mode=table;

运行截图如下：

• changelog（变更日志）

set sql-client.execution.result-mode=changelog；

运行截图如下：

• tableau（tableau）

set sql-client.execution.result-mode=tableau;

运行截图如下：

使用SQL-Client工具提交任务

准备表

---构建source表
create table source(
	word varchar
) with (
'connector' = 'socket',
'hostname' = 'node1',
'port' = '9999',
'format' = 'csv'
);
---构建sink表
create table sink (
	word varchar,
	counts bigint
) with (
'connector' = 'print'
);

备注：

如果需要使用socket连接器（‘connector’ = ‘socket’)

则需要使用额外的jar包https://repo.maven.apache.org/maven2/org/apache/flink/flink-examples-table_2.12/1.20.0/flink-examples-table_2.12-1.20.0.jar

这个包在我们资料里

把这个jar包放入到$FLINK_HOME/lib目录下，重启集群和sql Gatewey以及sqlClient，即可。

启动流式任务

insert into sink select word,count(*) from source group by word;

结果如下

FlinkSQL 基础概念

Flink SQL基本原理

Flink SQL建立在Apache Flink之上，利用Flink的强大处理能力，使得用户可以使用SQL语句进行流数据和批数据处理。Flink SQL既支持实时的流数据处理，也支持有界的批数据处理。

Flink SQL用SQL作为处理数据的接口语言，将SQL语句转换成数据流图（Dataflow Graph），再由Flink引擎执行。

Flink SQL 的工作机制

Flink SQL 引擎的工作流总结如图所示。

从图中可以看出，一段查询 SQL / 使用TableAPI 编写的程序（以下简称 TableAPI 代码）从输入到编译为可执行的 JobGraph 主要经历如下几个阶段：

• 将 SQL文本 / TableAPI 代码转化为逻辑执行计划（Logical Plan）
• Logical Plan 通过优化器优化为物理执行计划（Physical Plan）
• 通过代码生成技术生成 Transformations 后进一步编译为可执行的 JobGraph 提交运行

例子1 ：考虑如下表达 JOIN 操作的一段 SQL。

SELECT 
  t1.id, 1 + 2 + t1.value AS v 
FROM t1, t2 
WHERE 
  t1.id = t2.id AND 
  t2.id < 1000

Logical Planning（逻辑执行计划）

Flink SQL 引擎使用 Apache Calcite SQL Parser 将 SQL 文本解析为词法树，SQL Validator 获取 Catalog 中元数据的信息进行语法分析和验证，转化为关系代数表达式（RelNode），再由 Optimizer 将关系代数表达式转换为初始状态的逻辑执行计划。

备注：TableAPI 代码使用 TableAPI Validator 对接 Catalog 后生成逻辑执行计划。

Physical Planning on Batch（物理执行计划）

通过上述一系列操作后，得到了优化后的逻辑执行计划。逻辑执行计划描述了执行步骤和每一步需要完成的操作，但没有描述操作的具体实现方式。而物理执行计划会考虑物理实现的特性，生成每一个操作的具体实现方式。比如 Join 是使用 SortMergeJoin、HashJoin 或 BroadcastHashJoin 等。优化器在生成逻辑执行计划时会计算整棵树上每一个节点的 Cost，对于有多种实现方式的节点（比如 Join 节点），优化器会展开所有可能的 Join 方式分别计算。最终整条路径上 Cost 最小的实现方式就被选中成为 Final Physical Plan。

回顾上述的例子1 ，当它以批模式执行，同时可以拿到输入表的 Statistics 信息。在经过前述优化后，表 t2 到达 Join 节点时只有 1,000 条数据，使用 BroadcastJoin 的开销相对最低，则最终的 Physical Plan 如下图所示。

Translation & Code Generation（转换算子）

代码生成（Code Generation） 在计算机领域是一种广泛使用的技术。在 Physical Plan 到生成 Transformation Tree （转换算子树）过程中就使用了 Code Generation。

回顾例子1 ，以 表 t2 之上的 Calc 节点 t2.id < 1000 表达式为例，通过 Code Generation 后生成了描述 Transformation Operator（flink转换算子） 的一段 Java 代码，将接收到的 Row 中 id < 1000 的 Row 发送到下一个 Operator

Flink SQL 引擎会将 Physical Plan 通过 Code Generation 翻译为 Transformations，再进一步编译为可执行的 JobGraph。

流处理中的表

和Spark 和hive等组件中的表最大不同之处,flinkSQL中的表示动态表,动态指的是动态的结果输出,结果是流式,动态,持续的

• 数据源的输入是持续的
• 查询过程是持续的
• 结果输出也是持续的

动态:不仅仅是数据追加,也有对数据输出的结果的 撤回(删除),更新;

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

• 流被转换为动态表。
• 对动态表进行连续查询评估，生成一个新的动态表。
• 生成的动态表被转换回流。

动态表示例

点击事件流来解释动态表和连续查询的概念

CREATE TABLE clicks (
  user  VARCHAR,     -- the name of the user
  url   VARCHAR,     -- the URL that was accessed by the user
  cTime TIMESTAMP(3) -- the time when the URL was accessed
) WITH (...);

连续查询

连续查询在动态表上进行评估，并生成一个新的动态表作为结果。与批查询相反，连续查询永远不会终止并根据其输入表的更新更新其结果表。在任何时间点，连续查询在语义上等同于在输入表的快照上以批处理模式执行的相同查询的结果。

第一个查询是一个简单的GROUP-BY COUNT聚合查询。它clicks根据字段对user表格进行分组，并计算访问的 URL 的数量

第二个查询与第一个查询类似，但在计算 URL 数量之前clicks，除了属性之外，还在[每小时滚动窗口]user上对表进行分组（稍后讨论基于时间的计算，例如窗口是基于特殊 :时间窗口)

表到流转换

动态表可以像常规数据库表一样通过INSERT、UPDATE、DELETE和不断修改。

Flink 的 Table API 和 SQL 支持三种方式来编码动态表的变化：

• Append-only stream : 追加流可以通过发出插入的行将仅由更改修改的动态表INSERT转换为流
• Retract stream: 回撤流 撤回流是具有两种类型消息的流，添加消息和撤回消息

1.mary + , 2. bob + 3. mary 先delete在 insert 4. liz + 5.bob delete 后在insert

• Upsert stream : 两种消息的流，upsert messages和delete messages。转换为更新插入流的动态表需要一个（可能是复合的）唯一键。具有唯一键的动态表通过编码转换为流INSERT，并UPDATE更改为更新插入消息和DELETE更改为删除消息。流消费操作员需要知道唯一的键属性才能正确应用消息。与 retract 流的主要区别在于UPDATE更改是用单个消息编码的，因此效率更高

  

Append-only stream:追加流，是每个单词第一次出现的时候，都是+I

Retract stream:回撤流，每个单词再次出现的时候，会输出修改之前（-U）和修改之后(+U)的结果

Upsert stream : Insert+Update组合，第一次出现视为Insert和再次出现视为Update，组合在一起来表示如果第一次出现执行的是Insert，再次出现执行Update，两种状态组合一起，不会显示更新之前的输出，使用*表示，—D依然表示删除。

时间属性

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

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

时间属性的数据类型必须为TIMESTAMP，它的行为类似于常规时间戳，可以直接访问并且进行计算。

按照时间语义的不同，可以把时间属性的定义分成事件时间（event time）和处理时间（processing time）两种情况。都是固定写法，记住就完事了。

事件时间

事件时间属性可以在创建表DDL中定义，增加一个字段，通过WATERMARK语句来定义事件时间属性。具体定义方式如下：

CREATE TABLE EventTable(
  user STRING,
  url STRING,  
  ts TIMESTAMP(3),	--表示事件时间
   # WATERMARK FOR 时间字段 AS ts - 时间间隔(必须用单引号) 时间单位
  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND -- 通过for指定事件时间列，水印等待时间5秒钟
) WITH (
  ...
);

这里我们把ts字段定义为事件时间属性，而且基于ts设置了5秒的水位线延迟。

时间戳类型必须是 TIMESTAMP 或者TIMESTAMP_LTZ 类型。但是时间戳一般都是秒或者是毫秒（BIGINT 类型），这种情况可以通过如下方式转换

ts BIGINT,
# 精确到miao后面3位，也就是ms
time_ltz AS TO_TIMESTAMP_LTZ(ts, 3),

事件时间需要与水印结合一起使用，当然也需要跟窗口放在一起操作，事件时间列来自于数据本身携带的时间，若数据携带事件时间列不是 TIMESTAMP 或者TIMESTAMP_LTZ 类型，需要额外定义一个虚拟事件时间列，将事件时间列转换成虚拟事件时间列，后续水印执行该列。

处理时间

在定义处理时间属性时，必须要额外声明一个字段，专门用来保存当前的处理时间。

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

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

处理时间与数据携带的时间没有任何关系，是获取当前系统时间，因此直接在表中增加一个虚拟列，例如：ts as ProcTime()， 在EventTable表中增加一个虚拟处理时间列，叫做ts，处理时间跟水印没有任何关系

FlinkSQL 编程

DDL（Data Definition Language）数据定义

数据库

创建数据库

语法

CREATE DATABASE [IF NOT EXISTS] [catalog_name.]db_name
  [COMMENT database_comment]
  WITH (key1=val1, key2=val2, ...)

查询数据库

# 查询所有数据库
SHOW DATABASES;
# 查询当前数据库
SHOW CURRENT DATABASE;

修改数据库

ALTER DATABASE [catalog_name.]db_name SET (key1=val1, key2=val2, ...);

删除数据库

DROP DATABASE [IF EXISTS] [catalog_name.]db_name [ (RESTRICT | CASCADE) ]

• RESTRICT：删除非空数据库会触发异常。默认启用

• CASCADE：删除非空数据库也会删除所有相关的表和函数。

切换当前数据库

USE database_name;

表

创建表

(1)语法

CREATE TABLE [IF NOT EXISTS] [catalog_name.][db_name.]table_name
  (
    -- 正常的列 以及 元数据(比如Kafka数据携带的时间戳...)
    { <physical_column_definition> | <metadata_column_definition> | <computed_column_definition> }[ , ...n]
    -- 水印
    [ <watermark_definition> ]
    -- 表的限制，比如主键
    [ <table_constraint> ][ , ...n]
 
  )
  -- 给表添加注释
  [COMMENT table_comment]
  -- 像 hive 一样 partition by
  [PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
  -- with 里面指定这张表的一些属性和参数，比如连接器...
  WITH (key1=val1, key2=val2, ...)
 
  [ LIKE source_table [( <like_options> )] | AS select_query ]

① physical_column_definition

​ 物理列是数据库中所说的常规列。其定义了物理介质中存储的数据中字段的名称、类型和顺序。其他类型的列可以在物理列之间声明，但不会影响最终的物理列的读取。

② metadata_column_definition

​ 元数据列是 SQL 标准的扩展，允许访问数据源本身具有的一些元数据。元数据列由 METADATA 关键字标识。例如，我们可以使用元数据列从Kafka记录中读取和写入时间戳，用于基于时间的操作（这个时间戳不是数据中的某个时间戳字段，而是数据写入 Kafka 时，Kafka 引擎给这条数据打上的时间戳标记）。connector和format文档列出了每个组件可用的元数据字段。

CREATE TABLE MyTable ( 
  `user_id` BIGINT, 
  `name` STRING,
  -- 把元数据赋值给 record_time 字段
  `record_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' 
) WITH ( 
  'connector' = 'kafka' 
  ... 
);

如果自定义的列名称和 Connector 中定义 metadata 字段的名称一样， FROM xxx 子句可省略

CREATE TABLE MyTable ( 
`user_id` BIGINT, 
`name` STRING, 
`timestamp` TIMESTAMP_LTZ(3) METADATA 
) WITH ( 
'connector' = 'kafka' 
... 
);

如果自定义列的数据类型和 Connector 中定义的 metadata 字段的数据类型不一致，程序运行时会自动 cast强转，但是这要求两种数据类型是可以强转的。

CREATE TABLE MyTable ( 
`user_id` BIGINT, 
`name` STRING, 
-- 将时间戳强转为 BIGINT 
`timestamp` BIGINT METADATA 
) WITH ( 
'connector' = 'kafka' 
... 
);

默认情况下，Flink SQL planner 认为 metadata 列可以读取和写入。然而，在许多情况下，外部系统提供的只读元数据字段比可写字段多。因此，可以使用 VIRTUAL 关键字排除元数据列的持久化(表示只读)。

CREATE TABLE MyTable (
  -- 可读可写
  `timestamp` BIGINT METADATA,
  -- 只读
  `offset` BIGINT METADATA VIRTUAL, 
    `user_id` BIGINT, 
  `name` STRING, 
) WITH ( 
  'connector' = 'kafka' 
  ... 
);

③ computed_column_definition

计算列是使用语法column_name AS computed_column_expression生成的虚拟列。

计算列就是拿已有的一些列经过一些自定义的运算生成的新列，在物理上并不存储在表中，只能读不能写。列的数据类型从给定的表达式自动派生，无需手动声明。

CREATE TABLE MyTable ( 
  `user_id` BIGINT, 
  `price` DOUBLE, 
  `quantity` DOUBLE,
  -- 把 price 列和 quanitity 列的值的乘积作为一个新列
  `cost` AS price * quanitity 
) WITH ( 
  'connector' = 'kafka' 
  ... 
);

④ 定义Watermark

Flink SQL 提供了几种 WATERMARK 生产策略：

• 严格升序：WATERMARK FOR rowtime_column AS rowtime_column。

Flink 任务认为时间戳只会越来越大，也不存在相等的情况，只要相等或者小于之前的，就认为是迟到的数据。

• 递增：WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL ‘0.001’ SECOND 。

一般基本不用这种方式。如果设置此类，则允许有相同的时间戳出现。

• 有界无序： WATERMARK FOR rowtime_column AS rowtime_column – INTERVAL ‘string’ timeUnit 。

此类策略就可以用于设置最大乱序时间，假如设置为 WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL ‘5’ SECOND ，则生成的是运行 5s 延迟的Watermark。一般都用这种 Watermark 生成策略，此类 Watermark 生成策略通常用于有数据乱序的场景中，而对应到实际的场景中，数据都是会存在乱序的，所以基本都使用此类策略。

⑤ PRIMARY KEY

主键约束表明表中的一列或一组列是唯一的，并且它们不包含NULL值。主键唯一地标识表中的一行，只支持 not enforced（这是语法规则，必须加上）。

CREATE TABLE MyTable ( 
`user_id` BIGINT, 
`name` STRING, 
PARYMARY KEY(user_id) not enforced 
) WITH ( 
'connector' = 'kafka' 
... 
);

⑥ PARTITIONED BY

创建分区表

⑦ with语句

用于创建表的表属性，用于指定外部存储系统的元数据信息。配置属性时，表达式key1=val1的键和值都应该是字符串字面值。如下是Kafka的映射表：

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

一般 with 中的配置项由 Flink SQL 的 Connector（链接外部存储的连接器） 来定义，每种 Connector 提供的with 配置项都是不同的。

如kafka：https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/table/kafka/#connector-options

⑧ LIKE

用于基于现有表的定义创建表。此外，用户可以扩展原始表或排除表的某些部分。

可以使用该子句重用(可能还会覆盖)某些连接器属性，或者向外部定义的表添加水印。

CREATE TABLE Orders ( 
    `user` BIGINT, 
    product STRING, 
    order_time TIMESTAMP(3) 
) WITH ( 
    'connector' = 'kafka', 
    'scan.startup.mode' = 'earliest-offset' 
);

CREATE TABLE Orders_with_watermark (
     -- Add watermark definition 
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND 
) WITH ( 
    -- Overwrite the startup-mode 
    'scan.startup.mode' = 'latest-offset' 
) 
LIKE Orders;

⑨ AS select_statement（CTAS）

在一个create-table-as-select (CTAS)语句中，还可以通过查询的结果创建和填充表。CTAS是使用单个命令创建数据并向表中插入数据的最简单、最快速的方法。

CREATE TABLE my_ctas_table 
WITH ( 
    'connector' = 'kafka', 
    ... 
) 
AS SELECT id, name, age FROM source_table WHERE mod(id, 10) = 0;

注意:CTAS有以下限制:

• 暂不支持创建临时表。
• 目前还不支持指定显式列（create table 后面不能自己写列字段）。
• 还不支持指定显式水印（不能自己添加水印）。
• 目前还不支持创建分区表。
• 目前还不支持指定主键约束。

（2）简单建表示例

创建一个 test 表，指定连接器为 print ：

用 like 关键字创建一个结构和 test 表一样的表 test1 并在它的基础上增加一个字段 value：

使用查询结果来新建一个表：

我们可以看到，我们表 test 的查询结果只能被当做一个 Sink 来使用（也就是只能被插入），不能被当做输入源。

查看表

（1）查看所有表

SHOW TABLES [ ( FROM | IN ) [catalog_name.]database_name ] [ [NOT] LIKE <sql_like_pattern> ]

如果没有指定数据库，则从当前数据库返回表。

LIKE子句中sql pattern的语法与MySQL方言的语法相同:

• %匹配任意数量的字符，甚至零字符，%匹配一个’%'字符。
• 只匹配一个字符，_只匹配一个’'字符

（2）查看表信息

{ DESCRIBE | DESC } [catalog_name.][db_name.]table_name

修改表

（1）修改表名

ALTER TABLE [catalog_name.][db_name.]table_name RENAME TO new_table_name

（2）修改表属性

ALTER TABLE [catalog_name.][db_name.]table_name SET (key1=val1, key2=val2, ...)

删除表

DROP [TEMPORARY] TABLE [IF EXISTS] [catalog_name.][db_name.]table_name

查询

DataGen & Print

1）创建数据生成器源表

CREATE TABLE source (
     `id` INT,
     ts BIGINT, 
     vc INT 
) WITH (
    -- flink 自带的数据生成器
    'connector' = 'datagen',
    -- 每s生成的数据条数
    'rows-per-second'='1',
    -- 生成类型 sequence代表自增序列，需要指定起始值和结束值
    'fields.id.kind'='sequence',
    -- id字段自增起始值
    'fields.id.start'='1',
    -- id字段自增结束值
    'fields.id.end'='10000',
    -- ts字段的生成类型
    'fields.ts.kind'='sequence',
    'fields.ts.start'='1',
    'fields.ts.end'='1000000',
    -- vc字段类型 随机值
    'fields.vc.kind'='random',
    -- 最小值 1
    'fields.vc.min'='1',
    -- 最大值 100
    'fields.vc.max'='100' 
);
 
CREATE TABLE sink (
    id INT,
    ts BIGINT,
    vc INT
) WITH (
	'connector' = 'print'
);

2）查询源表

查询数据：

select * from source；

注意：如果发现刷新不动，就退出去查看一下log4j输出了什么警告，有的警告可以忽略，但是有的可能就是原因。比如因为没有在环境变量中添加 HADOOP_CONF_DIR ，导致数据生成器不生成数据。

可以看到结果显示模式是 table 模式，这是默认的显示模式，我们在前面的常用配置里讲过，还有一种 changelog 模式可以设置。

set sql-client.execution.result-mode=changelog;

再次查询：

我们可以看到，这种模式下，它的显示比 table 模式多了一列 op ，代表操作，+I 代表新增数据，撤回就是 -U。

此外还有一种模式叫做 tableau：

set sql-client.execution.result-mode=tableau;

可以看到，这种模式喜爱，我们不会进入那个专门的数据展示界面，更加方便。

select * from source;

3)插入sink表并查询

​ 创建 Sink表：

试着把 source 中的数据输出到 sink：

insert into sink select * from source;

可以看到它给我们返回了 一个 Job Id，我们可以直接查询 sink 表，或者也可以在 Web UI 中查看：

select * from sink;

With子句

WITH提供了一种编写辅助语句的方法，以便在较大的查询中使用。这些语句通常被称为公共表表达式(Common Table Expression, CTE)，可以认为它们定义了仅为一个查询而存在的临时视图。

1）语法

WITH <with_item_definition> [ , ... ] 
SELECT ... FROM ...;  
<with_item_defintion>: 
    with_item_name (column_name[, ...n]) AS ( <select_query> )

2）案例

查询这个临时表就相当于执行了 with 内部的查询，比如下面：

WITH source_with_total AS (
    SELECT id, vc+10 AS total
    FROM source
)
-- 注意这里没有分号 这两个句子是一个作业里面的
SELECT id, SUM(total) FROM source_with_total GROUP BY id;

查询 source_with_total 就相当于查询了它内部的语句：select id,vc+10 as total from source;当然，我们在查这张临时表的时候可以选择字段。

需要注意的地方就是我们生成临时表的句子和查询临时表的句子是一个语句没有分号的，它们同属于一个作业，这个临时表只在这里生效，就像帮我们的查询语句简化了一下，作业结束它也就不存在了。我们完全可以写成这样：

select id,vc+10 as total from source;

SELECT & WHERE 子句

1）语法

SELECT select_list FROM table_expression [ WHERE boolean_expression ]

2）案例

-- 自定义 Source 的数据
-- 不需要给表 t 的字段显示添加类型(添加会报错) flink会自动识别
SELECT order_id, price FROM (VALUES (1, 2.0), (2, 3.1)) AS t (order_id, price);
 
SELECT vc + 10 FROM source WHERE id >10;

通过查询结果，我们可以知道id=10的这条数据它的 vc 是<=10 的。

SELECT DISTINCT 子句

用作根据 key 进行数据去重

SELECT DISTINCT vc FROM source;

对于流查询，计算查询结果所需的状态可能无限增长。状态大小取决于不同行数。可以设置适当的状态生存时间(TTL)的查询配置，以防止状态过大。但是，这可能会影响查询结果的正确性。如某个 key 的数据过期从状态中删除了，那么下次再来这么一个 key，由于在状态中找不到，就又会输出一遍。

分组聚合

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

select COUNT(*) from source;

之前说过，动态表转为流，对于持续查询来说是一种更新查询，这里很明显是追加流和撤回流，而不是更新插入流。

而更多的情况下，我们可以通过GROUP BY子句来指定分组的键（key），从而对数据按照某个字段做一个分组统计。

SELECT vc, COUNT(*) as cnt FROM source GROUP BY vc;

这种聚合方式，就叫作“分组聚合”（group aggregation）。想要将结果表转换成流或输出到外部系统，必须采用撤回流（retract stream）或更新插入流（upsert stream）的编码方式；如果在代码中直接转换成DataStream打印输出，需要调用toChangelogStream()。

分组聚合既是SQL原生的聚合查询，也是流处理中的聚合操作，这是实际应用中最常见的聚合方式。当然，使用的聚合函数一般都是系统内置的，如果希望实现特殊需求也可以进行自定义。

1）group聚合案例

CREATE TABLE source1 (
dim STRING,
user_id BIGINT,
price BIGINT,
row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
-- 指定了水位线为 row_time 字段 - 5s
WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'datagen',
'rows-per-second' = '10',
'fields.dim.length' = '1',
'fields.user_id.min' = '1',
'fields.user_id.max' = '100000',
'fields.price.min' = '1',
'fields.price.max' = '100000'
);
 
CREATE TABLE sink1 (
dim STRING,
pv BIGINT,
sum_price BIGINT,
max_price BIGINT,
min_price BIGINT,
uv BIGINT,
window_start bigint
) WITH (
'connector' = 'print'
);
 
insert into sink1
select dim,
count(*) as pv,
sum(price) as sum_price,
max(price) as max_price,
min(price) as min_price,
-- 计算 uv 数
count(distinct user_id) as uv,
cast((UNIX_TIMESTAMP(CAST(row_time AS STRING))) / 60 as bigint) as window_start
from source1
group by
dim,
-- UNIX_TIMESTAMP得到秒的时间戳，将秒级别时间戳 / 60 转化为 1min， 
cast((UNIX_TIMESTAMP(CAST(row_time AS STRING))) / 60 as bigint);

这里可以看到，我们在自动生成数据的时候，并没有指定字段的生成类型（比如是自增序列还是随机数或者字符串） ，因为只要我们指定了 max 和 min 那么这就是一个随机数；如果我们指定了 start 和 end，那么就代表这是自增序列；如果指定了 length ，就代表这是一个字符串。

也可以不用 insert into 到 sink，而是直接查询，效果是一样的

Flink SQL> select dim,
 count(*) as pv,
 sum(price) as sum_price,
 max(price) as max_price,
 min(price) as min_price,
 -- 计算 uv 数
 count(distinct user_id) as uv,
 cast((UNIX_TIMESTAMP(CAST(row_time AS STRING))) / 60 as bigint) as window_start
 from source1
 group by
 dim,
 -- UNIX_TIMESTAMP得到秒的时间戳，将秒级别时间戳 / 60 转化为 1min,
 cast((UNIX_TIMESTAMP(CAST(row_time AS STRING))) / 60 as bigint);

2）多维分析

多维分析，举个例子比如我们要统计关于学生成绩的信息（最高分、最低分、平均分），我们可以从不同维度（年级、学科、性别）去统计，比如每个年级的最高分、最低分、平均分；或者不同性别的最高分… 不同年级不同学科的最高分… 或者不同年级、不同学科、不同性别的最高分…。

Group 聚合也支持 Grouping sets 、Rollup 、Cube，如下案例是Grouping sets：

SELECT
  supplier_id
, rating
, product_id
, COUNT(*)
FROM (
VALUES
  ('supplier1', 'product1', 4),
  ('supplier1', 'product2', 3),
  ('supplier2', 'product3', 3),
  ('supplier2', 'product4', 4)
)
-- 供应商id、产品id、评级
AS Products(supplier_id, product_id, rating)  
GROUP BY GROUPING SETS(
  (supplier_id, product_id, rating),
  (supplier_id, product_id),
  (supplier_id, rating),
  (supplier_id),
  (product_id, rating),
  (product_id),
  (rating),
  ()
);

这段 Flink SQL 代码的主要目的是对一组产品数据进行分组聚合。

• VALUES 语句:

VALUES
  ('supplier1', 'product1', 4),
  ('supplier1', 'product2', 3),
  ('supplier2', 'product3', 3),
  ('supplier2', 'product4', 4)
)

这部分定义了一个包含四行数据的虚拟表。每一行代表一个产品的供应商ID、产品ID和评级。

• AS Products(supplier_id, product_id, rating):

AS Products(supplier_id, product_id, rating)

这部分将虚拟表重命名为 “Products”，并为每一列定义了别名：supplier_id、product_id 和 rating。

• GROUP BY GROUPING SETS:

GROUPING SETS 是 SQL 中的一种功能，它允许你指定多个分组条件，并为每个分组条件返回一个结果。这在探索多个维度聚合时非常有用。

在这个例子中，我们可以看到以下分组条件：

• supplier_id、product_id、rating
• supplier_id、product_id
• supplier_id、rating
• supplier_id
• product_id、rating
• product_id
• rating
• ()（空分组）

​ 这意味着，对于每个供应商ID、产品ID和评级的组合，都会进行计数。这实际上是计算每个供应商的每个产品以及每个产品的总评级的计数。同时，也计算了每个供应商的总评级、每个产品的总评级以及所有产品的总评级。最后，还计算了所有记录的总数（这是通过空分组实现的）。

SELECT 语句:

这个部分选择了上述 GROUPING SETS 中的所有列，并添加了一个 COUNT(*) 函数来计算每个分组的记录数。

所以，这段代码的输出将为给定的数据集提供以下聚合信息：

• 每个供应商的每个产品的数量以及评级；
• 每个供应商的每个产品的数量；
• 每个供应商的评级数量；
• 每个产品的评级数量；
• 每个供应商的数量；
• 每个产品的数量；
• 评级的数量；
• 所有记录的数量。

Flink的窗口

分组窗口聚合

分组窗口其实就是我们之前学过的 滑动窗口、会话窗口、滚动窗口，之所以叫它分组窗口，其实是把它的一个窗口看做一个分组。

从1.13版本开始，分组窗口聚合已经标记为过时，鼓励使用更强大、更有效的窗口TVF聚合，在这里简单做个介绍。

直接把窗口自身作为分组key放在GROUP BY之后的，所以也叫“分组窗口聚合”。[SQL查询]分组窗口是通过 GROUP BY 子句定义的。类似于使用常规 GROUP BY 语句的查询，窗口分组语句的 GROUP BY 子句中带有一个窗口函数为每个分组计算出一个结果。

Flink SQL中只支持基于时间的窗口，不支持基于元素个数的窗口。

分组窗口函数	描述
TUMBLE(time_attr, interval)	定义一个滚动窗口。time_attr 是时间属性，也就是你选择的作为时间语义的字段，interval 是窗口长度。
HOP(time_attr, interval, interval)	定义一个跳跃的时间窗口（在 Table API 中称为滑动窗口）。滑动窗口的长度（ 第二个 interval 参数 ）以及一个滑动的步长（第一个 interval 参数 ）。
SESSION(time_attr, interval)	定义一个会话时间窗口。interval 是会话的间隔。

其中，ROWTIME 代表的是事件时间语义，PROCTIME 是处理时间语义。

1）准备数据

CREATE TABLE ws (
  id INT,
  vc INT,
  pt AS PROCTIME(), --处理时间
  et AS cast(CURRENT_TIMESTAMP as timestamp(3)), --事件时间
  WATERMARK FOR et AS et - INTERVAL '5' SECOND   --watermark
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10',
  'fields.id.min' = '1',
  'fields.id.max' = '3',
  'fields.vc.min' = '1',
  'fields.vc.max' = '100'
);
# 设置显示
set sql-client.execution.result-mode=tableau;

可以看到，一张 flink sql 的表是允许事件时间和处理时间都存在的，只是水位线只能指定一个。

滚动窗口示例（时间属性，窗口长度）

Flink SQL> select id,
 sum(vc) as vcSum,
 tumble_start(et,interval '5' second) as window_start,
 tumble_end(et,interval '5' second) as window_end
 from ws
 group by id,tumble(et,interval '5' second);

查询结果：

可以看到每个 key 的窗口每 5s 滚动一次。

滑动窗口（时间属性，滑动步长，窗口长度）

Flink SQL> select id,
 hop_start(et,interval '3' second,interval '5' second) window_start,
 hop_end(et,interval '3' second,interval '5' second) window_end,
 sum(vc) sum_vc
 from ws
 group by id,hop(et,interval '3' second,interval '5' second);

指定滑动步长为 3 窗口大小为 5，查看运行结果：

可以看到，相同 key 的窗口确实间隔 3s，窗口大小为 5s。

会话窗口（时间属性，会话间隔）

select  
id,
SESSION_START(et, INTERVAL '5' SECOND)  wstart,
SESSION_END(et, INTERVAL '5' SECOND)  wend,
sum(vc) sumVc
from ws
group by id, SESSION(et, INTERVAL '5' SECOND);

这里因为我们的数据生成器是源源不断生成数据的，而我们指定了会话间隔为 5s，也就是说只有连续 5s 收不到数据窗口才会关闭，所以我们是看不到数据结果的。

窗口表值函数（TVF）聚合

前面我们学习的 Group Window 在 Flink1.13 版本之后已经被标记为过时了，更推荐的是 TVF Window。

对比GroupWindow，TVF窗口更有效和强大。包括：

• 提供更多的性能优化手段
• 支持GroupingSets语法
• 可以在window聚合中使用TopN
• 提供累积窗口

​ 对于窗口表值函数（TVF），窗口本身返回的是就是一个表，所以窗口会出现在FROM后面，GROUP BY后面的则是窗口新增的字段 window_start 和 window_end。

FROM TABLE(
窗口类型(TABLE 表名, DESCRIPTOR(时间字段),INTERVAL 时间间隔)
)
GROUP BY [window_start,][window_end,] --可选

滚动窗口（tumble）

TUMBLE 函数通过时间属性字段为每行数据分配一个窗口。 在流计算模式，时间属性字段必须被指定为 事件或处理时间属性。 在批计算模式，窗口表函数的时间属性字段必须是 TIMESTAMP 或 TIMESTAMP_LTZ 的类型。 TUMBLE 的返回值包括原始表的所有列和附加的三个用于指定窗口的列，分别是：“window_start”，“window_end”，“window_time”。函数运行后，原有的时间属性 “timecol” 将转换为一个常规的 timestamp 列。

TUMBLE 函数有三个必传参数，一个可选参数：

TUMBLE(TABLE data, DESCRIPTOR(timecol), size [, offset ])

• data ：拥有时间属性列的表。
• timecol ：列描述符，决定数据的哪个时间属性列应该映射到窗口。
• size ：窗口的大小（时长）。
• offset ：窗口的偏移量 [非必填]。

下面是调用示例：

Flink SQL> select id,
 sum(vc) as vc_sum,
 window_start,
 window_end
 from table(    -- 这里的table是关键字 不是表名
 tumble(table ws,descriptor(et),interval '5' second)
 )
 group by id,window_start,window_end; -- 这里的window_start,window_end都是关键字

运行结果：

滑动窗口（hop）

HOP 函数通过时间属性字段为每一行数据分配了一个窗口。 在流计算模式，这个时间属性字段必须被指定为 事件或处理时间属性。 在批计算模式，这个窗口表函数的时间属性字段必须是 TIMESTAMP 或 TIMESTAMP_LTZ 的类型。 HOP 的返回值包括原始表的所有列和附加的三个用于指定窗口的列，分别是：“window_start”，“window_end”，“window_time”。函数运行后，原有的时间属性 “timecol” 将转换为一个常规的 timestamp 列。

HOP 有四个必填参数和一个可选参数：

HOP(TABLE data, DESCRIPTOR(timecol), slide, size [, offset ])

• data：拥有时间属性列的表。
• timecol：列描述符，决定数据的哪个时间属性列应该映射到窗口。
• slide：窗口的滑动步长。
• size：窗口的大小(时长)。
• offset：窗口的偏移量 [非必填]。

注意：在 TVF 中，滑动窗口的大小必须是步长的整数倍，因为 TVF 会对滑动窗口进行一个优化：把滑动窗口按照步长大小划分为 （窗口大小/步长）个滚动窗口，这样一些需要重复计算的数据所在的滚动窗口只需要计算一次即可。

如果不是整数倍，会报错：[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.TableException: HOP table function based aggregate requires size must be an integral multiple of slide, but got size xxx ms and slide xxx ms

Flink SQL> select
 id,
 sum(vc) as vc_sum,
 window_start,
 window_end
 from table(
 hop(table ws,descriptor(et),interval '2' second,interval '10' second)
 )
 group by id,window_start,window_end;

累积窗口（cumulate）

累积窗口在某些场景中非常有用，比如说提前触发的滚动窗口。例如：每日仪表盘从 00:00 开始每分钟绘制累积 UV，10:00 时 UV 就是从 00:00 到 10:00 的UV 总数。累积窗口可以简单且有效地实现它。

CUMULATE 函数指定元素到多个窗口，从初始的窗口开始，直到达到最大的窗口大小的窗口，所有的窗口都包含其区间内的元素，另外，窗口的开始时间是固定的。 你可以将 CUMULATE 函数视为首先应用具有最大窗口大小的 TUMBLE 窗口，然后将每个滚动窗口拆分为具有相同窗口开始但窗口结束步长不同的几个窗口。 所以累积窗口会产生重叠并且没有固定大小。

例如：1小时步长，24小时大小的累计窗口，每天可以获得如下这些窗口：[00:00, 01:00)，[00:00, 02:00)，[00:00, 03:00)， …， [00:00, 24:00)

CUMULATE　函数通过时间属性字段为每一行数据分配了一个窗口。 在流计算模式，这个时间属性字段必须被指定为 事件或处理时间属性。 在批计算模式，这个窗口表函数的时间属性字段必须是 TIMESTAMP 或 TIMESTAMP_LTZ 的类型。 CUMULATE 的返回值包括原始表的所有列和附加的三个用于指定窗口的列，分别是：“window_start”，“window_end”，“window_time”。函数运行后，原有的时间属性 “timecol” 将转换为一个常规的 timestamp 列。

CUMULATE 有四个必填参数和一个可选参数：

CUMULATE(TABLE data, DESCRIPTOR(timecol), step, size, [offset])

• data：拥有时间属性列的表。
• timecol：列描述符，决定数据的哪个时间属性列应该映射到窗口。
• step：指定连续的累积窗口之间增加的窗口大小。
• size：指定累积窗口的最大宽度的窗口时间。size必须是step的整数倍。
• offset：窗口的偏移量 [非必填]。

注意：累积窗口的窗口大小必须是步长的整数倍！

Flink SQL> select 
 id,
 sum(vc) vc_sum,
 window_start,
 window_end
 from table(
 cumulate(table ws,descriptor(et),interval '2' second,interval '10' second
 ))
 group by id,window_start,window_end;

累积窗口也很常用，需要指定两个参数（窗口的最大长度，窗口的步长），比如统计一天内每个小时的网站访问量就可以开一个累积窗口（窗口的最大长度是 24h，窗口的步长是 1h），语法也很简单，只需要修改一个滑动窗口的一个关键字：cumulate。当窗口结束后会再开启一个窗口，属性和上一个窗口是一样的。

累积窗口的底层还是一个滚动窗口，当我们定义了一个累积窗口，就相当于开了一个最大的滚动窗口，之后会根据用户指定的步长（也就是触发计算的时间）将这个窗口划分为多个窗口，这些窗口具有相同的起点和不同的终点。

会话窗口（session）

• 会话窗口 (目前仅支持流模式)

1. 会话窗口函数目前不支持批模式。
2. 会话窗口函数目前不支持 性能调优 中的任何优化。
3. 会话窗口 Join 、会话窗口 Top-N 、会话窗口聚合功能目前理论可用，但仍处于实验阶段。遇到问题可以在 JIRA 中报告。

SESSION　函数通过时间属性字段为每一行数据分配了一个窗口。 在流计算模式，这个时间属性字段必须被指定为 事件或处理时间属性。 SESSION 的返回值包括原始表的所有列和附加的三个用于指定窗口的列，分别是：“window_start”，“window_end”，“window_time”。函数运行后，原有的时间属性 “timecol” 将转换为一个常规的 timestamp 列。

SESSION 有三个必填参数和一个可选参数：

SESSION(TABLE data [PARTITION BY(keycols, ...)], DESCRIPTOR(timecol), gap)

• data：拥有时间属性列的表。
• keycols：列描述符，决定会话窗口应该使用哪些列来分区数据。
• timecol：列描述符，决定数据的哪个时间属性列应该映射到窗口。
• gap：两个事件被认为属于同一个会话窗口的最大时间间隔。

Flink SQL> SELECT id, window_start, window_end, sum(vc) vc_sum
   FROM TABLE(
       SESSION(TABLE ws PARTITION BY id, DESCRIPTOR(et), INTERVAL '5' second))
   GROUP BY id, window_start, window_end;

数据生成器是源源不断生成数据的，而我们指定了会话间隔为 5s，也就是说只有连续 5s 收不到数据窗口才会关闭，所以我们是看不到数据结果的。

窗口偏移

Offset 可选参数，可以用来改变窗口的分配。可以是正或者负的区间。默认情况下窗口的偏移是 0。不同的偏移值可以决定记录分配的窗口。 例如：在 10 分钟大小的滚动窗口下，时间戳为 2021-06-30 00:00:04 的数据会被分配到那个窗口呢？

• 当 offset 为 -16 MINUTE，数据会分配到窗口 [2021-06-29 23:54:00, 2021-06-30 00:04:00)。
• 当 offset 为 -6 MINUTE，数据会分配到窗口 [2021-06-29 23:54:00, 2021-06-30 00:04:00)。
• 当 offset 为 -4 MINUTE，数据会分配到窗口 [2021-06-29 23:56:00, 2021-06-30 00:06:00)。
• 当 offset 为 0，数据会分配到窗口 [2021-06-30 00:00:00, 2021-06-30 00:10:00)。
• 当 offset 为 4 MINUTE，数据会分配到窗口 [2021-06-29 23:54:00, 2021-06-30 00:04:00)。
• 当 offset 为 6 MINUTE，数据会分配到窗口 [2021-06-29 23:56:00, 2021-06-30 00:06:00)。
• 当 offset 为 16 MINUTE，数据会分配到窗口 [2021-06-29 23:56:00, 2021-06-30 00:06:00)。 我们可以发现，有些不同的窗口偏移参数对窗口分配的影响是一样的。在上面的例子中，-16 MINUTE，-6 MINUTE 和 4 MINUTE 对 10 分钟大小的滚动窗口效果相同。

注意：窗口偏移只影响窗口的分配，并不会影响 Watermark

注意：为了更好地理解窗口行为，这里把 timestamp 值得后面的 0 去掉了。例如：在 Flink SQL Client 中，如果类型是 TIMESTAMP(3)，2020-04-15 08:05 应该显示成 2020-04-15 08:05:00.000

Over 窗口

回想我们之前 Hive 学过的开窗函数，对每个数据，我们会去计算它的开窗范围进行计算输出。比如对上一行到这一行的数据进行计算；再比如计算第一行数据到当前行。

OVER 聚合为一系列有序行的每个输入行计算一个聚合值。与 GROUP BY 聚合相比，OVER聚合不会将每个组的结果行数减少为一行。相反，OVER聚合为每个输入行生成一个聚合值。

可以为事件时间或处理时间，以及指定为时间间隔、或行计数的范围内，定义Over windows。

语法

SELECT
  -- 聚合函数
  agg_func(agg_col) OVER (
    [PARTITION BY col1[, col2, ...]]
    ORDER BY time_col
    range_definition),
  ...
FROM ...

• ORDER BY：必须是时间戳列（事件时间、处理时间），只能升序
• PARTITION BY：标识了聚合窗口的聚合粒度
• range_definition：这个标识聚合窗口的聚合数据范围，在 Flink 中有两种指定数据范围的方式。第一种为按照行数聚合（rows 关键字），第二种为按照时间区间聚合（range 关键字）。
• 按照时间分区时，对于同一时刻的数据都会被聚合计算，但是按照行数聚合就不会。

案例

1. 按照时间区间聚合

   统计每个传感器前 10 秒到现在收到的水位数据条数

   select 
       id,
       et,
       vc,
       count(vc) over(partition by id order by et range between interval '10' second preceding and current row) cnt
   from ws;
   

   运行结果：

   

   也可以用 WINDOW 子句来在SELECT外部单独定义一个OVER窗口,可以多次使用：

   select 
       id,
       et,
       vc,
       count(vc) over w as cnt,    -- 这里的 as 可以省略
       sum(vc) over w as sum_vc
   from ws
   window w as(
       partition by id 
       order by et 
        range between interval '10' second preceding and current row
   );
   

   当我们有多个聚合函数并且这几个窗口函数的开窗属性是相同的时候这样写可以简化代码。

2. 按照行数聚合

   统计每个传感器前5条到现在数据的平均水位：

   select 
   	id,
   	et,
   	vc,
   	avg(vc) over w as avg_vc
   from ws
   window w as (
       partition by id
       order by et
        rows between 5 preceding and current row
   );
   

   运行结果：

   

特殊语法 - TopN

目前在Flink SQL中没有能够直接调用的TOP-N函数，而是提供了稍微复杂些的变通实现方法，是固定写法，特殊支持的over用法。

语法

SELECT [column_list]
FROM (
SELECT [column_list],
ROW_NUMBER() OVER ([PARTITION BY col1[, col2...]]
ORDER BY col1 [asc|desc][, col2 [asc|desc]...]) AS rownum
-- 不能指定行范围！！！ 只能从最早到当前
FROM table_name)
WHERE rownum <= N [AND conditions]

• ROW_NUMBER() ：标识 TopN 排序子句
• PARTITION BY col1[, col2…] ：标识分区字段，代表按照这个 col 字段作为分区粒度对数据进行排序取 topN，比如下述案例中的 partition by key ，就是根据需求中的搜索关键词（key）做为分区
• ORDER BY col1 [asc|desc][, col2 [asc|desc]…] ：标识 TopN 的排序规则，是按照哪些字段、顺序或逆序进行排序，可以不是时间字段，也可以降序（TopN特殊支持）
• WHERE rownum <= N ：这个子句是一定需要的，只有加上了这个子句，Flink 才能将其识别为一个TopN 的查询，其中 N 代表 TopN 的条目数
• [AND conditions] ：其他的限制条件也可以加上

案例

取每个传感器最高的3个水位值

select 
id,
et,
vc,
rownum
from(
    select id,
    et,
    vc,
    row_number() over (partition by id order by vc desc ) as rownum
    from ws
)where rownum <= 3;

运行结果：

特殊语法 - Deduplication 去重

这种语法也是借助于 over 窗口，语法和 topN 相似，唯一不同的是它要求排序的字段必须是时间属性列，同样可以降序，不可以是其它非时间属性的列。去重的原理是按时间升序或降序然后取第一条，所以升序和降序很影响性能，我们一般建议升序，因为这样当来了一条相同的数据时，由于按照时间升序，新数据的时间小，不会取代已有的数据；但是如果是降序的话，新数据的时间比较晚会被放到第一条数据的位置，就要发生更新操作；所以我们一般使用升序。

在 row_number = 1 时，如果排序字段是普通列 planner 会翻译成 TopN 算子，如果是时间属性列 planner 会翻译成 Deduplication，这两者最终的执行算子是不一样的，Deduplication 相比 TopN 算子专门做了对应的优化，性能会有很大提升。可以从web UI 看出是翻译成哪种算子（TopN 的算子叫 Rank、Deduplication 的算子叫 Deduplication、普通聚合（比如sum、avg）的算子叫 OverAggregate）。

如果是按照时间属性字段降序，表示取最新一条，会造成不断的更新保存最新的一条。如果是升序，表示取最早的一条，不用去更新，性能更好。

语法

SELECT [column_list]
FROM (
SELECT [column_list],
ROW_NUMBER() OVER ([PARTITION BY col1[, col2...]]
ORDER BY time_attr [asc|desc]) AS rownum
FROM table_name)
WHERE rownum = 1;

案例

对每个传感器的水位值去重

select
id,
et,
vc,
rownum
from (
    select 
    id,
    et,
    vc,
    row_number() over(partition by id,vc order by et asc) as rownum
    from ws 
)where rownum = 1;

运行结果：

所有的结果的操作都是 +I 说明没有重复数据，都是新插入的数据。

联结查询

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

在流处理中，动态表的Join对应着两条数据流的Join操作。Flink SQL中的联结查询大体上也可以分为两类：

• SQL原生的联结查询方式

• 流处理中特有的联结查询。

注意：什么 join 就以谁为主，left join 就是以左边的流为主，右边的流来了只能老老实实待着，待在状态里没有资格输出。right join 同理。

常规联结查询

常规联结（Regular Join）是SQL中原生定义的Join方式，是最通用的一类联结操作。它的具体语法与标准SQL的联结完全相同，通过关键字JOIN来联结两个表，后面用关键字ON来指明联结条件。

与标准SQL一致，Flink SQL的常规联结也可以分为内联结（INNER JOIN）和外联结（OUTER JOIN），区别在于结果中是否包含不符合联结条件的行。

Regular Join 包含以下几种（以 L 作为左流中的数据标识， R 作为右流中的数据标识）：

• Inner Join（Inner Equal Join）：流任务中，只有两条流 Join 到才输出，输出 +[L, R]
• Left Join（Outer Equal Join）：流任务中，左流数据到达之后，无论有没有 Join 到右流的数据，都会输出（Join 到输出 +[L, R] ，没 Join 到输出 +[L, null] ），如果右流之后数据到达之后，发现左流之前输出过没有 Join 到的数据，则会发起回撤流，先输出 -[L, null] ，然后输出 +[L, R]
• Right Join（Outer Equal Join）：有 Left Join 一样，左表和右表的执行逻辑完全相反
• Full Join（Outer Equal Join）：流任务中，左流或者右流的数据到达之后，无论有没有 Join 到另外一条流的数据，都会输出（对右流来说：Join 到输出 +[L, R] ，没 Join 到输出 +[null, R] ；对左流来说：Join 到输出 +[L, R] ，没 Join 到输出 +[L, null] ）。如果一条流的数据到达之后，发现之前另一条流之前输出过没有 Join 到的数据，则会发起回撤流（左流数据到达为例：回撤 -[null, R] ，输出+[L, R] ，右流数据到达为例：回撤 -[L, null] ，输出 +[L, R]

Regular Join 的注意事项：

• 实时 Regular Join 可以不是 等值 join 。等值 join 和 非等值 join 区别在于， 等值 join数据 shuffle 策略是 Hash，会按照 Join on 中的等值条件作为 id 发往对应的下游； 非等值 join 数据 shuffle 策略是 Global，所有数据发往一个并发，按照非等值条件进行关联。
• 流的上游是无限的数据，所以要做到关联的话，Flink 会将两条流的所有数据都存储在 State 中，所以 Flink 任务的 State 会无限增大，因此你需要为 State 配置合适的 TTL，以防止 State 过大。
• join 的时候，左右两条流中的数据都会先存到状态中去，先来的数据会去对方的状态中去查找有没有可以匹配上的，没有就匹配为 null。

我们再准备一张表用于join：

CREATE TABLE ws1 (
  id INT,
  vc INT,
  pt AS PROCTIME(), --处理时间
  et AS cast(CURRENT_TIMESTAMP as timestamp(3)), --事件时间
  WATERMARK FOR et AS et - INTERVAL '0.001' SECOND   --watermark
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.id.min' = '3',
  'fields.id.max' = '5',
  'fields.vc.min' = '1',
  'fields.vc.max' = '100'
);

(INNER Equi-JOIN）

内联结用INNER JOIN来定义，会返回两表中符合联接条件的所有行的组合，也就是所谓的笛卡尔积（Cartesian product）。目前仅支持等值联结条件。

select ws.id,ws1.id from ws join ws1 on ws.id=ws1.id;

可以看到，inner join 必须等待两条流都到齐后才会开始 join，否则就在状态里等着。

等值外联结（OUTER Equi-JOIN）

与内联结类似，外联结也会返回符合联结条件的所有行的笛卡尔积；另外，还可以将某一侧表中找不到任何匹配的行也单独返回。Flink SQL支持左外（LEFT JOIN）、右外（RIGHT JOIN）和全外（FULL OUTER JOIN），分别表示会将左侧表、右侧表以及双侧表中没有任何匹配的行返回。

具体用法如下：

left join：

select ws.id,ws1.id,ws.vc,ws1.vc from ws left join ws1 on ws.vc=ws1.vc;

运行结果：

right join：

select ws.id,ws.vc,ws1.id,ws1.vc from ws right join ws1 on ws.vc=ws1.vc;

运行结果：

full join：

间隔联结查询

我们曾经学习过 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

测试一下：

select * from ws,ws1 where ws.id=ws1.id and ws.et between ws1.et - interval '2' second and ws1.et + interval '2' second;

查看 Web UI ：

维表联结查询

Lookup Join 其实就是维表 Join，实时获取外部缓存的 Join，Lookup 的意思就是实时查找。

**维度表（Dimension Table）**是数据仓库中的一种表结构，通常包含有关业务过程的描述性属性。它可以提供多个角度来理解和分析数据。维度表的特点是宽而扁平，每一行代表一个实体，每一列代表一个属性。

比如我们只知道商品的 id ，不知道商品的名字，那么我们的Flink流就需要去 join 通过mysql 的连接器去获取mysql中的数据。

上面说的这几种 Join 都是流与流之间的 Join，而 Lookup Join 是流与 Redis，Mysql，HBase 这种外部存储介质的 Join。仅支持处理时间字段。

表A 
-- 固定写法 没有为什么
JOIN 维度表名 FOR SYSTEM_TIME AS OF 表A.proc_time AS 别名
ON xx.字段=别名.字段

比如维表在mysql，维表join的写法如下:

#1.在MySQL中创建库，创建表，往表中添加数据
--创建库
create database test;
--创建表
CREATE TABLE `user_profile` (
  `user_id` varchar(100) NOT NULL,
  `age` varchar(100) DEFAULT NULL,
  `sex` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
--添加数据
INSERT INTO test.user_profile (user_id,age,sex) VALUES
	 ('a','12-18','男'),
	 ('b','18-24','女'),
	 ('c','18-24','男');

#2.在FlinkSQL中，创建业务表
--创建表
CREATE TABLE click_log_table (
  log_id BIGINT, 
  `timestamp` bigint,
  user_id string,
  proctime AS PROCTIME()
)
WITH (
  'connector' = 'socket',
  'hostname' = 'node1',        
  'port' = '9999',
  'format' = 'csv'
);

#3.在FlinkSQL中，创建MySQL维度表的映射表（目的就是为了读取MySQL的数据）
CREATE TABLE user_profile (
  `user_id` string, 
  `age` string,
  `sex` string
)
WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://node1:3306/test',
   'table-name' = 'user_profile',
   'username'='root',
   'password'='123456'
);

#4.业务关联查询操作（join操作）
--业务SQL
SELECT 
    s.log_id as log_id
    , s.`timestamp` as `timestamp`
    , s.user_id as user_id
    , s.proctime as proctime
    , u.sex as sex
    , u.age as age
FROM click_log_table AS s
LEFT JOIN user_profile FOR SYSTEM_TIME AS OF s.proctime AS u
ON s.user_id = u.user_id;

需要将：flink-connector-jdbc-3.2.0-1.19.jar、mysql-connector-java-8.0.17.jar两个jar文件放到flink安装根目录的lib目录下，并重启flink集群和sqlGateway以及sqlclient。

维表join是单流驱动的，默认情况下，左流的数据到达会主动去维表中查询数据对应的信息，如果查询到则返回查询的结果，如果查询不到则返回Null，如果维表数据增加，不会主动的与左流数据进行主动匹配，而是被动匹配的

缺点：左流的数据到达都会去维表主动查询数据，当数据量非常大的情况下，会给维表带来非常大的访问压力

如何解决：可以将维表数据一旦查询过匹配的数据，就将该结果缓存起来，下一次再次有该key的数据查询的时候可以直接从缓存中获取，但是有带来新的问题，mysql的数据更新了呢？缓存的数据还是之前脏数据，又该如何解决？设置缓存过期的时间，尽可能在业务方面将缓存自动失效时间设置为远远小于业务数据更新频率时间即可！

https://nightlies.apache.org/flink/flink-docs-release-1.19/zh/docs/connectors/table/jdbc/#%E8%BF%9E%E6%8E%A5%E5%99%A8%E5%8F%82%E6%95%B0

Order by 和 limit

order by

支持 Batch\Streaming，但在实时任务中一般用的非常少。

实时任务中，Order By 子句中必须要有时间属性字段，并且必须写在最前面且为升序。

-- 时间属性必须为升序asc 
select * from ws order by et,id desc;    - 我们指定id字段降序

运行结果：

limit

select * form ws limit 3;

运行结果：

SQL Hints

翻译过来叫 SQL 暗示，确实语法就像我们 Java 中的注释一样。

在执行查询时，可以在表名后面添加SQL Hints来临时修改表属性，对当前job生效。放我们定义好一张 Flink SQL 表的时候，如果要修改参数，比如数据生成器产生数据的速度，我们不可能每次都把表删了重建，那样太麻烦了。我们可以使用 SQL Hints，它主要修改的就是我们建表时 with的参数。

-- 修改数据生成器产生数据的速度
select * from ws1/*+ OPTIONS('rows-per-second'='10')*/;

使用场景之一：

流批一体的时候使用居多，在离线业务中往往需要每天调度一次，需要添加条件执行日期为昨天，在实时计算中则不需要添加条件，为了复用这个sql语句可以使用Hint语法动态的定义为离线运行还是实时运行。

集合操作

UNION和 UNION ALL（合并）

将两张表上下拼接在一起，要求必须字段必须一样，至少字段数量一样，而且每个位置的字段类型是一样的。

• UNION：将集合合并并且去重
• UNION ALL：将集合合并，不做去重。

select id from ws
union
select id from ws1;

运行结果：

可以看到结果是去重的，因为我们 id 的范围（ws: 1-3,ws1:3-5）。

select id from ws
union all
select id from ws1;

运行结果：

可以看到，union all 不去重。

Intersect和Intersect All

• Intersect：交集并且去重
• Intersect ALL：交集不做去重

select id from ws
intersect
select id from ws1;

运行结果：

可以看到，运行结果就是我们两张表的共同 id

select id from ws
intersect all
select id from ws1;

运行结果：

可以看到，结果没有去重。

Except和Except All

select id from ws
except
select id from ws1;

运行结果：

可以看到，左边ws流中 id=3 的数据到了，但是右边ws1流中没有，于是认为是差值输出，直到ws1中出现了id=3的数据，发现这不是差值，于是撤回。

而且结果是相对于左边 ws 流的，也就是只会输出 ws 流中有而 ws1 没有的数据。

select id from ws
except all
select id from ws1;

运行结果：

上述 SQL 在流式任务中，如果一条左流数据先来了，没有从右流集合数据中找到对应的数据时会直接输出，当右流对应数据后续来了之后，会下发回撤流将之前的数据給撤回。这也是一个回撤流。

In 子查询

• In 子查询的结果集只能有一列

select id,vc from ws where id in (select id from ws1);

运行结果：

上述 SQL 的 In 子句和之前介绍到的 Inner Join 类似。并且 In 子查询也会涉及到大状态问题，要注意设置 State 的 TTL。

常用 Connector 读写

之前我们已经用过了一些简单的内置连接器，比如 ‘datagen’ 、‘print’ ，其它的可以查看官网：Overview | Apache Flink

Kafka

1）添加kafka连接器依赖

• 将flink-sql-connector-kafka-1.20.0.jar上传到flink的lib目录下
• 重启flink集群、sql gateway、sql-client

​ 使用 kafka 连接器，我们需要清楚，我们用 Flink SQL 往连接器为 kafka 的表中插入数据就相当于 Flink 往 Kafka 写入数据，而我们查询 Flink SQL 表中的数据就相当于 从 Kafka 中读取数据。所以当我们建表时就需要初始化读取 Kafka 数据和消费 Kafka 数据的参数。

2）创建 kfaka 的映射表

CREATE TABLE t1( 
  `event_time` TIMESTAMP(3) METADATA FROM 'timestamp',
  --列名和元数据名一致可以省略 FROM 'xxxx', VIRTUAL表示只读
  `partition` BIGINT METADATA VIRTUAL,
  `offset` BIGINT METADATA VIRTUAL,
	id int, 
	ts bigint , 
	vc int )
WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = 'node1.itcast.cn:9092',
  'properties.group.id' = 'test01',
	-- 'earliest-offset', 'latest-offset', 'group-offsets', 'timestamp' and 'specific-offsets'
  'scan.startup.mode' = 'earliest-offset',
  -- fixed为flink实现的分区器，一个并行度只写往kafka一个分区
	'sink.partitioner' = 'fixed',
  'topic' = 'test',
  'format' = 'json'
);

上面有一个参数 ‘sink.partitioner’ 的值是 ‘fixed’ ，我们之前学过 Kafka 的生产者的分区器有默认的 hash分区器和粘性分区器，这种 fixed 分区器是 kafka 为flink实现的 ，一个并行度只写往一个 kafka 分区，我们可以查看一下 FlinkFixedPartition 的源码：

创建好的表格是没有数据的，所以我们再创建一个数据源往 kfaka 里插入数据：

Flink SQL> CREATE TABLE source ( 
     id INT, 
     ts BIGINT, 
     vc INT
 ) WITH ( 
     'connector' = 'datagen', 
     'rows-per-second'='1', 
     'fields.id.kind'='random', 
     'fields.id.min'='1', 
     'fields.id.max'='10', 
     'fields.ts.kind'='sequence', 
     'fields.ts.start'='1', 
     'fields.ts.end'='1000000', 
     'fields.vc.kind'='random', 
     'fields.vc.min'='1', 
     'fields.vc.max'='100'
);

插入数据：

insert into t1(id,ts,vc) select id,ts,vc from source;

查询 kafka 表：

select * from t1;

3）upsert-kafka 表

如果当前表存在更新操作，那么普通的kafka连接器将无法满足（因为普通的连接器不支持更新操作），此时可以使用Upsert Kafka连接器。

Upsert Kafka 连接器支持以 upsert 方式从 Kafka topic 中读取数据并将数据写入 Kafka topic。

作为 source，upsert-kafka 连接器生产 changelog 流，其中每条数据记录代表一个更新或删除事件。更准确地说，数据记录中的 value 被解释为同一 key 的最后一个 value 的 UPDATE，如果有这个 key（如果不存在相应的 key，则该更新被视为 INSERT）。用表来类比，changelog 流中的数据记录被解释为 UPSERT，也称为 INSERT/UPDATE，因为任何具有相同 key 的现有行都被覆盖。另外，value 为空的消息将会被视作为 DELETE 消息。

作为 sink，upsert-kafka 连接器可以消费 changelog 流。它会将 INSERT/UPDATE_AFTER 数据作为正常的 Kafka 消息写入，并将 DELETE 数据以 value 为空的 Kafka 消息写入（表示对应 key 的消息被删除）。Flink 将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中。

（1）创建upsert-kafka的映射表(必须定义主键)

CREATE TABLE t2( 
    id int , 
    sumVC int ,
    -- 主键必须 not enforced
    primary key (id) NOT ENFORCED 
)
WITH (
  'connector' = 'upsert-kafka',
  'properties.bootstrap.servers' = 'node1.itcast.cn:9092',
  'topic' = 'test',
  'key.format' = 'json',
  'value.format' = 'json'
);

(2）插入 upset-kafka 表

insert into t2 select  id,sum(vc) sumVC  from source group by id;

(3) 查询 upset-kafka 表

select * from t2;

查询结果：

可以看到，upsert-kafka 表是支持数据更新操作的。

File

Flink 天生就支持本地系统、HDFS 等。

1）创建 FileSystem 映射表

CREATE TABLE t3( id int, ts bigint , vc int )
WITH (
  'connector' = 'filesystem',
  -- 如果是本地系统就用 file:/// 
  'path' = 'hdfs://node1:8020/data/t3',
  'format' = 'csv'
);

注意：之前我们在 flink 的 lib 目录下放了 hive 的连接器，这个包会和 flink 的依赖产生冲突：java.lang.ClassNotFoundException: org.apache.flink.table.planner.delegation.DialectFactory 我们需要把这个依赖移除掉或者改名并重启 sqlSession ：

# 重命名连接器
mv flink-connector-hive_2.12-1.20.0.jar flink-connector-hive_2.12-1.20.0.jar.del

插入数据：

查询插入结果：

除了上面这种方式，我们还可以把 flink 目录下 opt/ 的 flink-table-planner_2.12-1.20.0.jar 和 lib/ 下面的 flink-table-planner-loader-1.20.0.jar 替换一下位置，这样我们就不用把 hive 的连接器移除带了。

sql-client 使用 savepoint

前置操作，在flink安装目录的lib文件夹下，放入两个jar文件：

• commons-cli-1.5.0.jar

• flink-shaded-hadoop-3-uber-3.1.1.7.2.9.0-173-9.0.jar

1）提交一个insert作业，可以给作业设置名称

Flink SQL> create table sink(
 id int,
 ts bigint,
 vc int
 )with(
 'connector' = 'print'
);

insert into sink select * from source;

2）查看 job 列表

查看 job 列表是为了获得 job id，我们提交作业的时候会返回一个 job id 可以在 shell 命令行看到，或者从 web ui 端也可以看到，再或者通过下面的命令看：

show jobs;

3）停止作业，触发 savepoint

SET state.checkpoints.dir='hdfs://node1:8020/chk';
SET state.savepoints.dir='hdfs://node1:8020/sp';
-- 结束作业不设置保存点
stop job 'e6d3e9afed97aee7819c460a6e109445';
-- 结束作业设置保存点
stop job 'e6d3e9afed97aee7819c460a6e109445' with savepoint;

4）从 savepoint 恢复

-- 设置从savepoint恢复的路径 
SET execution.savepoint.path='hdfs://node1:8020/sp/savepoint-0e0742-7e2154873185';  
 
-- 之后直接提交sql，就会从savepoint恢复
 
--允许跳过无法还原的保存点状态
set 'execution.savepoint.ignore-unclaimed-state' = 'true'; 

5）恢复后重置路径

**注意：**我们设置 savepoint 恢复路径后，之后的所有 insert 任务都会默认使用这个 savepoint，所以下一个作业一定要重置这个配置参数：

指定execution.savepoint.path后，将影响后面执行的所有DML语句，可以使用RESET命令重置这个配置选项。

RESET execution.savepoint.path;

如果出现reset没生效的问题，可能是个bug（包括 pipeline.name 这个参数也是），我们可以退出sql-client，再重新进，不需要重启flink的集群。

CateLog

Catalog 提供了元数据信息，例如数据库、表、分区、视图以及数据库或其他外部系统中存储的函数和信息。它本来翻译过来就是目录，我们可以理解为它就是数据库的目录。

数据处理最关键的方面之一是管理元数据。元数据可以是临时的，例如临时表、UDF。我们之前上面使用的表都是基于内存的一个 Catelog ，所以每次我们退出 sql-client 客户端的时候，这些表和数据库就不见了。元数据也可以是持久化的，例如 Hive MetaStore 中的元数据。Catalog 提供了一个统一的API，用于管理元数据，并使其可以从 Table API 和 SQL 查询语句中来访问。

Catalog 允许用户引用其数据存储系统中现有的元数据，并自动将其映射到 Flink 的相应元数据。例如，Flink 可以直接使用 Hive MetaStore 中的表的元数据，不必在Flink中手动重写ddl，也可以将 Flink SQL 中的元数据存储到 Hive MetaStore 中。Catalog 极大地简化了用户开始使用 Flink 的步骤，并极大地提升了用户体验。

注意：catalog 可以使得 mysql 、hive 和 flink 互通有无，互通就是可以操作读写（除了建表），而不是说只是在某个生命周期内起作用，只要连接上，flink 操作的就是实实在在的 hive 、mysql 本身，这才叫互通，而不是自嗨。

CateLog 类型

目前 Flink 包含了以下四种 Catalog：

• GenericInMemoryCatalog：基于内存实现的 Catalog，所有元数据只在session 的生命周期（即一个 Flink 任务一次运行生命周期内）内可用。默认自动创建，会有名为“default_catalog”的内存Catalog，这个Catalog默认只有一个名为“default_database”的数据库。
• JdbcCatalog：JdbcCatalog 使得用户可以将 Flink 通过 JDBC 协议连接到关系数据库。Postgres Catalog和MySQL Catalog是目前仅有的两种JDBC Catalog实现，将元数据存储在数据库中。
• HiveCatalog：有两个用途，一是单纯作为 Flink 元数据的持久化存储，二是作为读写现有 Hive 元数据的接口。注意：Hive MetaStore 以小写形式存储所有元数据对象名称。Hive Metastore以小写形式存储所有元对象名称，而 GenericInMemoryCatalog会区分大小写。
• 用户自定义 Catalog：用户可以实现 Catalog 接口实现自定义 Catalog。从Flink1.16开始引入了用户类加载器，通过CatalogFactory.Context#getClassLoader访问，否则会报错ClassNotFoundException。

JdbcCatalog（MySQL）

JdbcCatalog不支持建表，只是打通flink与mysql的连接，可以去读写mysql现有的库表。

1）上传所需jar包到lib下

• flink-connector-jdbc-3.2.0-1.19.jar
• mysql-connector-j-5.1.7.jar

注意：Flink 是冷加载，所以上传后需要 sql-client

2）创建Catalog

JdbcCatalog支持以下选项:

• name:必需，Catalog名称。
• default-database:必需，连接到的默认数据库。
• username: 必需，Postgres/MySQL帐户的用户名。
• password:必需，该帐号的密码。
• base-url:必需，数据库的jdbc url(不包含数据库名)

对于Postgres Catalog，是"jdbc:postgresql://:<端口>"

对于MySQL Catalog，是"jdbc: mysql://:<端口>"

CREATE CATALOG my_jdbc_catalog WITH(
    'type' = 'jdbc',
    -- 这里指定的只是默认使用的数据库 它会把所有数据库导进这个catalog下
    'default-database' = 'test',
    'username' = 'root',
    'password' = '123456',
    'base-url' = 'jdbc:mysql://node1:3306'
);

3）查看 Catalog

show catalogs;

4）使用指定的 Catalog

use catalog my_jdbc_catalog;

我们发现，除了 mysql 的系统数据库看不到，别的都别导进来了。

我们也可以直接往表中插入数据，而不用向之前那样去建立映射表：

insert into ws2 values(2,2,2);

注意：在 jdbcCatalog 下是不支持建表的，什么表都不行（映射表或者普通表）！

要建表需要返回到之前默认的 default_catalog 才可以，但是我们是可以从 jdbc_catalog 去查 default_catalog 下的表数据的。

select * from default_catalog.mydatabase.source;

此外，我们也可以把不同类型catalog下不同的表数据关联在一起：

select * from default_catalog.mydatabase.source s join my_jdbc_catalog.test.ws2 w on s.id=w.id;

最后，每次我们退出 sql-client 的时候，其实我们创建的 jdbc_catalog 还是会被删除的，所以我们最好把创建catalog这些命令写进一个 sql 文件，初始化启动 sql-client 的时候执行一下。

HiveCatalog

同样，HiveCatalog 可以打通所有 Hive 的库和表，这样我们就可以在 Flink 直接读写 Hive 表。此外，我们还可以在 catalog 下创建我们 Flink 的表，比如带有 Kafka 连接器的表，而且即使我们退出客户端，再次进去 HiveCatalog ，那张表还是存在的。

1）上传 jar 包

• flink-connector-hive_2.12-1.20.0.jar
• mysql-connector-j-5.1.7.jar（我的 Hive 元数据存储在 MySQL）

2）更换planner依赖

只有在使用Hive方言或HiveServer2时才需要这样额外的计划器jar移动，但这是Hive集成的推荐设置。

这个我们之前使用 FileSystem 创建映射表的时候已经做过了。

3）重启flink集群和sql-client

4）启动外置的hive metastore服务

Hive metastore必须作为独立服务运行，也就是hive-site中必须配置hive.metastore.uris。（必须启动 hive 的元数据服务，不然我们flink无法获取hive中的数据）

# & 的意思是后台启动
# hive --service metastore &
# 这里直接启动我的 hive
hiveservice.sh start
# 查看hive 启动没有
hiveservice status

启动 hive 后会一直挂在那，我们可以判断一下元数据服务是否启动：

netstat -anp|grep 9083
# 或者
ps -ef|grep -i metastore

5）创建 Catalog

配置项	必需	默认值	类型	说明
type	Yes	(none)	String	Catalog类型，创建HiveCatalog时必须设置为’hive’。
name	Yes	(none)	String	Catalog的唯一名称
hive-conf-dir	No	(none)	String	包含hive -site.xml的目录,需要Hadoop文件系统支持。如果没指定hdfs协议，则认为是本地文件系统。如果不指定该选项，则在类路径中搜索hive-site.xml。
default-database	No	default	String	Hive Catalog使用的默认数据库
hive-version	No	(none)	String	HiveCatalog能够自动检测正在使用的Hive版本。建议不要指定Hive版本，除非自动检测失败。
hadoop-conf-dir	No	(none)	String	Hadoop conf目录的路径。只支持本地文件系统路径。设置Hadoop conf的推荐方法是通过HADOOP_CONF_DIR环境变量。只有当环境变量不适合你时才使用该选项，例如，如果你想分别配置每个HiveCatalog。

CREATE CATALOG myhive WITH (
    'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/opt/module/hive-3.1.2/conf'
);

6）查看 catalog

我们在 hive 中创建一个数据库 test 再创建一张表 ws：

再往 ws 中插入一条数据：

hive(test)> insert into ws values(1,1,1);

可以看到，即使是插入一条数据，hive 也是转换为一个 MapReduce 作业，所以很慢。

不对劲，是完全卡死了，估计是 flink 和 hive 同时占用 yarn 的资源，导致资源不足 的原因（暂且怀疑是 Yarn 的CPU核数的配置问题），在修改完 yarn 的最大内存发现在 flink 往 hive 插入查询数据都没有问题了，但是 hive 还是不行，那就暂且用 flink 端操作 hive 吧。

没毛病，在 flink sql 客户端往 hive 插入数据后，在 flink 和 hive 中都可以看到结果。

​ 此外，我们在 hiveCatalog 下创建一张连接器为 FileSystem 的表，那么这张表是只能在 flink 环境下才能查到的，在hive是只能看到有这张表，查不到数据的：

去 hive 端查看一下：

我们发现，hive 端可以查看到存在这张表。

当我们查的时候，发现直接报错，毕竟它不是一个满足 hive 规范的表。同样，我们在 hiveCatalog 下创建一个连接器为 jdbcCatalog 的表，同样在 flink sql 中也是可以查到并正常使用的，但是在 hive 端同样只能看到表名。

现在，我们退出 sql-client，我们重新变价一下初始化 sql 文件：

重新启动：

我们可以看到，虽然每次退出 sql-client 之后 catalog 下次启动就消失了，但是catalog 下面的表不会消失，我们只需要创建对应的 catalog 即可。

我们再解决一下刚才 hive插入数据失败的问题：

关闭hadoop、hive，修改yarn-site.xml 中 yarn的最大cpu核数=8（无所谓，反正在自己电脑上测试）。重启 hadoop、hive，测试插入：

说明 hive 没有问题，但是还不能说明我们配置的 yarn 有用，我们再启动 flink 的 yarn-session 和 sql-client：

提交插入作业（插入数据到 hive）：

没毛病，这下问题测彻底解决了。