31、Flink的SQL Gateway介绍及示例

Flink 系列文章

一、Flink 专栏

Flink 专栏系统介绍某一知识点，并辅以具体的示例进行说明。

1、Flink 部署系列本部分介绍Flink的部署、配置相关基础内容。
2、Flink基础系列本部分介绍Flink 的基础部分，比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。
3、Flik Table API和SQL基础系列本部分介绍Flink Table Api和SQL的基本用法，比如Table API和SQL创建库、表用法、查询、窗口函数、catalog等等内容。
4、Flik Table API和SQL提高与应用系列本部分是table api 和sql的应用部分，和实际的生产应用联系更为密切，以及有一定开发难度的内容。
5、Flink 监控系列本部分和实际的运维、监控工作相关。

二、Flink 示例专栏

Flink 示例专栏是 Flink 专栏的辅助说明，一般不会介绍知识点的信息，更多的是提供一个一个可以具体使用的示例。本专栏不再分目录，通过链接即可看出介绍的内容。

两专栏的所有文章入口点击：Flink 系列文章汇总索引

文章目录

本文介绍了Flink gateway的启动配置、支持的api以及简单的hiveserver2 endpoint部分，特别是api均以示例进行说明。
本文依赖flink集群能正常使用。
本文分为3个部分，即gateway启动、rest api使用和hiveserver2 endpoint。
本文的示例是在Flink 1.17版本中运行。

一、概述

SQL Gateway 是一种使远程多个客户端能够并发执行SQL的服务。它提供了一种提交Flink Job、查找元数据和在线分析数据的简单方法。

SQL Gateway 由可插入endpoints 和SqlGatewayService组成。SqlGatewayService是一个由endpoints 重用以处理请求的处理器。endpoints 是允许用户进行连接的入口点。根据endpoints 的类型，用户可以使用不同的utils进行连接。

在这里插入图片描述

1、启动

SQL Gateway 捆绑在常规的Flink发行版中，因此可以开箱即用。它只需要一个正在运行的Flink集群，在那里可以执行表程序。有关设置Flink集群的更多信息，请参阅集群和部署部分。如果您只是想试用SQL客户端，还可以使用以下命令启动一个带有一个工作进程的本地集群：

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

[root@minio_api_1049 bin]# hostname
minio_api_1049
[root@minio_api_1049 bin]# ./sql-gateway.sh start -Dsql-gateway.endpoint.rest.address=192.168.10.49
Starting sql-gateway daemon on host minio_api_1049.
[root@minio_api_1049 bin]# jps29396 SqlClient
6180 SqlGateway
28424 TaskManagerRunner
28077 StandaloneSessionClusterEntrypoint
6430 Jps
[root@minio_api_1049 bin]# curl http://192.168.10.49:8083/v1/info{"productName":"Apache Flink","version":"1.17.1"}[root@minio_api_1049 bin]#

或者在浏览器访问
在这里插入图片描述

2、查询示例

以下三个步骤是紧密结合在一起的，也就是从第一步开始做。

1）、Step 1: Open a session

SQL网关使用返回结果中的sessionHandle来唯一标识每个活动用户。

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

示例

[root@minio_api_1049 bin]# curl --request POST http://192.168.10.49:8083/v1/sessions{"sessionHandle":"52c748a3-60be-4131-9c5e-65e872beb3ac"}

2）、Step 2: Execute a query

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

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

示例

[root@minio_api_1049 bin]# curl --request POST http://192.168.10.49:8083/v1/sessions/52c748a3-60be-4131-9c5e-65e872beb3ac/statements/ --data '{"statement": "SELECT 1"}'
{"operationHandle":"5b85b27c-30df-4deb-9f46-7054777f0242"}

3）、Step 3: Fetch results

使用上面的sessionHandle和operationHandle，可以获取相应的结果。

$ curl--request GET http://localhost:8083/v1/sessions/${sessionHandle}/operations/${operationHandle}/result/0

示例

[root@minio_api_1049 bin]# curl --request GET http://192.168.10.49:8083/v1/sessions/52c748a3-60be-4131-9c5e-65e872beb3ac/operations/5b85b27c-30df-4deb-9f46-7054777f0242/result/0{"resultType":"PAYLOAD",
    "isQueryResult": true,
    "jobID":"583fd3a557efc5edc5783de1e1491245",
    "resultKind":"SUCCESS_WITH_CONTENT",
    "results":{"columns":[{"name":"EXPR$0",
            "logicalType":{"type":"INTEGER",
                "nullable":false},
            "comment": null
        }],
        "rowFormat":"JSON",
        "data":[{"kind":"INSERT",
            "fields":[1]}]},
    "nextResultUri":"/v1/sessions/52c748a3-60be-4131-9c5e-65e872beb3ac/operations/5b85b27c-30df-4deb-9f46-7054777f0242/result/1"}

结果中的nextResultUri用于在不为null的情况下获取下一批结果。

curl--request GET ${nextResultUri}

示例

[root@minio_api_1049 bin]# curl --request GET http://192.168.10.49:8083/v1/sessions/52c748a3-60be-4131-9c5e-65e872beb3ac/operations/5b85b27c-30df-4deb-9f46-7054777f0242/result/1{"resultType":"EOS",
    "isQueryResult": true,
    "jobID":"583fd3a557efc5edc5783de1e1491245",
    "resultKind":"SUCCESS_WITH_CONTENT",
    "results":{"columns":[{"name":"EXPR$0",
            "logicalType":{"type":"INTEGER",
                "nullable":false},
            "comment": null
        }],
        "rowFormat":"JSON",
        "data":[]}}

3、配置

1）、启动配置

截至Flink 1.17版本，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.

2）、运行配置

可以在下面启动SQL网关时配置SQL网关，也可以配置任何有效的Flink配置条目：

./sql-gateway -Dkey=value

在这里插入图片描述

4、支持的Endpoints

Flink本机支持REST Endpoints 和HiveServer2 Endpoints。默认情况下，SQL网关与REST端点绑定在一起。使用灵活的体系结构，用户可以通过调用

./bin/sql-gateway.sh start -Dsql-gateway.endpoint.type=hiveserver2

或者conf/flink-conf.yaml配置

sql-gateway.endpoint.type: hiveserver2

客户端的优先级高于配置文件的优先级，如果配置项一致的情况下。

二、REST Endpoint

REST端点允许用户使用REST API连接到SQL网关。

1、SQL处理概述

参考上文中的查询示例。

1）、Open Session

当客户端连接到SQL网关时，SQL网关会创建一个会话作为上下文，以在客户端和SQL网关之间的交互过程中存储用户指定的信息。创建会话后，SQL网关服务器将返回一个名为SessionHandle的标识符，用于以后的交互。

2）、Submit SQL

会话注册后，客户端可以向SQL网关服务器提交SQL。提交SQL时，SQL将被转换为Operation，并返回名为OperationHandle的标识符以便稍后获取结果。操作有其生命周期，客户端可以取消操作的执行或关闭操作以释放操作使用的资源。

3）、Fetch Results

使用OperationHandle，客户端可以从Operation中获取结果。如果操作准备就绪，SQL网关将返回一批具有相应模式的数据和用于获取下一批数据的URI。当提取完所有结果后，SQL网关将在响应中用值EOS填充resultType，并且下一批数据的URI为null。
在这里插入图片描述

2、Endpoint Options

在这里插入图片描述

3、REST API

该功能目前处于实验性的。

Open API v1 specification 用户可以提交sql并执行Allow users to submit statements to the gateway and execute.
Open API v2 specification 支持sql client连接gateway Supports SQL Client to connect to the gateway

默认是V2版本。V2版本的yaml文件。

以下示例API V2版本的使用。

1）、/api_versions

API描述

/api_versions请求方式:

GET

http响应码:

200 OK

获取Rest Endpoint的当前可用版本。客户端可以选择一个返回版本作为稍后通信的协议。

请求报文

{}

响应报文

{"type":"object","id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:util:GetApiVersionResponseBody","properties":{"versions":{"type":"array","items":{"type":"string"}}}}

示例

[root@minio_api_1049 conf]# curl --request GET http://192.168.10.49:8083/v2/api_versions{"versions":["V1","V2"]}[root@minio_api_1049 conf]# curl --request GET http://192.168.10.49:8083/v1/api_versions{"versions":["V1","V2"]}

2）、/info

描述

/infohttp请求方式:

GET

http响应码:

200 OK

获取集群的元数据

请求报文

{}

响应报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:util:GetInfoResponseBody",
  "properties":{"productName":{"type":"string"},
    "version":{"type":"string"}}}

示例

[root@minio_api_1049 conf]# curl --request GET http://192.168.10.49:8083/v2/info{"productName":"Apache Flink","version":"1.17.1"}[root@minio_api_1049 conf]# curl --request GET http://192.168.10.49:8083/v1/info{"productName":"Apache Flink","version":"1.17.1"}

3）、/sessions

描述

/sessionshttp请求方式:

POST

http响应码:

200 OK

打开具有特定属性的新会话。可以为当前会话指定特定属性，该属性将覆盖网关的默认属性。

请求报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:session:OpenSessionRequestBody",
  "properties":{"properties":{"type":"object",
      "additionalProperties":{"type":"string"}},
    "sessionName":{"type":"string"}}}

响应报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:session:OpenSessionResponseBody",
  "properties":{"sessionHandle":{"type":"string"}}}

示例

[root@minio_api_1049 conf]# curl --request POST http://192.168.10.49:8083/v1/sessions{"sessionHandle":"2daa0882-3c17-46bc-b7d8-a23ca48d41e2"}[root@minio_api_1049 conf]# curl --request POST http://192.168.10.49:8083/v2/sessions{"sessionHandle":"93597fe1-2574-406f-bce7-6cafb8dee434"}

4）、/sessions/:session_handle

1、Delete 请求方式-关闭session

描述

/sessions/:session_handlehttp请求方式:

DELETE

http响应码:

200 OK

关闭指定的session。路径参数

session_handle - 上文示例中获取的session，作为url中的一个路径参数。
请求报文

{}

响应报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:session:CloseSessionResponseBody",
  "properties":{"status":{"type":"string"}}}

示例

[root@minio_api_1049 conf]# curl --request DELETE http://192.168.10.49:8083/v1/sessions/2daa0882-3c17-46bc-b7d8-a23ca48d41e2{"status":"CLOSED"}

2、Get请求方式-获取session配置信息

描述

/sessions/:session_handlehttp请求方式:

GET

http响应码:

200 OK

获取指定session的配置信息路径参数

session_handle - 上文示例中获取的session，作为url中的一个路径参数。
请求报文

{}

响应报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:session:GetSessionConfigResponseBody",
  "properties":{"properties":{"type":"object",
      "additionalProperties":{"type":"string"}}}}

示例

[root@minio_api_1049 conf]# curl --request GET http://192.168.10.49:8083/v2/sessions/93597fe1-2574-406f-bce7-6cafb8dee434{"properties":{"state.checkpoints.num-retained":"20",
        "jobmanager.execution.failover-strategy":"region",
        "jobmanager.rpc.address":"localhost",
        "jobmanager.bind-host":"0.0.0.0",
        "execution.savepoint.ignore-unclaimed-state":"false",
        "table.resources.download-dir":"/tmp/sql-gateway-93597fe1-2574-406f-bce7-6cafb8dee434",
        "taskmanager.host":"localhost",
        "parallelism.default":"1",
        "taskmanager.numberOfTaskSlots":"2",
        "pipeline.classpaths":"",
        "taskmanager.memory.process.size":"4096m",
        "execution.checkpointing.mode":"EXACTLY_ONCE",
        "taskmanager.bind-host":"0.0.0.0",
        "execution.savepoint-restore-mode":"NO_CLAIM",
        "sql-gateway.endpoint.rest.address":"192.168.10.49",
        "web.cancel.enable":"true",
        "execution.target":"remote",
        "jobmanager.memory.process.size":"2048m",
        "jobmanager.rpc.port":"6123",
        "rest.bind-address":"0.0.0.0",
        "execution.checkpointing.interval":"5000",
        "execution.attached":"true",
        "execution.checkpointing.externalized-checkpoint-retention":"RETAIN_ON_CANCELLATION",
        "execution.shutdown-on-attached-exit":"false",
        "pipeline.jars":"file:/usr/local/bigdata/flink-1.17.1/opt/flink-python-1.17.1.jar",
        "web.submit.enable":"true",
        "rest.address":"192.168.10.49",
        "state.backend":"filesystem"}}

5）、/sessions/:session_handle/complete-statement

描述

/sessions/:session_handle/complete-statementhttp请求方式:

GET

http响应码:

200 OK

在给定位置获取给定语句的完成提示。路径参数

session_handle -上文示例中获取的session，作为url中的一个路径参数。
请求报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:statement:CompleteStatementRequestBody",
  "properties":{"position":{"type":"integer"},
    "statement":{"type":"string"}}}

响应报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:statement:CompleteStatementResponseBody",
  "properties":{"candidates":{"type":"array",
      "items":{"type":"string"}}}}

示例

# 1、获取sessioncurl--request POST http://192.168.10.49:8083/v2/sessions
{"sessionHandle":"95e5a02a-be4f-4158-a3b0-adb23407f865"}# 2、查询curl--request GET http://192.168.10.49:8083/v2/sessions/95e5a02a-be4f-4158-a3b0-adb23407f865/complete-statement -H"Content-Type: application/json"-d'{"statement": "SELECT 1", "position": 0}'[root@minio_api_1049 bin]# curl --request GET http://192.168.10.49:8083/v2/sessions/95e5a02a-be4f-4158-a3b0-adb23407f865/complete-statement -H "Content-Type: application/json" -d '{"statement": "SELECT 1", "position": 0}'{"candidates":[]}

6）、/sessions/:session_handle/configure-session

描述

/sessions/:session_handle/configure-sessionhttp请求方式:

POST

http响应码:

200 OK

使用以下语句配置会话: CREATE TABLE, DROP TABLE, ALTER TABLE, CREATE DATABASE, DROP DATABASE, ALTER DATABASE, CREATE FUNCTION, DROP FUNCTION, ALTER FUNCTION, CREATE CATALOG, DROP CATALOG, USE CATALOG, USE [CATALOG.]DATABASE, CREATE VIEW, DROP VIEW, LOAD MODULE, UNLOAD MODULE, USE MODULE, ADD JAR.路径参数

session_handle - 上文示例中获取的session，作为url中的一个路径参数。
请求报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:session:ConfigureSessionRequestBody",
  "properties":{"executionTimeout":{"type":"integer"},
    "statement":{"type":"string"}}}

响应报文

{}

示例

# 1、获取session# 1、获取sessioncurl--request POST http://192.168.10.49:8083/v2/sessions
{"sessionHandle":"95e5a02a-be4f-4158-a3b0-adb23407f865"}# 2、创建数据库和表curl--request POST http://192.168.10.49:8083/v2/sessions/95e5a02a-be4f-4158-a3b0-adb23407f865/configure-session  -H"Content-Type: application/json"-d'{"statement": " create database db1"'}curl--request POST http://192.168.10.49:8083/v2/sessions/95e5a02a-be4f-4158-a3b0-adb23407f865/configure-session  -H"Content-Type: application/json"-d"{\"statement\": \"CREATE TABLE myTable2 (id INT,name STRING,score DOUBLE) WITH ('connector' = 'csv','csv.filepath' = '/tmp/test.csv','format' = 'csv');\"}"# 创建数据库[root@minio_api_1049 bin]# curl --request POST http://192.168.10.49:8083/v2/sessions/95e5a02a-be4f-4158-a3b0-adb23407f865/configure-session  -H "Content-Type: application/json" -d '{"statement": " create database db1"'}{}# 创建表[root@minio_api_1049 bin]# curl --request POST http://192.168.10.49:8083/v2/sessions/95e5a02a-be4f-4158-a3b0-adb23407f865/configure-session  -H "Content-Type: application/json" -d "{\"statement\": \"CREATE TABLE myTable2 (id INT,name STRING,score DOUBLE) WITH ('connector' = 'csv','csv.filepath' = '/tmp/test.csv','format' = 'csv');\"}"{}

7）、/sessions/:session_handle/heartbeat

描述

/sessions/:session_handle/heartbeathttp请求方式:

POST

http响应码:

200 OK

触发heartbeat以告知服务器客户端处于活动状态，并在配置的超时值内保持会话的活动状态。路径参数

session_handle - 上文示例中获取的session，作为url中的一个路径参数。
请求报文

{}

响应报文

{}

示例

[root@minio_api_1049 conf]# curl --request POST http://192.168.10.49:8083/v1/sessions/5e2c4c35-12c2-45f2-b87c-66506a32ca48/heartbeat{}

8）、/sessions/:session_handle/operations/:operation_handle/cancel

描述

/sessions/:session_handle/operations/:operation_handle/cancelhttp请求方式:

POST

http响应码:

200 OK

取消操作。路径参数

session_handle - 上文示例中获取的session，作为url中的一个路径参数。
operation_handle - 标识操作的OperationHandle。获取方式参考上文中的示例。
请求报文

{}

响应报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:operation:OperationStatusResponseBody",
  "properties":{"status":{"type":"string"}}}

示例

# 1、获取sessioncurl--request POST http://192.168.10.49:8083/v2/sessions
{"sessionHandle":"34e27e78-a138-4524-986d-31dd632a274d"}# 2、获取opertionalcurl--request POST http://192.168.10.49:8083/v2/sessions/34e27e78-a138-4524-986d-31dd632a274d/statements/ --data'{"statement": "SELECT 1"}'{"operationHandle":"0ff95159-85e6-488c-ad4c-e113353ccb73"}# 3、取消操作curl--request POST http://192.168.10.49:8083/v2/sessions/34e27e78-a138-4524-986d-31dd632a274d/operations/0ff95159-85e6-488c-ad4c-e113353ccb73/cancel
# 日志输出，由于本示例是select 1，状态是已经完成，已经完成的不能取消
org.apache.flink.table.gateway.api.utils.SqlGatewayException: Failed to convert the Operation Status from FINISHED to CANCELED for 0ff95159-85e6-488c-ad4c-e113353ccb73.

9）、/sessions/:session_handle/operations/:operation_handle/close

描述

/sessions/:session_handle/operations/:operation_handle/closehttp请求方式:

DELETE

http响应码:

200 OK

Close the operation.路径参数

session_handle - 上文示例中获取的session，作为url中的一个路径参数。
operation_handle - 标识操作的OperationHandle。获取方式参考上文中的示例。
请求报文

{}

响应报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:operation:OperationStatusResponseBody",
  "properties":{"status":{"type":"string"}}}

示例

# 1、获取sessioncurl--request POST http://192.168.10.49:8083/v2/sessions
{"sessionHandle":"34e27e78-a138-4524-986d-31dd632a274d"}# 2、获取opertionalcurl--request POST http://192.168.10.49:8083/v2/sessions/34e27e78-a138-4524-986d-31dd632a274d/statements/ --data'{"statement": "SELECT 1"}'{"operationHandle":"0ff95159-85e6-488c-ad4c-e113353ccb73"}# 3、关闭[root@minio_api_1049 bin]# curl --request DELETE http://192.168.10.49:8083/v2/sessions/34e27e78-a138-4524-986d-31dd632a274d/operations/0ff95159-85e6-488c-ad4c-e113353ccb73/close{"status":"CLOSED"}

10）、/sessions/:session_handle/operations/:operation_handle/result/:token

描述

/sessions/:session_handle/operations/:operation_handle/result/:tokenhttp请求方式:

GET

http响应码:

200 OK

Fetch results of Operation.路径参数

session_handle - 上文示例中获取的session，作为url中的一个路径参数。
operation_handle - 标识操作的OperationHandle。获取方式参考上文中的示例。
token - The OperationHandle that identifies a operation.
查询参数
rowFormat (mandatory): The row format to serialize the RowData.
请求报文

{}

响应报文

{"type":"any"}

示例

# session_handle[root@minio_api_1049 conf]#  curl --request POST http://192.168.10.49:8083/v1/sessions{"sessionHandle":"5e2c4c35-12c2-45f2-b87c-66506a32ca48"}# operation_handle[root@minio_api_1049 conf]# curl --request POST http://192.168.10.49:8083/v1/sessions/5e2c4c35-12c2-45f2-b87c-66506a32ca48/statements/ --data '{"statement": "SELECT 1"}'{"operationHandle":"e1774434-b657-4b20-ab2b-2579f3c0fc47"}# /sessions/:session_handle/operations/:operation_handle/result/:token[root@minio_api_1049 conf]# curl --request GET http://localhost:8083/v1/sessions/5e2c4c35-12c2-45f2-b87c-66506a32ca48/operations/e1774434-b657-4b20-ab2b-2579f3c0fc47/result/0{"resultType":"PAYLOAD",
    "isQueryResult": true,
    "jobID":"49eae7c3a3016df44c36296157958afb",
    "resultKind":"SUCCESS_WITH_CONTENT",
    "results":{"columns":[{"name":"EXPR$0",
            "logicalType":{"type":"INTEGER",
                "nullable":false},
            "comment": null
        }],
        "rowFormat":"JSON",
        "data":[{"kind":"INSERT",
            "fields":[1]}]},
    "nextResultUri":"/v1/sessions/5e2c4c35-12c2-45f2-b87c-66506a32ca48/operations/e1774434-b657-4b20-ab2b-2579f3c0fc47/result/1"# /v1/sessions/5e2c4c35-12c2-45f2-b87c-66506a32ca48/operations/e1774434-b657-4b20-ab2b-2579f3c0fc47/result/1[root@minio_api_1049 conf]# curl --request GET http://localhost:8083//v1/sessions/5e2c4c35-12c2-45f2-b87c-66506a32ca48/operations/e1774434-b657-4b20-ab2b-2579f3c0fc47/result/1{"resultType":"EOS",
    "isQueryResult": true,
    "jobID":"49eae7c3a3016df44c36296157958afb",
    "resultKind":"SUCCESS_WITH_CONTENT",
    "results":{"columns":[{"name":"EXPR$0",
            "logicalType":{"type":"INTEGER",
                "nullable":false},
            "comment": null
        }],
        "rowFormat":"JSON",
        "data":[]}}}

11）、/sessions/:session_handle/operations/:operation_handle/status

描述

/sessions/:session_handle/operations/:operation_handle/statushttp请求方式:

GET

http响应码:

200 OK

Get the status of operation.路径参数

session_handle - 上文示例中获取的session，作为url中的一个路径参数。
operation_handle - 标识操作的OperationHandle。获取方式参考上文中的示例。
请求报文

{}

响应报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:operation:OperationStatusResponseBody",
  "properties":{"status":{"type":"string"}}}

示例

# session_handle[root@minio_api_1049 conf]#  curl --request POST http://192.168.10.49:8083/v1/sessions{"sessionHandle":"5e2c4c35-12c2-45f2-b87c-66506a32ca48"}# operation_handle[root@minio_api_1049 conf]# curl --request POST http://192.168.10.49:8083/v1/sessions/5e2c4c35-12c2-45f2-b87c-66506a32ca48/statements/ --data '{"statement": "SELECT 1"}'{"operationHandle":"e1774434-b657-4b20-ab2b-2579f3c0fc47"}[root@minio_api_1049 conf]# curl --request GET http://localhost:8083//v1/sessions/5e2c4c35-12c2-45f2-b87c-66506a32ca48/operations/e1774434-b657-4b20-ab2b-2579f3c0fc47/status{"status":"FINISHED"}

12）、/sessions/:session_handle/statements

描述

/sessions/:session_handle/statementshttp请求方式:

POST

http响应码:

200 OK

Execute a statement.路径参数

session_handle - 上文示例中获取的session，作为url中的一个路径参数。
请求报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:statement:ExecuteStatementRequestBody",
  "properties":{"executionConfig":{"type":"object",
      "additionalProperties":{"type":"string"}},
    "executionTimeout":{"type":"integer"},
    "statement":{"type":"string"}}}

响应报文

{"type":"object",
  "id":"urn:jsonschema:org:apache:flink:table:gateway:rest:message:statement:ExecuteStatementResponseBody",
  "properties":{"operationHandle":{"type":"string"}}}

示例

# session_handle[root@minio_api_1049 conf]#  curl --request POST http://192.168.10.49:8083/v1/sessions{"sessionHandle":"5e2c4c35-12c2-45f2-b87c-66506a32ca48"}[root@minio_api_1049 conf]# curl --request POST http://localhost:8083/v1/sessions/5e2c4c35-12c2-45f2-b87c-66506a32ca48/statements{"operationHandle":"81432111-c49a-4c6b-b5cc-c7657e68a1ab"}

4、Data Type Mapping

截至Flink 1.17版本，REST端点支持使用查询参数rowFormat序列化RowData。REST端点使用JSON格式来序列化表对象。请参考映射的JSON格式。
REST端点还支持使用PLAIN_TEXT格式序列化RowData，该格式会自动将所有列强制转换为String。

三、HiveServer2 Endpoint

HiveServer2 Endpoint is compatible with HiveServer2 wire protocol and allows users to interact (e.g. submit Hive SQL) with Flink SQL Gateway with existing Hive clients, such as Hive JDBC, Beeline, DBeaver, Apache Superset and so on.
HiveServer2 Endpoint与HiveServer2有线协议兼容，允许用户与现有的配置单元客户端（如配置单元JDBC、Beeline、DBeaver、Apache Superset等）进行Flink SQL网关交互（例如提交配置单元SQL）。

It suggests to use HiveServer2 Endpoint with Hive Catalog and Hive dialect to get the same experience as HiveServer2. Please refer to the Hive Compatibility for more details.

它建议将HiveServer2 Endpoint与配置单元目录和配置单元方言一起使用，以获得与HiveServer2相同的体验。有关详细信息，请参阅配置单元兼容性。

以上，简单介绍了Flink gateway的启动配置、支持的api以及简单的hiveserver2 endpoint部分，特别是api均以示例进行说明。

标签： flink sql gateway

本文转载自: https://blog.csdn.net/chenwewi520feng/article/details/132039965
版权归原作者 一瓢一瓢的饮 alanchan 所有，如有侵权，请联系我们删除。

31、Flink的SQL Gateway介绍及示例

Flink 系列文章

文章目录

一、概述

1、启动

2、查询示例

1）、Step 1: Open a session

2）、Step 2: Execute a query

3）、Step 3: Fetch results

3、配置

1）、启动配置

2）、运行配置

4、支持的Endpoints

二、REST Endpoint

1、SQL处理概述

1）、Open Session

2）、Submit SQL

3）、Fetch Results

2、Endpoint Options

3、REST API

1）、/api_versions

2）、/info

3）、/sessions

4）、/sessions/:session_handle

1、Delete 请求方式-关闭session

2、Get请求方式-获取session配置信息

5）、/sessions/:session_handle/complete-statement

6）、/sessions/:session_handle/configure-session

7）、/sessions/:session_handle/heartbeat

8）、/sessions/:session_handle/operations/:operation_handle/cancel

9）、/sessions/:session_handle/operations/:operation_handle/close

10）、/sessions/:session_handle/operations/:operation_handle/result/:token

11）、/sessions/:session_handle/operations/:operation_handle/status

12）、/sessions/:session_handle/statements

4、Data Type Mapping

三、HiveServer2 Endpoint

发表评论

“31、Flink的SQL Gateway介绍及示例”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航