Flink SQL客户端

Flink SQL客户端

1.概述

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

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

2.入门

启动Flink集群

./bin/start-cluster.sh
1

启动SQL客户端命令行界面

./bin/sql-client.sh embedded
1

执行SQL查询

SELECT 'Hello World';
1

该查询不需要 table source，并且只产生一行结果。CLI 将从集群中检索结果并将其可视化。按 Q 键退出结果视图。

CLI 为维护和可视化结果提供三种模式。

表格模式（table mode）在内存中实体化结果，并将结果用规则的分页表格可视化展示出来。执行如下命令启用：

SET sql-client.execution.result-mode=table;
1

变更日志模式（changelog mode）不会实体化和可视化结果，而是由插入（+）和撤销（-）组成的持续查询产生结果流。

SET sql-client.execution.result-mode=changelog;
1

Tableau模式（tableau mode）更接近传统的数据库，会将执行的结果以制表的形式直接打在屏幕之上。具体显示的内容会取决于作业 执行模式的不同(execution.type)：

SET sql-client.execution.result-mode=tableau;
1

你可以用如下查询来查看三种结果模式的运行情况：

SELECT name, COUNT(*) AS cnt FROM (VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name) GROUP BY name;
1

此查询执行一个有限字数示例：

变更日志模式 下，看到的结果应该类似于：

+ Bob, 1
+ Alice, 1
+ Greg, 1
- Bob, 1
+ Bob, 2
1
2
3
4
5

表格模式 下，可视化结果表将不断更新，直到表程序以如下内容结束：

Bob, 2
Alice, 1
Greg, 1
1
2
3

Tableau模式 下，如果这个查询以流的方式执行，那么将显示以下内容：

+-----+----------------------+----------------------+
| +/- |                 name |                  cnt |
+-----+----------------------+----------------------+
|   + |                  Bob |                    1 |
|   + |                Alice |                    1 |
|   + |                 Greg |                    1 |
|   - |                  Bob |                    1 |
|   + |                  Bob |                    2 |
+-----+----------------------+----------------------+
Received a total of 5 rows
1
2
3
4
5
6
7
8
9
10

如果这个查询以批的方式执行，显示的内容如下：

+-------+-----+
|  name | cnt |
+-------+-----+
| Alice |   1 |
|   Bob |   2 |
|  Greg |   1 |
+-------+-----+
3 rows in set
1
2
3
4
5
6
7
8

这几种结果模式在 SQL 查询的原型设计过程中都非常有用。这些模式的结果都存储在 SQL 客户端 的 Java 堆内存中。为了保持 CLI 界面及时响应，变更日志模式仅显示最近的 1000 个更改。表格模式支持浏览更大的结果，这些结果仅受可用主内存和配置的最大行数（sql-client.execution.max-table-result.rows）的限制。

注意:

在批处理环境下执行的查询只能用表格模式或者Tableau模式进行检索。

3.配置

进入SQL 客户端后，输入help命令：

help;
1

其他配置，请参考配置

4.提交任务

如何具体使用SQL客户端提交一个任务，请参考配置

5.兼容

为了与以前兼容，SQL Client仍然支持用环境YAML文件初始化，并允许在YAML文件中设置key。当设置YAML文件中定义了key时，SQL Client将打印要通知的警告消息。

Flink SQL> SET execution.type = batch;
[WARNING] The specified key 'execution.type' is deprecated. Please use 'execution.runtime-mode' instead.
[INFO] Session property has been set.

-- all the following DML statements will be restored from the specified savepoint path
Flink SQL> INSERT INTO ...
1
2
3
4
5
6

当使用SET命令打印属性时，SQL Client也将打印所有的属性。为了区分deprecated的key，sql客户端使用[deprecated] 作为标识符。

Flink SQL>SET;
execution.runtime-mode=batch
sql-client.execution.result-mode=table
table.planner=blink
[DEPRECATED] execution.planner=blink
[DEPRECATED] execution.result-mode=table
[DEPRECATED] execution.type=batch
ult-mode=table
table.planner=blink
[DEPRECATED] execution.planner=blink
[DEPRECATED] execution.result-mode=table
[DEPRECATED] execution.type=batch
1
2
3
4
5
6
7
8
9
10
11
12