Flink SQL

1.概述

SQL API 是基于 SQL 标准的 Apache Calcite 框架实现的，可通过纯 SQL 来开发和运行一个Flink 任务。

2. 动态表和持续查询

2.1 动态表（Dynamic Tables）

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

2.2 持续查询（Continuous Query）

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

3. 时间属性

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

3.1 事件时间

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

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

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

ts BIGINT, time_ltz AS TO_TIMESTAMP_LTZ(ts, 3)

3.2 处理时间

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

4. 常用的 Connector 

参考 https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/table/overview/

5. Catelog

Catalog 提供了元数据信息，例如数据库、表、分区、视图以及数据库或其他外部系统中存储的函数和信息。

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

• GenericInMemoryCatalog：基于内存实现的 Catalog，所有元数据只在session 的生命周期（即一个 Flink 任务一次运行生命周期内）内可用。
• JdbcCatalog：JdbcCatalog 使得用户可以将 Flink 通过 JDBC 协议连接到关系数据库。Postgres Catalog和MySQL Catalog是目前仅有的两种JDBC Catalog实现，将元数据存储在数据库中。
• HiveCatalog：有两个用途，一是单纯作为 Flink 元数据的持久化存储，二是作为读写现有 Hive 元数据的接口。注意：Hive MetaStore 以小写形式存储所有元数据对象名称。Hive Metastore以小写形式存储所有元对象名称，而 GenericInMemoryCatalog会区分大小写。
  

6. 代码中使用 FlinkSQL

6.1 创建表环境

Table API和SQL需要一个特别的运行时环境，这就是所谓的“表环境”（TableEnvironment）。它主要负责：

（1）注册Catalog和表；

（2）执行 SQL 查询；

（3）注册用户自定义函数（UDF）；

（4）DataStream 和表之间的转换。

每个表和 SQL 的执行，都必须绑定在一个表环境（TableEnvironment）中。TableEnvironment 是 Table API 中提供的基本接口类，可以通过调用静态的 create() 方法来创建一个表环境实例。方法需要传入一个环境的配置参数 EnvironmentSettings，它可以指定当前表环境的执行模式和计划器（planner）。执行模式有批处理和流处理两种选择，默认是流处理模式；计划器默认使用 blink planner。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

6.2 创建表

1）连接器表（Connector Tables）

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

2）虚拟表

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

tableEnv.createTemporaryView("NewTable", newTable);

6.3 表和流的转换

6.3.1 将流（DataStream）转换成表（Table）

（1）调用 fromDataStream() 方法

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

SingleOutputStreamOperator<WaterSensor> sensorDS = env.fromSource(...)

Table sensorTable = tableEnv.fromDataStream(sensorDS);

（2）调用createTemporaryView()方法

tableEnv.createTemporaryView("sensorTable",sensorDS, $("id"),$("ts"),$("vc"));

6.3.2 将表（Table）转换成流（DataStream）

（1）调用 toDataStream() 方法

tableEnv.toDataStream(table).print();

（2）调用 toChangelogStream() 方法

tableEnv.toChangelogStream(table).print();

6.4 输出表

tableEnv.createTemporaryView("NewTable", newTable);

Table result = ...

result.executeInsert("OutputTable");

6.5 自定义函数（UDF）

UDF主要有以下几类：

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

• 表聚合函数（Table Aggregate Functions）：将多行数据里的标量值转换成一个或多个新的行数据。

（1）注册函数

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

tableEnv.createTemporarySystemFunction("MyFunction", MyFunction.class);

（2）在SQL中调用函数

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

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