关于Flink:基于-Flink-打造的伴鱼实时计算平台-Palink-的设计与实现

作者：李辉

在伴鱼倒退晚期，呈现了一系列实时性相干的需要，比方算法工程师冀望能够拿到用户的实时特色数据做实时举荐，产品经理心愿数据方能够提供实时指标看板做实时经营剖析。

这个阶段中台数据开发工程师次要是基于 Spark 实时计算引擎开发作业来满足业务方提出的需要。然而这类作业并没有对立的平台进行治理，工作的开发模式、提交形式、可用性保障等也齐全因人而异。

随同着业务的减速倒退，越来越多的实时场景涌现进去，对实时作业的开发效率和品质保障提出了更高的要求。为此，咱们从去年开始着手打造伴鱼公司级的实时计算平台，平台代号 Palink，由 Palfish + Flink 组合而来。

之所以抉择 Flink 作为平台惟一的实时计算引擎，是因为近些年来其在实时畛域的优良体现和主导地位，同时沉闷的社区气氛也提供了十分多不错的实践经验可供借鉴。目前 Palink 我的项目曾经落地并投入使用，很好地满足了伴鱼业务在实时场景的需要。

外围准则

通过调研阿里云、网易等各大厂商提供的实时计算服务，咱们根本确定了 Palink 的整个产品状态。同时，在零碎设计过程中紧紧围绕以下几个外围准则：

• 极简性： 放弃繁难设计，疾速落地，不适度谋求性能的完整性，满足外围需要为主；
• 高质量： 放弃我的项目品质严要求，外围模块思虑周全；
• 可扩大： 放弃较高的可扩展性，便于后续计划的迭代降级。

零碎设计

平台整体架构

以下是平台整体的架构示意图：

整个平台由四局部组成：

• Web UI： 前端操作页面；
• Palink (GO) 服务： 实时作业管理服务，负责作业元信息及作业生命周期内全副状态的治理，承接全副的前端流量。包含作业调度、作业提交、作业状态同步及作业 HA 治理几个外围模块；
• PalinkProxy(JAVA) 服务：SQL 化服务，Flink SQL 作业将由此模块编译、提交至远端集群。包含 SQL 语法校验、SQL 作业调试及 SQL 作业编译和提交几个外围模块；
• Flink On Yarn： 基于 Hadoop Yarn 做集群的资源管理。

这里之所以将后盾服务拆分成两块，并且别离应用 GO 和 JAVA 语言实现，起因次要有三个方面：

• 一是伴鱼领有一套十分欠缺的基于 GO 语言实现的微服务根底框架，基于它能够疾速构建服务并领有包含服务监控在内的一系列周边配套，公司目前 95% 以上的服务是基于此服务框架构建的；
• 二是 SQL 化模块是基于开源我的项目二次开发实现的（这个在后文会做具体介绍），而该开源我的项目应用的是 JAVA 语言；
• 三是外部服务减少一次近程调用的老本是能够承受的。

这里也体现了咱们极简性准则中对疾速落地的要求。事实上，以 GO 为外围开发语言是十分具备 Palfish 特色的，在接下来伴鱼大数据系列的相干文章中也会有所体现。

接下来本文将着重介绍 Palink 几个外围模块的设计。

作业调度 & 执行

后端服务接管到前端创立作业的申请后，将生成一条 PalinkJob 记录和一条 PalinkJobCommand 记录并长久化到 DB，PalinkJobCommand 为作业提交执行阶段形象出的一个实体，整个作业调度过程将围绕该实体的状态变更向前推动。其构造如下：

type PalinkJobCommand struct {
 ID            uint64 `json:"id"`                       
 PalinkJobID   uint64 `json:"palink_job_id"`  
 CommandParams string `json:"command_params"` 
 CommandState  int8   `json:"command_state"`  
 Log           string `json:"log"`                      
 CreatedAt     int64  `json:"created_at"`        
 UpdatedAt     int64  `json:"updated_at"`      
}

这里并没有间接基于 PalinkJob 实体来串联整个调度过程，是因为作业的状态同步会间接作用于这个实体，如果调度过程也基于该实体，两局部的逻辑就紧耦合了。

调度流程

下图为作业调度的流程图：

palink pod 异步执行竞争分布式锁操作，保障同一时刻有且仅有一个实例获取周期性监测权限，满足条件的 Command 将间接被发送到 Kafka 待执行队列，同时变更其状态，保障之后不再被调度。此外，所有的 palink pod 将充当待执行队列消费者的角色，并归属于同一个消费者组，生产到音讯的实例将获取到最终的执行权。

执行流程

作业的执行实则是作业提交的过程，依据作业类型的不同提交工作流有所区别，可细分为三类：

• Flink JAR 作业： 咱们摒弃了用户间接上传 JAR 文件的交互方式。用户只需提供作业 gitlab 仓库地址即可，打包构建全流程平台间接实现。因为每一个服务实例都内嵌 Flink 客户端，工作是间接通过 Flink run 形式提交的。
• PyFlink 作业： 与 Flink JAR 形式相似，少了编译的过程，提交命令也有所不同。
• Flink SQL 作业： 与上两种形式区别较大。对于 Flink SQL 作业而言，用户只需提交绝对简略的 SQL 文本信息，这个内容咱们是间接保护在平台的元信息中，故没有和 gitlab 仓库交互的中央。SQL 文本将进一步提交给 PalinkProxy 服务进行后续的编译，而后应用 Yarn Client 形式提交。

Command 状态机

PalinkJobCommand 的状态流转如下图所示：

• UNDO： 初始状态，将被调度实例监测。
• DOING： 执行中状态，同样会调度实例监测，避免长期处于进行中的脏状态产生。
• SUCCESSED： 执行胜利状态。随着用户的后续行为，如从新提交、重新启动操作，状态会再次回到 UNDO 态。
• FAILED： 执行失败状态。同上，状态可能会再次回到 UNDO 态。

作业状态同步

作业胜利提交至集群后，因为集群状态的不确定性或者其余的一些因素最终导致工作异样终止了，平台该如何及时感知到？这就波及到咱们行将要论述的另一个话题 “状态同步“。

状态同步流程

这里首先要答复的一个问题是：同步谁的状态？

有过离线或者 Flink on yarn 开发教训的同学肯定晓得，作业在部署到 yarn 上之后会有一个 application 与之对应，每一个 application 都有其对应的状态和操作动作，比方咱们能够执行 Yarn UI 上 Kill Application 操作来杀掉整个工作。

同样的，当咱们翻阅 Flink 官网文档或者进入 Flink UI 页面也都能够看到每一个工作都有其对应的状态和一系列操作行为。最间接的想法必定是以 Flink 工作状态为准，毕竟这是咱们最想拿到的。

但仔细分析，其实二者的状态对于平台而言没有太大区别，只是状态的粒度有所不同而已，yarn application 的状态曾经是对 Flink 状态做了一次 state mapping。可是思考到，Flink 在 HA 的时候，作业对外裸露的 URL 会产生变更，这种状况下只能通过获取作业对应的 application 信息能力拿到最新的地址。

与此同时，一次状态同步的过程不仅仅只是心愿拿到最新的状态，对于工作的 checkpoint 等相干信息同样是有同步的诉求。看来二者的信息在一次同步的过程中都须要获取，最终的状态同步设计如下：

前置流程和作业调度流程相似，有且仅有一个实例负责周期性监测工作，符合条件的 Job ID（注，并非所有的作业都用同步的必要，比方一些处于终态的作业）将发送到外部提早队列。之所以采纳提早队列而非 Kafka 队列，次要是为了将同一时间点批量同步的需要在肯定工夫距离内随机打散，升高同步的压力。最初，在获取到作业的残缺信息后，再做一次 state mapping 将状态映射为平台形象的状态类型。

因为状态同步是周期性进行的，存在肯定的提早。因而在平台获取作业详情时，也会同步触发一次状态同步，保障获取最新数据。

Job 状态机

PalinkJob 的状态流转如下图所示：

• DEPLOYING： 作业初始状态，将随着 PalinkJobCommand 的状态驱动向 DEPLOY_SUCCESSED 和 DEPLOY_FAILED 流转。
• DEPLOY_SUCCESSED： 部署胜利状态，依赖作业「状态同步」驱动向 RUNNING 状态或者其余终态流转。
• DEPLOY_FAILED： 部署失败状态，依赖用户从新提交向 DEPLOYING 状态流转。
• RUNNING： 运行中状态。可通过用户执行暂停操作向 FINISHED 状态流转，或执行终止操作向 KILLED 状态流转，或因为外部异样向 FAILED 状态流转。
• FINISHED： 实现状态，作业终态之一。通过用户执行暂停操作，作业将回到此状态。
• KILLED： 终止状态，作业终态之一。通过用户执行终止操作，作业将回到此状态。
• FAILED： 失败状态，作业终态之一。作业异样会转为此状态。

作业 HA 治理

解决了上述问题之后，另一个待探讨的话题便是 “作业 HA 治理”。咱们须要答复用户以下的两个问题：

• 作业是有状态的，然而作业须要代码降级，如何解决？
• 作业异样失败了，怎么做到从失败的工夫点复原？

Flink 提供了两种机制用于复原作业：Checkpoint 和 Savepoint，本文统称为保留点。Savepoint 能够看作是一种非凡的 Checkpoint，只不过不像 Checkpoint 定期的从零碎中生成，它是用户通过命令触发的，用户能够管制保留点产生的工夫点。

工作启动时，通过指定 Checkpoint 或 Savepoint 内部门路，就能够达到从保留点复原的成果。咱们对于平台作业 HA 的治理也是基于这两者开展的。下图为治理的流程图：

用户有两种形式来手动进行一个作业：暂停和终止。

• 暂停操作通过调用 Flink cancel api 实现，将触发作业生成 Savepoint。
• 终止操作则是通过调用 yarn kill application api 实现，用于疾速完结一个工作。

被暂停的作业重启时，零碎将比拟 Savepoint 和 Checkpoint 的生成工夫点，依照最近的一个保留点启动，而当作业被从新提交时，因为用户可能变更了代码逻辑，将间接由用户决定是否依照保留点复原。对于被终止的作业，无论是重启或者是从新提交，都间接采取由用户决定的形式，因为终止操作自身就带有抛弃作业状态的色调。

失败状态的作业是因为异样谬误被迫进行的。对于这类作业，有三重保障：

• 一是工作本身能够设置重启策略主动复原，内部平台无感知；
• 二是，对于外部重启仍旧失败的工作在平台侧可再次设置下层重启策略；
• 三是，手动重启或从新提交。仅在从新提交时，由用户决定依照那种形式启动，其余场景皆依照最近的保留点启动。

工作 SQL 化

Flink JAR 和 PyFlink 都是采纳 Flink API 的模式开发作业，这样的模式必然极大地减少用户的学习老本，影响开发的效率。须要一直输出和造就具备该畛域开发技能的工程师，能力满足源源不断的业务需要。

而产品定位不仅仅是面向数据中台的开发工程师们，咱们冀望能够和离线指标用户保持一致，将目标群体浸透至剖析人员乃至业务研发和局部的产品经理，简略的需要齐全能够本人入手实现。要达到这个目标，必然开发的模式也要向离线看齐，作业 SQL 化是势在必行的。

咱们冀望 Flink 能够提供一种相似于 Hive Cli 或者 Hive JDBC 的作业提交形式，用户无需写一行 Java 或 Scala 代码。查阅官网文档，Flink 的确提供了一个 SQL 客户端以反对以一种简略的形式来编写、调试和提交表程序到 Flink 集群，不过截止到目前最新的 release 1.13 版本，SQL 客户端仅反对嵌入式模式，相干的性能还不够健全，另外对于 connector 反对也是无限的。因而，须要寻求一种更稳固、更高可扩展性的实现计划。

通过一番调研后，咱们发现袋鼠云开源的「FlinkStreamSQL」根本能够满足咱们目前的要求。此我的项目是基于开源的 Flink 打造的，并对其实时 SQL 进行了扩大，反对原生 Flink SQL 所有的语法。

实现机制

下图为 Flink 官网提供的作业角色流程图，由图可知，用户提交的代码将在 Client 端进行加工、转换（最终生成 Jobgraph）而后提交至近程集群。

那么要实现用户层面的作业 SQL 化，底层的实现同样是绕不开这个流程。实际上 FlinkStreamSQL 我的项目就是通过定制化的伎俩实现了 Client 端的逻辑，能够将整个过程简要地形容为：

构建 PackagedProgram

利用 PackagedProgramUtils 生成 JobGraph。

通过 YarnClusterDescriptor 提交作业。

其中，第一步是最要害的，PackagedProgram 的构造方法如下：

PackagedProgram.newBuilder()
                .setJarFile(coreJarFile)
                .setArguments(execArgs)
                .setSavepointRestoreSettings(savepointRestoreSettings)
                .build();

execArgs 为内部输出参数，这里就蕴含了用户提交的 SQL。而 coreJarFile 对应的就是 API 开发方式时用户提交的 JAR 文件，只不过这里零碎帮咱们实现了。coreJarFile 的代码对应我的项目中的 core module，该 module 实质上就是 API 开发方式的一个 template 模板。module 内实现了自定义 SQL 解析以及各类 connector plugin 注入。更多细节可通过开源我的项目进一步理解。

定制开发

咱们基于 FlinkStreamSQL 进行了二次开发，以满足外部更多样化的需要。次要分为以下几点：

• 服务化：整个 SQL 化模块作为 proxy 独立部署和治理，以 HTTP 模式裸露服务；
• 反对语法校验个性；
• 反对调试个性：通过解析 SQL 构造可间接获取到 source 表和 sink 表的构造信息。平台可通过人工结构或线上抓取源表数据的形式失去测试数据集，sink 算子被 localTest connector 算子间接替换，以截取后果数据输入；
• 反对更多的 connector plugin，如 pulsar connector；
• 其余个性。

除了上文提到的一些性能个性，平台还反对了：

• DDL 语句注入
• UDF 治理
• 租户治理
• 版本治理
• 作业监控
• 日志收集

这些点就不在本文具体论述，但作为一个实时计算平台这些点又是必不可少的。

线上成果

作业总览

作业详情

作业监控

将来工作

随着业务的持续推动，平台将在以下几方面持续迭代优化：

• 稳定性建设： 实时工作的稳定性建设必然是将来工作中的首要事项。作业参数如何设置，作业如何主动调优，作业在流量顶峰如何保持稳定的性能，这些问题须要一直摸索并积淀更多的最佳实际；
• 晋升开发效率：SQL 化建设。只管 SQL 化已初具雏形，但开发起来仍旧具备肯定的学习老本，其中最显著的就是 DDL 的构建，用户对于 source、sink 的 schema 并不分明，最好的形式是平台能够和咱们的元数据中心买通将构建 DDL 的过程自动化，这一点也是咱们目前正在做的；
• 优化应用体验： 体验上的问题在肯定水平上也间接影响到了开发的效率。通过一直收集用户反馈，继续改良；
• 摸索更多业务场景： 目前伴鱼外部已开始基于 Flink 发展 AI、实时数仓等场景的建设。将来咱们将持续推动 Flink 在更多场景上的实际。