关于数据库:Flink-Sql-Gateway的原理与实践

1 背景

咱们在应用 Flink 开发实时工作时，都会用到框架自身提供的 DataStream API，这使得用户不能不用 Java 或者 Scala 甚至 Python 来编写业务逻辑；这种形式尽管灵便且表白性强，但对用户具备肯定的开发门槛，并且随着版本的不断更新，DataStream API 也有很多老版本不兼容的问题。所以 Flink SQL 就成了宽广开发用户的最佳抉择，之所以 Flink 推出 SQL API，次要是因为 SQL 有如下几个重要个性：

1. 申明式 API：
   用户只关怀做什么, 而不关怀怎么做；
2. 主动优化：
   屏蔽底层 API 的复杂性, 主动做优化；
3. 简略易懂：
   SQL 利用于不同行业和畛域，学习老本较低；
4. 不易变动：
   语法遵循 SQL 标准规范, 不易变动；
5. 流批对立：
   同样的 SQL 代码, 能够用流和批的形式执行。

尽管 Flink 提供了 SQL 能力，但还是有必要基于 Flink SQL 打造属于本人的平台，目前搭建 SQL 平台的形式有如下几种：

1. Flink 原生 API：
   应用 Flink 提供的 SQL API，封装一个通用的 pipeline jar，利用 flink shell 脚本工具提交 sql 工作；
2. Apache Zeppelin：
   一款开源产品，利用 notebook 形式治理 sql 工作，目前曾经与 Flink 集成，且提供了丰盛的 SDK；
3. Flink Sql Gateway：
   Flink 官网出品的一个 Sql 网关，用 Rest 形式执行 Flink Sql。

第一种形式不足灵活性，且大量提交工作时，有性能瓶颈；而 Zeppelin 尽管功能强大，但页面性能无限，如果要基于 Zeppelin 打造 SQL 平台，要么应用 SDK，要么对 Zeppelin 做重度的二次开发；所以 Flink Sql Gateway 比拟适宜做平台化建设，因为它是一个独立的网关服务，不便与公司现有系统集成，齐全与其它零碎解耦，本文也次要论述 Flink Sql Gateway 的实际与摸索。

2 Flink Sql Gateway 简介

2.1 架构

如上图所示，Flink Sql Gateway 的架构比较简单，次要组件是 SqlGatewayEndpoint，它是基于 Flink 的 RestServerEndpoint 实现的一个 Netty 服务，通过自定义实现多种 handler 来实现 sql 工作的创立和部署，以及治理的能力。SqlGatewayEndpoint 外部次要由 SessionManager（会话治理）组成，SessionManager 保护了一个 session map，而 session 外部次要是一些上下文配置和环境信息。

1. SqlGatewayEndpoint：
   基于 RestServerEndpoint 实现的 Netty 服务，对外提供 Rest Api;
2. SessionManager：
   会话管理器，治理 session 创立与删除；
3. Session：
   一个会话，外面寄存着工作所须要的 Flink 配置和上下文环境信息，负责工作的执行；
4. Classpath：
   Flink Sql Gateway 启动时会加载 flink 装置目录的 classpath，所以 flink sql gateway 基本上没有除 flink 以外的相干依赖。

2.2 执行流程

sql gateway 其实只是一个一般的 NIO 服务器，每个 Handler 都会持有 SessionManager 的援用，因而能够独特拜访同一个 SessionManager 对象。当申请达到时，Handler 会获取申请中的参数，如 SessionId 等，去 SessionManager 中查问对应的 Session，从而执行提交 sql、查问工作状态等工作。申请流程如下图所示：

创立 session，这是应用 sql gateway 的第一步，SessionManager 会把用户传入的工作执行模式、配置、planner 引擎形式等参数封装成 Session 对象，放入 map 中，并返回 sessionid 给用户；

用户持有 sessionid，发动 sql request 的申请，gateway 依据 sessionid 找到对应的 Session 对象，开始部署 sql job 到 yarn / kubernetes；

2.3 性能

2.3.1 工作部署

Flink Sql Gateway 作为 Flink 的客户端，工作部署间接使用了 Flink 的能力，而 Flink 目前反对三种部署模式：

1. in Application Mode，
2. in a Per-Job Mode，
3. in Session Mode。

三种模式有如下两个区别：

1. 集群生命周期和资源隔离：
   per-job mode 的集群生命周期与 job 雷同，但有较强的资源隔离保障。
2. 应用程序的 main()办法是在客户端还是在集群上执行：
   session mode 和 per-job mode 在客户端上执行，而 application mode 在集群上执行。
   

从以上能够看出，Application Mode 为每个应用程序创立一个会话集群，并在集群上执行应用程序的 main() 办法，所以它是 session mode 和 per-job 的一个折中计划。

目前为止，Flink 只反对 jar 包工作的 application mode，所以想要实现 sql 工作的 application mode，须要本人革新实现，前面会讲实现办法。

2.3.2 SQL 能力

Flink Sql Gateway 反对的 Sql 语法如下：

Flink Sql Gateway 反对所有 Flink Sql 语法，但自身也有一些限度：

1. 不反对多条 sql 执行，多条 insert into 执行会产生多个工作；
2. 不残缺的 set 反对，对于 set 语法反对存在 bug；
3. Sql Hit 反对不是很敌对，写在 sql 里比拟容易出错。

3 平台化革新

3.1 SQL 的 application mode 实现

后面说到，flink 不反对 sql 工作的 application mode 部署，只反对 jar 包工作。jar 包工作的 application mode 实现如下图所示：

1. flink-clients 解析出用户的配置和 jar 包信息；
2. ApplicationConfiguration 里指定了 main 办法的入口类名和入参；
3. ApplicationDeployer 负责把 Jobmanager 启动，并且启动时执行 Flink Application 的 main 办法。

通过以上流程能够看出，要实现 sql 的 application mode，实现通用执行 sql 的 pipeline jar 是要害：

实现一个执行 sql 的通用 pipeline jar 包，并且事后传到 yarn 或者 k8s, 如下所示：

在 ApplicationConfiguration 中指定
pepeline jar 的 main 办法入口和参数：

3.2 多 Yarn 集群反对

目前 Flink 只反对单 Yarn 环境的工作部署，对于领有多套 Yarn 环境的场景，须要部署多套 Flink 环境，每个 Flink 对应一个 Yarn 环境配置；尽管这种形式能解决问题，但并不是最优的解决方案。相熟 Flink 应该都晓得，Flink 应用 ClusterClientFactory 的 SPI 来生成与内部资源零碎（Yarn/kubernetes）的拜访介质（ClusterDescriptor），通过 ClusterDescriptor 能够实现与资源零碎的交互，比方 YarnClusterDescriptor，它持有 YarnClient 对象，能够实现与 Yarn 的交互；所以对于多 Yarn 环境，咱们只有保障 YarnClusterDescriptor 里持有的 YarnClient 对象与 Yarn 环境一一对应即可，代码如下图所示：

作者简介

Zheng OPPO 高级数据平台工程师

次要负责基于 Flink 的实时计算平台开发, 对 Flink 有较丰盛的研发教训, 也曾参加过 Flink 社区的奉献。

获取更多精彩内容，扫码关注 [OPPO 数智技术] 公众号