背景数仓以及分析人员在面对日益增长的数据需求时，理想化的方式是让他们专注在模型建设以及业务分析上，其他流程上的工作尽量由系统工程解决。本文将介绍流利说当前工作流中的任务是如何编排的以及治理在整个流程中发挥的价值。

工作流系统

我们所熟知的 Apache Oozie，Airflow 以及 Azkaban 都是优秀的工作流调度系统，简单的配置或者少量的代码就可以创建 DAG(Directed Acyclic Graph 有向无环图)，这里不展开。我们所讨论的是这些系统应该面向谁，在早期，任务调度系统是直接开放给分析人员的，任务的依赖，执行周期由他们自行决定，数据工程师辅助任务的上线，这样的流程有以下问题：上线的任务可能存在问题

任务的依赖可能配置错误，即使提供配置管理页面

分析人员需要分散精力排查任务问题

迷失在错综复杂的依赖关系中，引起其他任务依赖问题

依赖路径过长，集群资源无法被充分利用

显然这样的流程是存在问题的，有没有可能通过系统化的方式来解决这样的问题呢？

任务自动化编排

数据仓库以及分析人员的任务大部分是以 SQL 为主，我们可以借助语法解析器来获取 SQL 的输入表和输出表，在解析的同时还可以校验一下语法是否正确，根据各个任务的输入及输出可以自动构建出一个 DAG。以 SparkSQL 为例，利用 Antlr4 来解析 SQL Query 获得脚本的输入及输出的表，以下为 SqlBaseBaseVisitor 的部分代码：@Overridepublic T visitTableIdentifier(SqlBaseParser.TableIdentifierContext ctx) {String rawName = (null != ctx.db ? ctx.db.getText() + "." + ctx.table.getText() : ctx.table.getText()).replace("`", "”);if (!withCTEList.contains(rawName)) {String table = completeTableName(rawName);meta.addTable(table);if (ctx.parent instanceof SqlBaseParser.InsertOverwriteTableContext) {meta.getInsertedTables().add(table);} else if (ctx.parent instanceof SqlBaseParser.TableNameContext) {meta.getInputTables().add(table);} else if {// other table context}}return this.visitChildren(ctx);}

在成功构建好任务之间的关系后，还需要解决以下问题：建立 ODS 层任务与外部数据源(通过 DataX、Delta 接入的业务表数据)的依赖关系，毕竟在任务被调度时，是不能直接执行 DW 层的任务的，需要等待数据接入的任务完成。以 airflow 为例，需要自定义一个 sensor 去轮询外部数据源的任务状态。如下图 DAG 中的蓝色节点，它做的工作是轮循业务表的数据摄取情况，其状态要么成功，要么超时失败引发告警。由于 SQL 代码托管在 gitlab 上，天然可以支持版本管理，还可以借助 CI 环节来检查 SQL 的语法及语义问题。此外在 repo 中，可以用不同的目录来区分调度频率(如下图)，比如使用 daily，weekly，monthly 来区分其目录下的任务所属的不同调度频率，这样基本满足了大部分离线场景的任务，然后在生成 DAG 时，可以利用 Python Operator 来决定是否达到运行条件，比如在非周一时，可以过滤掉 weekly 的任务。

在成功构建好 DAG 后，需要对其做一系列的校验工作，比如生成的 code 是否可以正常通过编译，在 DAG 中有没有回路，每一份 SQL 脚本作为任务，是不是可以在 DAG 中找到等。所有的检查工作全部通过后，那么当前的 DAG code 可以与最后一次线上的 code 做一次diff，并把 diff 的 git 链接通过 bot 推到 slack channel 中。随后可以部署该 DAG，完成一轮上线工作。整体流程如下：

在完成任务自动化编排后，数仓分析人员脚本的上线工作将会由系统自动完成，而他们只管安心写业务脚本就可以，编写好的脚本提交 merge request，在 MR 被 merge 后脚本被自动并入 DAG。在该系统上线后，很大程度上提高了数仓分析人员的研发效率，但随着脚本数量的不断增加，新的问题也不断凸显，比如整个 DAG 的运行时间没有办法控制在一个可接受的时间范围内，在流利说，作为数据工程师，是需要对线上所有脚本的执行情况负责的。

任务治理

关键路径

缩短 DAG 的整体运行时间，一方面我们可以使用更快的计算引擎来提高任务的执行效率，比如可以把执行较久的 Hive 任务使用 SparkSQL 来改写(因为迭代次数较多，分析类的 SQL 特别适合使用 Spark 来跑)，但事实情况远不是换计算引擎这么简单，比如任务 t3 依赖 t2，t2 又依赖 t1，那么 t3 永远是要等它的前置任务跑完才可以跑，如果每个任务需要 20 分钟，那么整体完成时间就是 1 小时，串行依赖更糟糕的是，更多的集群资源对任务的效率提升有限，大量的集群资源反而会被浪费。

对于一个 DAG，我们可以把它当作一颗树，而叶子节点就是最后完成的那批任务，树的高度越高，那么叶子节点任务的开始执行时间就会越晚。因此我们要想尽早完成 DAG 的运行，就需要降低树的高度，让任务尽可能的并行运行，虽然依赖关系是由实际任务的输入输出决定的，但并不是所有的依赖都是合理的，基于这样的假设，我们拉取了任务在凌晨运行的并行情况(Y 轴表示正在运行的任务数量)：

图中可以清晰的看出整个 DAG 的执行在某一段时间内的并行度没有那么高，比如在 1:30 时刻并行的任务数量已经下降到了 40，并且在 02:25 时只有 25，显然这段时间是路径优化的关键，我们可以圈出这段时间中的任务，然后利用这些任务重新生成一个 sub-dag，求出该 sub-dag 中的最长路径就可以得出这段时间串行最严重的一批任务，从而反馈到数仓团队，由他们介入优化。

基于上述的实践优化，我们对数仓任务的依赖层数同样做了限制，大的原则是确保整个 DAG 的高度不会无限增长。

不可靠数据源

前面提到，数据工程师需要对任务的执行状况负责，并且需要确保核心任务的 SLA。但我们发现有些外部数据源本身就是不太可靠的，比如第三方的 API 接口就无法保证在某个时间点返回数据，又由于依赖的复杂性，使得一些核心任务的上游依赖了这些数据源。在 DAG 生成之后，我们对所有核心任务做了前置路径的检查，如果从核心任务出发能够到达不可靠数据源，会给出相应的警告。另外，我们也做了字段级的血缘依赖，用来辅助任务路径的分析，在数据的流转中，这些不可靠的数据源可能并不是核心任务的必要依赖，如果其前置任务的中间表依赖了不可靠数据源，那么就要跟据实际情况对中间层进行拆表。

任务分级

任务分级的目的是在极端情况下能够保证核心任务的输出，在流利说，我们的 ETL 集群会随着负载而自动伸缩集群的节点数量，但仍然要考虑到一些不可抗拒的外部因素，比如任务执行期间无法弹出更多的资源，或者一些数据源存在问题而短时间没有办法恢复等。此时就需要用现有仅存的资源，来尽量保证核心任务以及它们的上游任务的执行，但在 DAG 上又该如何操作呢？

假设现在有一个 DAG，如上图，核心任务被着成黑色，我们利用这两个核心任务，可以计算出它们的前置依赖，并同时认为这些前置依赖的任务也是核心任务：

如上图的 DAG 中所有黑色节点都被认为核心任务，此时可以利用这些核心任务生成一个单独的 core DAG，并把该 DAG 中的核心节点改成依赖 core DAG 的外部任务。此时 core DAG 如下图：

在正常情况下所有任务按照 DAG / core DAG 的正常编排顺序执行，而在非正常情况下，我们可以只执行 core DAG，从而达到优先保证核心任务的目的。

总结

本文简单介绍了流利说目前在任务自动化编排以及治理的各种手段，通过编排我们实现了任务的 CI/CD 流程，在任务执行时，我们还需要考虑个体任务的 SLA (任务不得早于预期执行时间、延时结束、超时、重试次数等等)，同时还需要建立合理的告警机制。而通过治理我们发现了任务之间关系的合理性，也极大的提高了整个 DAG 的执行效率，治理是一个长期的过程，除本文提到的这些手段以外，我们还需要深入任务中去解决影响性能的各种因素。

作者简介

董亚军  技术部数据工程团队 Tech Lead

戳“阅读原文” get 流利说工作机会噢