第一章 Airflow基本原理

标签：start   导出   rate   python   led   配置文件   数据处理   邮件   通过   

一、Airflow简介

airflow是Airbnb开源的一个用python编写的调度工具，项目于2014年启动，2015年春季开源，2016年加入Apache软件基金会的孵化计划，使用Python编写实现的任务管理、调度、监控工作流平台。
Airflow 是基于DAG(有向无环图)的任务管理系统，可以简单理解为是高级版的crontab，但是它解决了crontab无法解决的任务依赖问题。与crontab相比Airflow可以方便查看任务的执行状况（执行是否成功、执行时间、执行依 赖等），可追踪任务历史执行情况，任务执行失败时可以收到邮件通知，查看错误日志。

二、Airflow使用场景

在实际项目中，我们经常遇到以下场景：

1.运维人员，定时对服务器执行脚本某些脚本，最简单的方式是添加一些crond任务，但如果想追溯各个任务的执行结果时？
2.在大数据场景下，每隔一段时间需导出线上数据、导入到大数据平台、触发数据处理等多个子操作，且各个子操作含有依赖关系时？
3.在管理大量主机时，想要一个统一的作业管理平台，能在上面定义各种任务来管理下面的设备？

airflow通过DAG配置文件，能轻松定义各种任务及任务之间的依赖关系和调度执行，并一个可视化的操作web界面。

三、Airflow优势

#1.自带web管理界面，易上手；
#2.业务代码和调度代码完全解耦；
#3.通过python代码定义子任务，并支持各种Operate操作器，灵活性大，能满足用户的各种需求；
#4.python开源项目，支持扩展operate等插件，便于二次开发；
#5.类似的工具有akzban，quart等；

四、Airflow基本架构

在一个可扩展的生产环境中，Airflow 含有以下组件：
#1.元数据库：
	这个数据库存储有关任务状态的信息。

#2.调度器：
	Scheduler 是一种使用 DAG 定义结合元数据中的任务状态来决定哪些任务需要被执行以及任务执行优先级的过程。调度器通常作为服务运行。

#3.执行器：
	Executor 是一个消息队列进程，它被绑定到调度器中，用于确定实际执行每个任务计划的工作进程。有不同类型的执行器，每个执行器都使用一个指定工作进程的类来执行任务。
最关键的执行器就有四种选择:
SequentialExecutor：单进程顺序执行任务，默认执行器，通常只用于测试
LocalExecutor：多进程本地执行任务,使用与调度器进程在同一台机器上运行的并行进程执行任务
CeleryExecutor：分布式调度，生产常用,使用存在于独立的工作机器集群中的工作进程执行任务
DaskExecutor ：动态任务调度，主要用于数据分析

#4.Workers：
	这些是实际执行任务逻辑的进程，由正在使用的执行器确定。

五、Airflow 基本概念

1.DAG

DAG（Directed Acyclic Graph）是Airflow的核心概念之一，DAG体现的是你的工作流，它由Python脚本定义，其中包含了你想要运行的一系列task，同时其中还定义了这些task的依赖关系。 DAG代表了一个供调度的工作流，它的主要配置项包括owner，schedule。DAG支持多种调度方式，你可以指定该DAG定时调度，如每天的5am，也可以指定它周期性调度，如每二十分钟调度一次。 DAG由其中的task组成。例如，一个简单的DAG可以包括三个任务：A，B，C。我们可以让B依赖于A，在A成功执行之后运行B，而C可以在任意时候运行。需要注意的是，DAG本身并不关注A，B，C三个任务的具体内容，它关注的是三个任务的执行顺序以及依赖条件。

2.DAG Run

DAG run是指DAG的实例，DAG run通常由Airflow的scheduler创建，在特定时间运行，并包含了DAG中定义的task的实例。

3.execution_date

execution_date是logical datetime，是DAG指定运行的时间，它可以是过去或将来的某个时间尽管DAG实际上是正在运行的。

4.Task

Task是DAG中定义工作的基本单位，它的地位等同于工作流中的一个节点。Task和Operator是一枚硬币的正反两面，Task代表工作的抽象概念，Operator定义了对应Task要做的具体任务。 同一个DAG中的Task之间一般有先后顺序和依赖关系，考虑如下代码：
with DAG(‘my_dag‘, start_date=datetime(2016, 1, 1)) as dag:
    task_1 = DummyOperator(‘task_1‘)
    task_2 = DummyOperator(‘task_2‘)
    task_1 >> task_2 # Define dependencies
这段代码定义了一个包含两个task的DAG，task_1是task_2的上游，而task_2依赖于task_1，当DAG run被创建时，task_1首先运行，task_2在task_1成功执行后才会执行。

5.Task Instance

Task Instance是Task的实例，其关系等同于DAG和DAG run之间的关系。

6.Task的生命周期

Task的生命周期有一下八种：

Task的正常流程包括一下几个阶段

No status (scheduler 创建了空的task实例)
Scheduled (scheduler对该task实例进行了调度)
Queued (scheduler将task实例传给executor，放入执行队列)
Running (task开始执行)
Success (task成功结束)

7.Operators

DAG定义了一个工作流如何执行，而Operator定义了一个task执行的具体任务，是Airflow中编写具体任务的类。Operator包括很多种类，BashOperator用来执行Bash命令，PythonOperator可以执行Python函数，MySOperator可以操作MySQL数据库执行相关操作，当然你也可以从BaseOperator中继承并开发自己的Operator。

8.Scheduler

Scheduler监控所有的task和DAG，同时触发依赖已经满足的task。Scheduler在后端开启子进程，和DAG文件夹同步，并周期性（可配置时长）收集DAG的解析结果来找到可以条件满足的task。Scheduler会将可以运行的task交给我们配置好的executor执行。 Scheduler是Airflow环境中的顶层服务，简单地在命令行运行airflow scheduler可以开启。

9.Executor

Executor是task的执行器，它有多种配置方式，如SequetialExecutor串行运行task（适用于开发环境，是默认的配置），LocalExecutor可以在本地并发运行task，CeleryExecutor可以分布式地运行task。

六、Airflow常用命令

4.1，常用命令
$ airflow webserver -D		 守护进程运行webserver
$ airflow scheduler -D		 守护进程运行调度器
$ airflow worker -D		     守护进程运行调度器
$ airflow worker -c 1 -D 	 守护进程运行celery worker并指定任务并发数为1
$ airflow pause dag_id　 	暂停任务
$ airflow unpause dag_id 	 取消暂停，等同于在管理界面打开off按钮
$ airflow list_tasks dag_id  查看task列表
$ airflow clear dag_id       清空任务实例
$ airflow trigger_dag dag_id -r RUN_ID -e EXEC_DATE  运行整个dag文件
$ airflow run dag_id task_id execution_date       运行task

七、Airflow与同类产品的对比

第一章 Airflow基本原理

标签：start   导出   rate   python   led   配置文件   数据处理   邮件   通过   

原文地址：https://www.cnblogs.com/jhno1/p/14943297.html