关于后端:OPPO大数据离线任务调度系统OFLOW

1 离线调度零碎

在整个大数据体系中，在原始数据被采集之后，须要应用各种逻辑进行整合和计算之后能力输入理论无效的数据，能力最终用于商业目标，实现大数据的价值。在整个解决流程中，无论是抽取、转换、装载（ETL）的这些过程，还是数据用户剖析处理过程，都是须要蕴含泛滥的解决工作，而且这些工作都不是孤立的，而是存在相互依赖和束缚关系的。如何高效的调度和治理这些工作是十分要害的，影响到各个流程中的数据的及时性和准确性。在这个过程中工作的高效治理和调度是十分要害的，会影响到各个流程中的数据的及时性和准确性。

一个最简略的任务调度零碎莫过于 Linux 零碎自带的 crontab，应用简略，运行稳固。

在我的项目刚起步时应用 crontab 无可非议，随着调度工作的增多，相互之间又有着依赖，crontab 就远远满足不了开发的需要了。这些工作的状态各种各样，工作之间也存在多种多样的依赖关系。一个工作的执行须要一系列的前置工作的实现。比方一个上游工作 A 实现特定逻辑之后，而上游的工作 B 则依赖工作 A 输入的数据后果能力产生本人的数据和后果。因而为了保证数据的准确性和可靠性，就必须依据这些工作之间的依赖关系从上游到上游有序的执行。怎么样让大量的工作精确的实现调度而不呈现问题，甚至在任务调度执行中呈现谬误的状况下，工作可能实现自我复原甚至执行谬误告警与残缺的日志查问。大数据离线任务调度零碎就是要施展这样的作用。

调度零碎的外围性能次要就是如下三点：

组织和治理工作流程，定时调度和执行工作，解决工作间依赖关系。

对于一个欠缺的离线调度零碎，须要有以下外围性能：

1. 作为大数据体系中的一个指挥核心，负责依据工夫，依赖，工作优先级，资源等条件调度工作；
2. 须要能解决工作的多种依赖关系，包含工夫依赖，工作上下游依赖，本身依赖等；
3. 数据量微小，工作品种繁多，须要执行多种工作类型，如 MapReduce，hive，spark 以及 shell，python 等；
4. 须要有一个欠缺的监控系统监控整个调度和执行的过程，保障任务调度和执行的整个链条，过程中出现异常状况能即便发送告警告诉。

咱们的 OFLOW 零碎就是为了实现以上需要的。

2 OFLOW 零碎在 OPPO 的利用

OFLOW 目前提供的外围性能次要以下几点：

1. 高效准时的任务调度；
2. 灵便的调度策略：工夫，上下游依赖，工作本身依赖；
3. 多种工作类型：数据集成、Hive、Python、Java、MapReduce、Spark、SparkSQL、Sqoop、机器学习工作等；
4. 业务间隔离，工作过程间隔离；
5. 高可用，可扩大；
6. 工作配置：参数，失败重试(次数，距离)，失败和超时告警，不同级别告警，工作回调；
7. 丰盛全面的操作页面，工作的开发、运维、监控等操作图形化页面化；
8. 权限治理；
9. 实时查看工作状态和剖析日志，并进行进行、重跑、补录等各种运维操作；
10. 工作历史数据分析；
11. 脚本开发，测试，公布流程；
12. 告警监控：多种异常情况的状态监控，灵便配置；
13. 外围工作重点监控，保障准点率；
14. 反对 API 接入。

目前 OFLOW 在我司曾经承当了十分多的工作的调度。

OFLOW 现有国内，新加坡，印度，欧盟和北美 5 大集群，欧盟和北美集群最近不久上线的，工作临时还没上量。目前主力集群是国内，新加坡和印度。

目前用户能够通过以下几种形式接入到 OFLOW：

1. oflow 的 webserver;
2. 南天门平台，其中的数据研发 – 离线工作模块，数据集成工作模块，离线脚本开发模块。后端的任务调度和执行全副也是在 oflow 零碎上；
3. oflow 还反对通过 api 的形式接入，目前也曾经有多个业务通过 api 的形式应用 oflow 零碎；

3 OFLOW 零碎的设计和演进

依据后面的信息，能够看到整个离线调度零碎最外围的是两个组件，一个的调度引擎，一个是执行引擎。

调度引擎依据工作属性 (周期，提早，依赖关系等) 调度工作，依据工作优先级，队列和资源状况散发到不同的执行节点；

执行引擎获取满足执行条件的工作，执行工作，同时输入工作执行过程中的日志，并监控工作执行过程。

在目前市面上常见的离线调度零碎中，airflow 能够说是其中的佼佼者，通过了多年的倒退，性能曾经十分欠缺，在开源社区也十分沉闷。

Airflow 于 2014 年 10 月由 Airbnb 的 Maxime Beauchemin 开始；

2015 年 6 月发表正式退出 Airbnb Github；

2016 年 3 月退出了 Apache Software Foundation 的孵化打算；

目前的更新迭代版本曾经到了 1 -10 版本；2- 1 版本。

咱们 oppo 的离线调度零碎是在 airflow 1.8 版本上引入进来的。

上面是几个在 airflow 零碎中的概念，在其它的离线调度零碎中也有相似的概念。

1. DAG：即有向无环图(Directed Acyclic Graph)，将所有须要运行的 tasks 依照依赖关系组织起来，形容的是所有 tasks 执行的依赖关系。在 airflow 中，DAG 由一个可执行的 python 脚本来定义。
2. Operators：能够了解为一个工作模板，形容了 DAG 中一个具体的 task 要做的事件。airflow 内置了很多 operators，如 BashOperator 用来执行 bash 命令，PythonOperator 调用任意 Python 函数，EmailOperator 用于发送邮件，HTTPOperator 用于发送 HTTP 申请，SqlOperator 用于执行 SQL 命令…同时，用户能够自定义 Operator，这给用户提供了极大的便利性。他的作用就像 java 中的 class 文件。
3. Sensor 是一类非凡的 Operator，是被特定条件触发的，比方 ExteralTaskSensor, TimeSensor, TimeDeltaSensor。
4. Tasks：Task 是 Operator 的一个实例，也就是 DAGs 中的一个 node，当用户实例化一个 operator，即用一些参数特例化一个 operator，就生成了一个 task。
5. DagRun：当 dag 文件被 airflow 辨认并调度时，运行中的 DAG 即为 dagRun。在 webUi 界面能够看到。
6. Task Instance：task 的一次运行。即运行起来的 task，task instance 有本人的状态，包含“running”,“success”,“failed”,“skipped”,“up for retry”等。

在 airflow 中，定义 dag 和 dag 中的工作是通过一个 python 文件实现的，这就是一个例子。

这个 py 文件是来定义 dag 的，尽管它也开源间接运行，但独自运行并没有什么成果，只是检测 python 语法是否正确。他也不执行具体的工作，只是形容工作之间的依赖关系，以及调度的工夫距离等要求。

这个须要在任务调度和执行时进行解析，能力依照设定逻辑调度，依照用户设定的执行步骤运行。

这样一个 python 文件就对数据开发人员提出了比拟高的要求，须要平台的用户对 python 编码很纯熟才行。

下图是 airflow 的整体架构设计，其中 airflow home dags 用于存储定义 dag 和 task 的 python 文件，webserver 用于提供 web 服务，展现 dag 视图，实例，日志等十分多的信息，airflow 的 web 页面也是很欠缺的。scheduler 是调度节点，执行 dag 解析，任务调度的工作；worker 节点是执行节点，能够有很多组，能够监听不同的队列，作用是执行 scheduler 调度起来的工作。

咱们 oppo 的离线调度零碎 oflow 就是在开源 airflow 的根底上开发的。

开发中解决的几个比拟外围的问题是：

1. 将 dag 和 task 的定义从 python 文件批改为 web 配置数据库存储，同时 dag 的解析也是从解析 python 文件批改为了从数据库查问并解析。
2. 另外一个就是和公司的大数据开发平台联合，实现开发，测试和公布流程，不便用户的开发，测试验证和公布流。
3. 另外还增加了很多的监控告警，用来比拟全面的监控任务调度和执行的整个流程；

如下是咱们 OFLOW 平台的整个架构：

1. webserver 用来提供 web 服务，不便用户进行 dag，task 的配置以及十分多的信息查问；
2. scheduler 是调度节点，负责工作的调度，通过解析 dag 和 task，进行一系列的逻辑判断工作是否满足调度条件；
3. worker 是执行节点，负责工作实例的执行；
4. api server 是咱们起初开发中新增的一个组件，用来解耦 worker 和咱们数据库的操作，后续也承当的其它的一些性能；
5. 应用 mysql 存储 dag，task，task_instance 等所有的元数据信息；
6. 应用 celery 做音讯队列，broker 应用的是 redis；同时 redis 也充当了缓存的作用；
7. oflow 也同时接入云监控负责发送告警信息，应用 ocs 用于存储日志和用户脚本文件；
8. 同时 oflow 也接入了诊断平台，这个是最新接入的，帮助用户对异样的 oflow 工作进行诊断；

如下这个图显示了整个任务调度和执行的整个流程：

目前 OFLOW 也有了比拟全面的监控：

以上就是 OFLOW 的整体架构，任务调度和执行整个流程。

目前 OFLOW 的整个服务也存在一些问题：

1. 任务调度距离问题：
   依据后面的任务调度的流程，咱们能够看到，oflow 工作的调度是通过 scheduler 周期扫描解析 dag 和 task 的。这种形式就会造成工作上下游之间会有肯定工夫的提早。比方 A 工作实现后，间接上游工作 B 并不能马上被调度执行，须要期待 scheduler 下次扫描时扫到改工作能力被触发。如果工作的依赖深度比拟深，上下游链条很长，每两个工作间有肯定距离，整体的间隔时间就会比拟久。尤其是在凌晨任务调度顶峰这样的工夫点。
2. 服务高可用问题：
   原生的 oflow 不反对高可用。目前咱们的计划是筹备一个备节点，在检测到 scheduler 异样时，能够拉起备用节点。
3. 业务增长造成的调度压力问题：
   目前 oflow 每日的任务量十分多，而且也在快速增长，oflow 的调度压力也是越来越高，目前的计划的对 scheduler 进行横向扩大，让不同的 scheduler 调度不同的 dag；
4. 调度峰谷的老本问题：
   离线调度工作的一个很显著的特色就是存在工作的顶峰和低谷。oflow 的天级别和小时级别的调度工作是最多的，这样就会造成在每天的凌晨工夫是任务调度的大顶峰，在每小时的前一段时间是调度的小顶峰，而其它时间段则是低谷。顶峰状态工作会呈现队列拥挤状况，而低谷工夫，机器是处于比拟闲暇的状态。如何更无效的利用系统资源，也是值得咱们后续思考和优化的点。

4 全新的离线调度零碎 OFLOW 2.0

上面再向大家介绍一下，近期曾经上线试用的 OFLOW 2.0 的产品非凡和架构设计。

咱们 oflow 2.0 平台想解决的问题有以下几点：

1. 工作实时触发，升高上下游工作之间的提早；
2. 不再以 dag 去组织和调度工作。以 dag 为调度维度，就会存在跨周期依赖的问题。理论中会有很多工作须要依赖其它 dag 的工作，比方一个天级别的工作须要依赖另一个小时级别的 dag 的某个工作在 24 个周期要全副实现。目前 oflow 的解决方案是通过一个跨 dag 依赖工作 ExternalTaskSensor 去实现的。这个无论是在工作配置上，还是在对概念的了解上，都存在一些问题；
3. 另外就是心愿能简化配置，oflow 的 dag 和 task 的性能比拟弱小，然而配置也十分多，用户实现一个 dag，一个 task 的配置须要了解很多概念，输出很多信息。这样益处是比拟灵便，然而毛病就是很不不便。咱们 2.0 就心愿可能简化配置，暗藏一些不必要的概念和配置；
4. 同时还心愿能更使用户在工作开发，测试和公布等一系列流程更加便捷；
5. 2.0 的各个组件能在高可用和可扩展性上更加便捷简略。

oflow 2.0 零碎就通过以和 1.0 差异很大的设计实现这些需要：

1. 工作实时触发；
2. 认为业务流程形式组织工作，而非 dag，不再须要跨 dag 依赖的概念；
3. 各个组件的可扩展性；
4. 零碎的标准化：简化了很多工作的配置，操作门槛更低。工作执行环境标准化，缩小环境问题，升高运维方面的老本。

oflow 2.0 的整体架构设计如下：

oflow 2.0 以后是没有供用户应用的前端页面，是通过南天门 2.0 的离线模块调用 oflwo 1.0 的 api server。所以你们在应用 oflow 2.0 的离线模块时，后端的数据存储，工作触发，调度，执行等一系列流程都是在 oflow 2.0 的平台上实现的。

1. 首先的这个组件就是 api server。除了南天门调用之外，oflow 2.0 外部的 worker 执行节点也和 api server 有很多交互；apiserver 次要实现的是和 2.0 数据库的交互，业务流程，工作，实例等各项操作，以及上游工作触发等外在逻辑；
2. Trigger 组件的性能比拟纯正，就是负责扫描工作进行触发；
3. scheduler 调度节点负责工作的调度解析，通过工夫轮，工作依赖信息管理，工作优先级和队列等一系列的服务和治理来剖析和调度工作；
4. worker 节点和 1.0 的逻辑比拟靠近，负责工作的理论执行过程，反对了包含 shell, python, sparkSQL 和数据集成工作这四种大的类型的工作，同时也反对用户对开发的脚本进行测试，工作执行日志的解决，反对对正在执行的工作进行进行操作，同时还有工作执行完结后的回调逻辑；
5. Monitor 组件一方面是负责监控外部各个组件，其它各个组件在启动后都会向 monitor 进行注册，后续一旦节点出问题，monitor 能够对在该节点上调度和执行的工作进行解决。monitor 同时还负责解决工作执行过程中的各种告警信息和一些告诉性信息的发送；

其中还有两个音讯队列，

1. 一个是 Schedule MQ，负责接管满足局部调度条件能够开始调度的工作并转交给 scheduler 去解决；
2. 另一个是 Task MQ，负责接管满足所有依赖条件，能够执行的工作，worker 端从队列中获取工作并生产。

除了这些开发的组件之外，oflow 2.0 也用到了一些通用的产品，包含 MySQL, Redis，以及对象存储存在，云监控零碎，以及调用了公司 IT 零碎的一些 api。

这张图展现了 OFLOW 的任务调度和执行的整个流程：

其中调度开始入口有两个，一个是 trigger, 一个是 webserver。

trigger 负责提前 5 分钟扫描行将要执行的工作，扫描进去之后放入到 schedule mq 中；

webserver 负责多个触发逻辑，一方面是用户手动触发的工作重跑和补录操作，另一个是上游某个工作实现后，将其间接上游获取进去，放入到 schedule mq；

这些音讯在 schedule mq 中会被 scheduler 生产，schedule 会剖析工作实例的所有依赖关系，包含工夫依赖，上下游依赖，本身依赖等信息。如果工作的各种依赖条件都满足，则会被放到 task mq 中被 worker 生产；不满足工夫依赖的工作会被放入到工夫轮中，等达到相应工夫刻度后会主动触发；不满足执行条件的工作的所有依赖信息保留在 redis 中，等后续工夫达到，或者依赖的上游工作实现，会不断更新该实例的依赖信息，直到所有依赖条件满足。满足依赖条件的工作，schedule 也会剖析工作所属的我的项目以及工作优先级等配置信息，将工作放入到 task mq 中的不同的音讯队列中；

worker 会从 task mq 中生产工作。拿到工作后，通过获取的工作的详细信息，而后执行。判断工作执行后果，如果执行胜利，则会告诉到 api server, api server 除了更新实例状态等信息外，还会同时查问该工作的间接上游，将其间接上游放入到 schedule mq 中；如果工作失败，则会依据是否还有重试次数决定是否要重试，如果没有重试次数则会认定工作失败，告诉到 api server, api serer 更新实例状态。

目前 OFLOW 2.0 曾经实现了所有的设计，开发和测试环境，应通过了一段时间的内测和压力测试等环节。最近也曾经凋谢试用了。欢送大家试用 2.0 零碎，并在试用过程中给与反馈和倡议。

目前用户如果想应用咱们的 OFLOW 2.0 零碎的话，能够登录南天门 2.0 平台上试用。

5 结语

以上就是我跟大家分享的 OFLOW 的一些信息。

在此我也瞻望一下咱们后续 OFLOW 平台的倒退：

1）OFLOW 1.0 的调度性能问题。因为 2.0 和 1.0 零碎的变动较大，后续 OFLOW 1.0 和 2.0 平台会在一段较长的工夫内共存，因而对 1.0 零碎的调度性能咱们也须要一直去优化，以应答高速增长的任务量；

一方面是想方法缩短工作间的调度距离，以晋升工作执行效率；

另一方面是心愿能摸索更便捷无效的扩大形式，应答调度任务量的减少。

2）交互体验上

页面交互的敌对性上进行欠缺；增加一系列的批量工作操作和运维方面的性能；同时还心愿以 dag 或者 task 等维度展现历史统计信息，以供用户参考；另外就是针对工作操作审计，工作的监控零碎进行优化；

3）老本优化

另外一个就是后面提到的老本优化，下图反映的是一天中 24 个小时的工作并发执行状况，工作存在非常明显的顶峰和低谷。

后续思考想方法对工作错峰执行，比方在计费模式下来激励用户将时效性要求不高的工作放在工作低谷进行执行；另外一个就是心愿摸索一下资源的动静扩缩容来实现老本优化。

4）另外还心愿后续 OFLOW 不单单起到一个任务调度的作用，心愿后续能和后端的大数据集群有更多的交互；

5）还有一点就是心愿对监控进行进一步的欠缺。其中比拟要害的一个是外围工作的链路的辨认和监控。

就是岂但要能监控到外围工作，还能将该外围工作的所有上游逻辑监控到，链路中的某个环节一旦异样，可能很快的告警进去；另外一点是用户收到告警时的解决，很多用户收到工作告警后不分明如何解决，后续 oflow 会想方法疏导用户解决。

作者简介

Chengwei OPPO 高级后端工程师

次要负责 OPPO 的大数据离线任务调度零碎的开发工作，对大数据离线调度零碎有比拟丰盛的开发教训。

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