Airflow 架构及原理

一、Airflow 架构

Airflow 我们可以构建 Workflow 工作流，工作流使用 DAG 有向无环图来表示，DAG 指定了任务之间的关系，如下图：

Airflow 架构图如下：

Airflow 在运行时有很多守护进程，这些进程提供了 airflow 全部功能，守护进程包括如下：

• webserver：

WebServer 服务器可以接收 HTTP 请求，用于提供用户界面的操作窗口，主要负责中止、恢复、触发任务；监控任务；断点续跑任务；查询任务状态、详细日志等。

• Scheduler:

调度器，负责周期性调度处理工作流，并将工作流中的任务提交给 Executor 执行。

• Executor:

执行器，负责运行 task 任务，在默认本地模式下（单机 airflow）会运行在调度器 Scheduler 中并负责所有任务的处理。但是在 airflow 集群模式下的执行器 Executor 有很多类型，负责将任务 task 实例推送给 Workers 节点执行。

在 Airflow 中执行器有很多种选择，最关键的执行器有以下几种：

• SequentialExecutor：默认执行器，单进程顺序执行任务，通常只用于测试。

• LocalExecutor:多进程本地执行任务。

• CeleryExecutor：分布式执行任务，多用于生产场景，使用时需要配置消息队列。

• DaskExecutor:动态任务调度，支持远程集群执行 airflow 任务。

生产环境中建议使用 CeleryExecutor 作为执行器，Celery 是一个分布式调度框架，本身无队列功能，需要使用第三方插件，例如：RabbitMQ 或者 Redis。

关于不同 Executor 类型可以参考官网：https://airflow.apache.org/docs/apache-airflow/stable/executor/index.html

• work:

Worker 负责执行具体的 DAG 任务，会启动 1 个或者多个 Celery 任务队列，当 ariflow 的 Executor 设置为 CeleryExecutor 时才需要开启 Worker 进程。

• DAG Directory:

存放定义 DAG 任务的 Python 代码目录，代表一个 Airflow 的处理流程。需要保证 Scheduler 和 Executor 都能访问到。

• metadata database:

Airflow 的元数据库，用于 Webserver、Executor 及 Scheduler 存储各种状态数据，通常是 MySQL 或 PostgreSQL。

二、Airflow 术语

• DAG

DAG 是 Directed Acyclic Graph 有向无环图的简称，描述其描述数据流的计算过程。

• Operators

描述 DAG 中一个具体 task 要执行的任务，可以理解为 Airflow 中的一系列“算子”，底层对应 python class。不同的 Operator 实现了不同的功能，如：BashOperator 为执行一条 bash 命令，EmailOperator 用户发送邮件，HttpOperators 用户发送 HTTP 请求，PythonOperator 用于调用任意的 Python 函数。

• Task

Task 是 Operator 的一个实例，也就是 DAG 中的一个节点，在某个 Operator 的基础上指定具体的参数或者内容就形成一个 Task，DAG 中包含一个或者多个 Task。

• Task Instance

task 每一次运行对应一个 Task Instance，Task Instance 有自己的状态，例如：running,success,failed,skipped 等。

• Task Relationships：

一个 DAG 中可以有很多 task，这些 task 执行可以有依赖关系，例如：task1 执行后再执行 task2，表明 task2 依赖于 task1，这就是 task 之间的依赖关系。

三、Airflow 工作原理

airflow 中各个进程彼此之间是独立不互相依赖，也不互相感知，每个进程在运行时只处理分配到自身的任务，各个进程在一起运行，提供了 Airflow 全部功能，其工作原理如下：

1. 调度器 Scheduler 会间隔性轮询元数据库（Metastore）已注册的 DAG 有向无环图作业流，决定是否执行 DAG，如果一个 DAG 根据其调度计划需要执行，Scheduler 会调度当前 DAG 并触发 DAG 内部 task，这里的触发其实并不是真正的去执行任务，而是推送 task 消息到消息队列中，每一个 task 消息都包含此 task 的 DAG ID,Task ID 以及具体需要执行的函数，如果 task 执行的是 bash 脚本，那么 task 消息还会包含 bash 脚本代码。

2. Worker 进程将会监听消息队列，如果有消息就从消息队列中获取消息并执行 DAG 中的 task，如果成功将状态更新为成功，否则更新成失败。

3. 用户可以通过 webserver webui 来控制 DAG，比如手动触发一个 DAG 去执行，手动触发 DAG 与自动触发 DAG 执行过程都一样。