Flink on Zeppelin 作业管理系统实践

1. 业务背景介绍

在数仓ETL、实时计算的场景下，我们基于Flink SQL批流一体的框架进行了一定规模的作业迁移。在研发作业管理系统中，我们引入Apache Zeppelin组件作为Flink SQL作业提交客户端，Flink 批流作业可视化预览的核心组件。在一年多时间的产线实践中，我们对作业提交的方式策略进行了几次演进，目前在跑作业规模Flink Batch 任务日均运行超5000次，流作业500 ，均稳定运行。

1.1 Apache Zeppelin 介绍

Apache Zeppelin是一款基于Web的Notebook产品，能够交互式数据分析。使用Zeppelin，您可以使用丰富的预构建语言后端（或解释器）制作交互式的协作文档，例如Scala、Python、SparkSQL、Hive、FlinkSQL等。

在Flink的集成方面，Zeppelin支持Flink的3种主流语言，包括Scala、PyFlink和SQL。同时，Zeppelin也提供了三种模式的Flink 任务提交，Local模式、Remote模式、Yarn Session/Application模式，社区在对Kubernetes Application模式也在开发中。Zeppelin还支持在解析器/任务作用域的Flink运行时参数配置，集成hive catalog ，并支持简易的cron job执行，并对多版本Flink均提供了支持，最新的master分支支持了最新的Flink 1.15版本。

以下是Flink on Zeppelin 主要feature：‍

1.2 基于NoteBook作业提交的痛点

在最初任务较少时，我们将批、流作业都运行在单节点Zeppelin server中，直接使用SQL模式进行运行，由于每个长跑作业都需要建立实时监控，对server压力很大，调度任务从外部运行SQL，也经常出现卡顿，无法提交作业的情况。后来我们改用pyflink后台作业提交，作业监控额外通过监控程序管理，但随着任务增加，单台节点无法满足任务提交需要，期间做了批、流server独立拆分，增加单节点机器配置等，但依然无法稳定。

主要问题有以下：

• Zeppelin Server单点故障导致已经运行流作业失败，批作业无法正常提交；最初使用yarn这种模式提交，客户端 Flink Interpreter 进程运行在 Zeppelin所在的机器这边，每个客户端对应一个Yarn上的Flink Cluster，如果Flink Interpreter进程很多，会对Zeppelin这台机器造成很大的压力，导致进程挂死。
• 并发提交任务几乎不可能，虽然后续切换Yarn Application 模式可以把Flink interpreter 跑在了JobManager里 缓解客户端压力，但同时大规模提交pyflink作业仍存在执行效率问题；
• 无法灵活个性化参数，解析器提前创建出，只能通过不断的新建notebook，控制session cluster 通过解析器提供的作用域，解析器配置错误影响所有关联notebook的任务提交。

2. 架构改进

2.1 改造后批/流作业提交架构

• 流作业提交优化

1. 通过调用Zeppelin Server的rest api 新建Flink解析器；
2. 新建notebook及paragraph，执行Pyflink程序，解析作业参数，执行依赖包加载及作业执行时配置；
3. 通过自研job  scheduler 对作业配置信息进行分析，判断作业提交方式为remote 还是yarn；
4. 并发提交作业时，首先会进入资源队列，通过判断临时解析器数量，超过一定数量时，等待释放资源提交；
5. remote模式提交到hadoop yarn 中已经存在的job manager中，共享管理资源；
6. yarn模式通过解析器新建flink cluster ；
7. 作业提交后，通过回调Zeppelin api，获取当次作业的提交信息记录到作业日志数据库中，包含yarn application id及job id，并提交至flink统一后台监控程序监控；
8. 销毁解析器进程，归档作业notebook归档。
9. 批作业提交优化

1. 在统一作业管理中注册Flink Batch SQL 作业，并配置调度时间及依赖关系；
2. Airflow 生成dag，定时触发执行；
3. 每一组任务执行时，首先新建EMR 集群，初始化Zeppelin环境；
4. 通过Airflow 程序访问Zeppelin API使用同一个作用域为全局的解析器配置模板生成解析器；
5. 同时为每一个Flink SQL 作业新建notebook，并执行作业SQL；
6. 通过Zeppelin同步API执行所有notebook完成后，记录此组作业的最终执行结果及异常日志；
7. 完成写入日志表后，销毁EMR集群。

2.2 作业提交架构优化收益

• 流作业支持了以作业组为单位的Flink On Yarn作业提交，每次提交作业独立创建解析器，提交完成后销毁解析器，有效降低了Zeppelin server的负载，通过作业调度管理器可以将同一个分组的作业提交到同一个cluster，共享作业资源。支持并发提交流作业并不会对Zeppelin server造成负担。具有水平扩展性，作业调度器可以兼容多个Zeppelin server 作为客户端提交作业；
• 批作业与流作业的Zeppelin server独立开，每次运行批作业使用AWS EMR 集成的Zeppelin进行任务提交，具备很强的水平扩展性。同一批作业运行规模也可随EMR的节点规模及节点类型进行垂直扩展，使得批作业提交不受Zeppelin单节点限制。

3. 实践要点

3.1 Python 环境及包管理

在运行pyflink过程中，需要提交将python依赖包安装到环境中，这里我们使用anaconda将python环境预先打包通过code build 存储到S3存储中，在执行pyflink 之前，首先使用Shell解析器初始化python环境，通过配置Flink 解析中python的路径，访问安装好依赖的环境。

环境包管理流程

3.2 AirFlow 批作业调度

我们通过对Zeppelin Rest API 封装了Zeppelin Airflow的operator，支持了几个重要的操作，如通过yaml模板创建Zeppelin解析器，创建notebook、paragraph，运行指定paragraph，记录Zeppelin 运行日志，销毁解析器，归档notebook等。可以很方便地基于operator对Zeppelin server进行访问。

通过作业管理系统，我们将注册的任务记录在mysql数据库中，使用Airflow 通过扫描数据库动态创建及更新运行dag，将flink batch sql 封装为一类task group，包含了创建AWS EMR 临时集群，初始化Zeppelin服务，并通过Airflow的operator进行作业提交。

3.3 Flink SQL流作业资源调度

如前所述，通过自研作业管理系统，提交流作业时，主要执行pyflink进行任务的后台提交，虽然通过临时创建解析器，提交后销毁的方式可以有效减轻Zeppelin server压力，但是如果作业并发提交时，依然会遇到执行python造成内存及cpu负载。所以，在作业提交的资源调度上，进行提交队列的缓冲，限制Zeppelin server同时执行的并发数，并可以方便地进行多server提交作业。

对于同一个作业组的作业提交，如果第一次提交，会通过解析器创建flink cluster进行任务提交，而对于组内已有作业运行，即通过获取rpc 地址进行remote模式提交，类似如下图的入参。

4. 未来展望

Jobschedule对多版本Flink支持

Flink 及相关组件的版本频繁，为了支持A/B测试及业务迁移验证，后续需要支持提交不同的Flink 版本，而Zeppelin天然提供了对多版本Flink的支持，我们后续只需简单适配就可以实现。

更加灵活的参数及依赖包管理模式

后续对特定作业的运行时参数及依赖包需要支持可定制，灵活配置，当然仅限新任务提交到新的cluster生效。

Developed by KLOOK Data Team(Taylor、Wilson、Singh)

​

本文为从大数据到人工智能博主「jellyfin」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。

原文链接：https://cloud.tencent.com/developer/article/2100179