Flink系列之：Flink SQL Gateway
- 一、Flink SQL Gateway
- 二、部署
- 三、启动SQL Gateway
- 四、运行 SQL 查询
- 五、SQL 网关启动选项
- 六、SQL网关配置
- 七、支持的端点
一、Flink SQL Gateway
SQL 网关是一项允许多个客户端从远程并发执行 SQL 的服务。它提供了一种简单的方法来提交 Flink 作业、查找元数据并在线分析数据。
SQL 网关由可插入端点和 SqlGatewayService 组成。 SqlGatewayService 是一个由端点重用来处理请求的处理器。端点是允许用户连接的入口点。根据端点的类型，用户可以使用不同的实用程序进行连接。
二、部署
本节介绍如何从命令行设置和运行您的第一个 Flink SQL 程序。
SQL Gateway 捆绑在常规 Flink 发行版中，因此可以开箱即用。它只需要一个正在运行的Flink集群，可以在其中执行表程序。如果您只是想尝试 SQL 客户端，您还可以使用以下命令启动一个由一名工作人员组成的本地集群：
$ ./bin/start-cluster.sh
三、启动SQL Gateway
SQL Gateway 脚本也位于 Flink 的二进制目录中。用户可以通过调用以下方式开始：
$ ./bin/sql-gateway.sh start -Dsql-gateway.endpoint.rest.address=localhost
该命令启动带有 REST 端点的 SQL 网关，该端点侦听地址 localhost:8083。您可以使用curl命令检查REST端点是否可用。
$ curl http://localhost:8083/v1/info
{"productName":"Apache Flink","version":"1.20-SNAPSHOT"}
四、运行 SQL 查询
要验证您的设置和集群连接，您可以执行以下步骤。
第 1 步：打开会话
$ curl --request POST http://localhost:8083/v1/sessions
{"sessionHandle":"..."}
SQL Gateway 使用返回结果中的 sessionHandle 来唯一标识每个活动用户。
第 2 步：执行查询
$ curl --request POST http://localhost:8083/v1/sessions/${sessionHandle}/statements/ --data '{"statement": "SELECT 1"}'
{"operationHandle":"..."}
返回结果中的operationHandle用于SQL Gateway唯一标识提交的SQL。
第 3 步：获取结果
有了上面的sessionHandle和operationHandle，就可以取到相应的结果了。
$ curl --request GET http://localhost: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 网关启动选项
$ ./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 网关。
$ ./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网关配置
您可以在下面启动 SQL Gateway 时配置 SQL Gateway，或者任何有效的 Flink 配置条目：
$ ./sql-gateway -Dkey=value
| key | Default | Type | Description |
|---|---|---|---|
| sql-gateway.session.check-interval | 1 min | Duration | 空闲会话超时的检查间隔，可以通过设置为零来禁用。 |
| sql-gateway.session.idle-timeout | 10 min | Duration | 当会话在该时间间隔内没有被访问时关闭会话的超时时间。如果设置为零，会话将不会关闭。 |
| sql-gateway.session.max-num | 1000000 | Integer | SQL Gateway 服务的最大活动会话数。 |
| sql-gateway.session.plan-cache.enabled | false | Boolean | 如果为 true，sql gateway 将缓存并重用每个会话的查询计划。 |
| sql-gateway.session.plan-cache.size | 100 | Integer | 计划缓存大小，当table.optimizer.plan-cache.enabled为true时生效。 |
| sql-gateway.session.plan-cache.ttl | 1 hour | Duration | plan缓存的TTL，它控制缓存写入后多久过期，当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 服务的最小工作线程数。 |
七、支持的端点
Flink 原生支持 REST Endpoint 和 HiveServer2 Endpoint。默认情况下，SQL 网关与 REST 端点捆绑在一起。凭借灵活的架构，用户可以通过调用来启动具有指定端点的SQL Gateway
$ ./bin/sql-gateway.sh start -Dsql-gateway.endpoint.type=hiveserver2
或者在Flink配置文件中添加以下配置：
sql-gateway.endpoint.type: hiveserver2
注意：如果 Flink 配置文件中还包含 sql-gateway.endpoint.type 选项，则 CLI 命令具有更高的优先级。
