关于任务调度:任务调度系统系列之Airflow

这是任务调度零碎调研系列文章的开篇，后续会陆续调研 Oozie，Azkaban，Dolphin Scheduler 等零碎。本文的次要内容是来自对官网文档和网上相干材料的调研，并非基于理论应用的经验总结，文章中难免会有一些不尽的细节或者对于 Airflow 谬误的观点，如有不当之处，欢送斧正交换。

Airflow 是一个基于 Python 开发的，通过编程的形式创立、调度和监控工作流 (workflow) 的平台。最早由 Maxime Beauchemin 于 2014 年 10 月在 Airbnb 创立，并且从创立之初就以开源的模式开发。2016 年 3 月进入 Apache 基金孵化器，2019 年 1 月正式成为 Apache 顶级我的项目。

官网文档中，特意强调了应用代码定义工作流的长处，使得工作流的保护、版本治理、测试和合作变得更加容易，间接复用代码开发过程中用到工具、零碎就能够了，无需再反复造轮子，能够像开发软件系统一样开发数据工作，继续集成也是开箱即用。然而数据工作的测试向来不是一件简略的事件，不晓得在理论应用中基于 Airflow 的数据开发 CI/CD 晦涩度如何。这种基于 Python 代码定义工作流的形式应用门槛略微高了一点。基于代码定义 flow 中节点的依赖关系，并不如通过界面拖拽那么直观，是不是也会使易用性大大折扣？

如上图所示，Airflow 次要由以下几个局部组成：

存储定义 DAG 的 Python 文件的目录，调度器、执行器和执行节点会读取该目录下的文件获取 DAG 相干信息，所以要确保所有节点上 DAG 目录的数据同步。如何确保文件同步到也是一项简单的工程。

数据库次要用于存储系统的配置信息（零碎变量，数据源链接信息，用户信息等）、解析 DAG 文件生成的 DAG 对象和工作执行的状态等信息。

独立部署的过程，负责触发执行定时调度的工作流。调度器次要负责两件事：1）定时扫描 DAG 文件目录，解析变更或新增的 DAG 文件，并将解析后生成的 DAG 对象 (Serialized DAG) 存储到数据库；2）扫描数据库中的 DAG 对象，如果满足调度执行条件，则触发生成工作实例并提交给 Executor 执行。

从 2.0 开始，Airflow 调度器反对高可用部署，采纳了我之前实现调度服务高可用时应用的策略，通过数据库行锁的机制，实现多主的高可用。这样实现的益处是缩小了 leader 选举、节点故障转移的复杂度。多个节点同时工作相较于主从模式也能获取较好的解决性能，能够通过横向扩大调度器晋升调度服务的解决能力，但究竟要受限于底层单点数据库的解决能力。如果执行事务的时长比拟久，特地是事务中存在校验并发限度、资源应用配额的操作时，就很容易造成死锁，所以在 Airflow 理论部署中，高可用对数据库有着非凡的要求，须要数据库反对 SKIP LOCKED 或者NOWAIT。

执行器负责执行或者提交执行工作实例(Task Instances)。执行器在理论部署中集成在调度器过程内，Airflow 提供了两种类型的执行器，1）本地执行器，工作间接在调度器所在节点上执行；2）近程执行器，依据各执行器的实现形式，将工作提交到近程执行节点上执行。如果零碎自带的执行器无奈满足你的业务需要，能够自行实现自定义执行器。

零碎自带本地执行器：

Debug Executor：次要用于在 IDE 中对工作进行测试执行。
Local Executor：在调度器本地新建过程执行工作实例，能够通过 parallelism 参数管制最大工作并发数。
Sequential Executor：能够了解为最大并发数 1 的 Local Executor。

零碎自带近程执行器：

Celery Executor：Celery 是一个基于 Python 开发的分布式异步音讯工作队列，通过它能够轻松的实现工作的异步解决。Celery Executor 将工作发送到音讯队列 (RabbitMQ, Redis 等)，而后 Celery Worker 从音讯队列中生产执行工作，并将执行后果写入到 Celery 的 Backend 中。Celery Executor 通过队列(queues) 实现资源隔离，定义工作时指定应用的具体队列，则该工作只能由相应队列的 worker 执行。然而这个资源隔离的粒度有点粗，如果想实现更细粒度的资源，能够抉择 Kubernetes Executor。
Kubernetes Executor：通过 K8S 集群执行工作。Kubernetes Executor 调用 K8S API 申请 Worker Pod，而后由 Pod 负责工作的执行。
CeleryKubernetes Executor：是下面两个执行器的组合，因为 Airflow 部署时只能指定一种类型执行器，如果既须要通过 Celery 执行又想提交到 K8S 集群执行，则能够抉择该执行器。
Dask Executor: Dask 是基于 Python 实现的分布式计算框架，Dask Executor 次要是通过 Dask 分布式集群执行工作。

近程执行时所有执行节点都会间接数据库，鉴于弹性伸缩是 Airflow 的一大个性，如果执行节点规模太大对数据库造成的压力不可小觑，所以为什么要采纳执行节点直连数据的形式呢？

负责具体任务的执行，依据执行器不同，可能是调度器所在节点，Celery Worker 节点，K8S Pod 等。

WebServer 次要为用户提供了治理 DAG（启用、禁用，手动执行），查看和操作 DAG 的执行状态，管理系统权限，查看和批改系统配置，治理数据源等性能。前文提到的通过代码定义依赖关系不直观的问题，Airflow 在 WebServer 给了解决方案，运行 DAG，而后通过 WebServer 的 Graph 视图以可视化的形式展现 DAG。如果肯定要在执行前可视化的形式查看 DAG 也能够在命令行执行 airflow dags show 生成 Graph 视图的图片。兴许是我调研的还不够深刻，难道就没有实时可视化展现 DAG 的计划？

Airflow 通过 Python 代码以 DAG 的模式定义工作流，以下代码片段定义了上图由 7 个工作节点组成的 DAG。

// 从 2021 年 1 月 1 日开始，每天零点调度
with DAG("daily_dag", schedule_interval="@daily", start_date=datetime(2021, 1, 1)
) as dag:
    ingest = DummyOperator(task_id="ingest")
    analyse = DummyOperator(task_id="analyze")
    check = DummyOperator(task_id="check_integrity")
    describe = DummyOperator(task_id="describe_integrity")
    error = DummyOperator(task_id="email_error")
    save = DummyOperator(task_id="save")
    report = DummyOperator(task_id="report")

    ingest >> analyse // 通过 `>>`,`<<` 定义节点依赖关系
    analyse.set_downstream(check) // 通过 `set_downstream`,`set_upstream` 定义节点依赖关系
    check >> Label("No errors") >> save >> report // 通过 `Label` 正文依赖关系
    check >> Label("Errors found") >> describe >> error >> report

DAG 由节点、节点间的依赖关系以及节点间的数据流组成。节点的类型次要有以下三种：

Operator：工作节点，负责执行某种类型的工作。Airflow 和社区曾经实现了大量的 Operator，根本笼罩了罕用数据库，Hadoop 生态沉闷的零碎和服务，以及 AWS、Google 和 Azure 三大海内云平台的零碎和服务。
Sensor: 一种非凡的 Operator，次要用来监听内部事件，可用作对外部零碎、数据的依赖。Airflow 通过external_task Sensor 实现了 DAG 工作间的依赖。
@task注解的 Python 函数，能够了解为基于 Python 装璜器定义的语法糖，能疾速简洁的定义PythonOperator。

如以上 Python 代码所示，节点间的依赖关系能够通过位操作符 >>/<< 或set_upstream/set_downstream办法定义。

默认状况下上游节点要等上游所有节点执行胜利后才开始执行，Airflow 提供了多种形式来扭转这一默认行为。第一种形式就是通过自定义节点的触发规定 (Trigger Rules)。Airflow 提供了上游 所有节点都失败 、 所有节点执行实现 、 局部节点失败 、 局部节点胜利 等多种规定，详情参考上述链接。另一种形式就是通过 管制节点。目前有三种管制节点能够扭转默认行为：

分支节点 (Branching): 通过python_callable 函数返回的 task_id 决定执行上游哪个节点。
仅执行最新节点 (Latest Only): 如果 仅执行最新节点 以后所属的 DAG 执行实例，不是改 DAG 最新的执行实例，则改节点及其所有子节点都不会被执行。
自依赖节点(Depends On Past): 依赖节点的上一次执行，只有上一次 DAG 调度中该节点执行胜利了，才触发这一次执行。

除了定义节点间的依赖关系，Airflow 还通过 XComs(cross-communications)实现了节点间的数据流。节点能够通过 xcom_push 办法输入数据，其余节点能够通过 xcom_pull 办法获取节点的输入数据。

定时调度是通过 DAG 的 schedule_interval 参数定义，传参能够是 datetime.timedelta 对象、cron 表达式（Unix 格局）或者 @daily、@monthly 等预设 cron 表达式。对于任务调度工夫的定义，Airflow 采纳了目前我所接触到的任务调度零碎中不同的视角，用数据工夫（Airflow 称它为 logical date，有的零碎称它为 etl_date）来定义任务调度工夫。举个例子，如果 DAG 配置每天调度一次，在 Airflow 中 2021-12-26 这次的调度实例，要在 2021-12-27 这天凌晨才会生成，解决的是 2021-12-26 的数据。而在其余零碎中 2021-12-26 这次的调度实例就是在 2021-12-26 生成，解决的是 2021-12-25 的数据。

针对 MissFire 策略（概念来自 quartz），Airflow 提供了 catchup 参数。如果 catchup 设置为 false，则未生成的调度时间段间接跳过，只生成最新的调度实例。另外在禁用和启用调度 DAG 后catchup 逻辑也会触发。

如果要限度节点最大执行工夫，能够设置 execution_timeout 参数，节点在 execution_timeout 配置工夫内未执行胜利则主动超时失败。工作执行超时报警是通过 sla 参数配置的，节点在 sla 指定的工夫内没有执行胜利，零碎主动发送 SLA 未满足邮件，也能够通过 sla_miss_callback 回调函数，自定义工作超时的逻辑。

对于报警，Airflow 提供了 email_on_failure，email_on_retry 参数管制节点在执行失败、重试时是否发送邮件报警。在理论生产环境中，邮件报警必定是不能满足需要的，其余报警形式能够通过自定义 on_failure_callback,on_retry_callback 回调函数实现。

Airflow 提供了多种粒度的并发限度。

零碎级别

parallelism: Airflow 并发执行的工作数
max_active_runs_per_dag: 每个 DAG 可并产生成的 DAG 调度实例数
dag_concurrency: 每个 DAG 实例并发执行的工作数
worker_concurrency: 每个执行节点可并发执行的工作数，仅 Celery Executor 的执行节点反对该配置

DAG 级别

max_active_runs: 以后 DAG 可并产生成的 DAG 调度实例数，该配置会笼罩零碎级别的max_active_runs_per_dag
concurrency: 以后 DAG 实例并发执行的工作数，该配置会笼罩零碎级别的dag_concurrency

工作级别

pool：pool 是 Airflow 用于实现跨 DAG、跨工作的并发限度计划。定义工作时指定工作所属 pool、工作应用的 slot 数、工作优先级；pool 资源应用达到下限后，所有附属该 pool 的工作实例进入排队状态，有闲暇资源开释时，高优先工作优先获取资源。资源分配的具体策略这里就没有在深入研究。
task_concurrency：当前任务节点的最大并发执行实例个数，相似于max_active_runs，只是粒度更细

补数据（backfill）能够通过以下命令触发：


airflow dags backfill --start-date START_DATE --end-date END_DATE dag_id

默认状况下，补数据只生成并执行指定工夫范畴内缺失的调度记录。再举个例子，DAG 每天调度一次，当初要补 2021-12-01 到 2021-12-03 之间的数据，其中 2021-12-02 这天曾经调度执行过，则补数据工作只会创立执行 2021-12-01 和 20210-12-03 的调度记录。backfill 命令提供了多种选项来笼罩这一默认策略。

手动执行能够通过命令 airflow dags trigger --exec-date logical_date run_id 或者通过 WebServer 触发。

本着让业余的人干业余的事的理念，Airflow 依靠于第三方元数据管理系统实现数据血统治理，平台自身只实现血统的收集和上报。通过工作的 inlets 和outlets属性定义工作的血统信息，血统信息在工作的 post_execute 办法中推送到 XCOM，而后再由 LineageBackend 把血统信息写到 Atlas、DataHub（WhereHows）或者自定义的元数据管理系统。

本文通过官网文档和网上相干材料，“纸上”动态地调研了 Airflow 的零碎架构和性能个性。整体而言，Airflow 是一个调度功能完善、扩大伸缩性良好、文档详尽、社区弱小沉闷的工作流调度平台。个人感觉在任务调度零碎选型上，可能妨碍 Airflow 入选的最次要因素是基于 Python 技术栈实现的整个零碎和 DAG 定义。如果负责平台的同学和零碎面向的用户有 Python 相干技术背景，从纸面上看，Airflow 是个十分不错、甚至是第一优先级的抉择。

本文最早公布于集体博客

关于任务调度:任务调度系统系列之Airflow

架构

DAG 目录(DAG Directory)

数据库（Metadata Database）

调度器（Scheduler）

调度器高可用

执行器（Executor）

执行节点（worker）

WebServer

性能个性

工作流定义

调度

超时失败和报警

并发限度

零碎级别

DAG 级别

工作级别

补数据和手动执行

数据血统

总结