Python 实现定时任务的八种方案( 六 ) _Python

Airflow使用Python开发，它通过DAGs(Directed Acyclic Graph, 有向无环图)来表达一个工作流中所要执行的任务，以及任务之间的关系和依赖。比如，如下的工作流中，任务T1执行完成，T2和T3才能开始执行，T2和T3都执行完成，T4才能开始执行。

文章插图

Airflow提供了各种Operator实现，可以完成各种任务实现：

BashOperator – 执行 bash 命令或脚本。
SSHOperator – 执行远程 bash 命令或脚本（原理同 paramiko 模块）。
PythonOperator – 执行 Python 函数。
EmailOperator – 发送 Email 。
HTTPOperator – 发送一个 HTTP 请求。
MySqlOperator, SqliteOperator, PostgresOperator, MsSqlOperator, OracleOperator, JdbcOperator, 等，执行 SQL 任务。
DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator…

除了以上这些 Operators 还可以方便的自定义 Operators 满足个性化的任务需求。
一些情况下，我们需要根据执行结果执行不同的任务，这样工作流会产生分支。如:

文章插图

这种需求可以使用BranchPythonOperator来实现。
Airflow 产生的背景通常，在一个运维系统，数据分析系统，或测试系统等大型系统中，我们会有各种各样的依赖需求。包括但不限于：

crontab 可以很好地处理定时执行任务的需求，但仅能管理时间上的依赖。Airflow 的核心概念 DAG（有向无环图）—— 来表现工作流。

Airflow 核心概念

DAGs：即有向无环图(Directed Acyclic Graph)，将所有需要运行的tasks按照依赖关系组织起来，描述的是所有tasks执行顺序。
Operators：可以简单理解为一个class，描述了DAG中某个的task具体要做的事。其中，airflow内置了很多operators，如BashOperator 执行一个bash 命令，PythonOperator 调用任意的Python 函数，EmailOperator 用于发送邮件，HTTPOperator 用于发送HTTP请求，SqlOperator 用于执行SQL命令等等，同时，用户可以自定义Operator，这给用户提供了极大的便利性。
Tasks：Task 是 Operator的一个实例，也就是DAGs中的一个node 。
Task Instance：task的一次运行。Web 界面中可以看到task instance 有自己的状态，包括”running”, “success”, “failed”, “skipped”, “up for retry”等。
Task Relationships：DAGs中的不同Tasks之间可以有依赖关系，如 Task1 >> Task2，表明Task2依赖于Task2了。通过将DAGs和Operators结合起来，用户就可以创建各种复杂的工作流（workflow）。

Airflow 的架构在一个可扩展的生产环境中，Airflow 含有以下组件：

元数据库：这个数据库存储有关任务状态的信息。
调度器：Scheduler 是一种使用 DAG 定义结合元数据中的任务状态来决定哪些任务需要被执行以及任务执行优先级的过程。调度器通常作为服务运行。
执行器：Executor 是一个消息队列进程，它被绑定到调度器中，用于确定实际执行每个任务计划的工作进程。有不同类型的执行器，每个执行器都使用一个指定工作进程的类来执行任务。例如，LocalExecutor 使用与调度器进程在同一台机器上运行的并行进程执行任务。其他像 CeleryExecutor 的执行器使用存在于独立的工作机器集群中的工作进程执行任务。
Workers：这些是实际执行任务逻辑的进程，由正在使用的执行器确定。