作者：Corwien
来源：SegmentFault 思否社区
一、Airflow简介
Airflow 是一个使用 Python 语言编写的 Data Pipeline 调度和监控工作流的平台。
Airflow 是通过 DAG（Directed acyclic graph 有向无环图）来管理任务流程的任务调度工具，不需要知道业务数据的具体内容，设置任务的依赖关系即可实现任务调度。
这个平台拥有和 Hive、Presto、MySQL、HDFS、Postgres 等数据源之间交互的能力，并且提供了钩子（hook）使其拥有很好地扩展性。除了使用命令行，该工具还提供了一个 WebUI 可以可视化的查看依赖关系、监控进度、触发任务等。
Airflow 的架构
在一个可扩展的生产环境中，Airflow 含有以下组件：
元数据库：这个数据库存储有关任务状态的信息。
调度器：Scheduler 是一种使用 DAG 定义结合元数据中的任务状态来决定哪些任务需要被执行以及任务执行优先级的过程。调度器通常作为服务运行。
执行器：Executor 是一个消息队列进程，它被绑定到调度器中，用于确定实际执行每个任务计划的工作进程。有不同类型的执行器，每个执行器都使用一个指定工作进程的类来执行任务。例如，LocalExecutor 使用与调度器进程在同一台机器上运行的并行进程执行任务。其他像 CeleryExecutor 的执行器使用存在于独立的工作机器集群中的工作进程执行任务。
Workers：这些是实际执行任务逻辑的进程，由正在使用的执行器确定。
Airflow 解决哪些问题
通常，在一个运维系统，数据分析系统，或测试系统等大型系统中，我们会有各种各样的依赖需求。包括但不限于：
时间依赖：任务需要等待某一个时间点触发。
外部系统依赖：任务依赖外部系统需要调用接口去访问。
任务间依赖：任务 A 需要在任务 B 完成后启动，两个任务互相间会产生影响。
资源环境依赖：任务消耗资源非常多， 或者只能在特定的机器上执行。
crontab 可以很好地处理定时执行任务的需求，但仅能管理时间上的依赖。
Airflow 是一种 WMS，即：它将任务以及它们的依赖看作代码，按照那些计划规范任务执行，并在实际工作进程之间分发需执行的任务。
Airflow 提供了一个用于显示当前活动任务和过去任务状态的优秀 UI，并允许用户手动管理任务的执行和状态。
Airflow 中的工作流是具有方向性依赖的任务集合。
具体说就是 Airflow 的核心概念 DAG（有向无环图）—— 来表现工作流。
DAG 中的每个节点都是一个任务，DAG 中的边表示的是任务之间的依赖（强制为有向无环，因此不会出现循环依赖，从而导致无限执行循环）。
Airflow 在 ETL 上的实践
ETL，是英文 Extract，Transform，Load 的缩写，用来描述将数据从来源端经过抽取（extract）、转换（transform）、加载（load）至目的端的过程。ETL 一词较常用在数据仓库，Airflow 在解决 ETL 任务各种依赖问题上的能力恰恰是我们所需要的。
在现阶段的实践中，我们使用 Airflow 来同步各个数据源数据到数仓，同时定时执行一些批处理任务及带有数据依赖、资源依赖关系的计算脚本。
本文立意于科普介绍，故在后面的用例中只介绍了 BashOperator，PythonOperator这俩个最为易用且在我们日常使用中最为常见的 Operator。
Airflow 同时也具有不错的集群扩展能力，可使用 CeleryExecuter 以及多个 Pool 来提高任务并发度。
Airflow在 CeleryExecuter 下可以使用不同的用户启动 Worker，不同的 Worker 监听不同的 Queue，这样可以解决用户权限依赖问题。Worker 也可以启动在多个不同的机器上，解决机器依赖的问题。
Airflow 可以为任意一个 Task 指定一个抽象的 Pool，每个 Pool 可以指定一个 Slot 数。每当一个 Task 启动时，就占用一个 Slot，当 Slot 数占满时，其余的任务就处于等待状态。这样就解决了资源依赖问题。
二、安装及使用
假设：你已经安装好了 Python 及配置好了其包管理工具 pip。
1、安装airflow
pip install apache-airflow
在安装airflow的时候可能会报错：
Cannot uninstall 'PyYAML'. It is a distutils installed project and thus we cannot
忽略掉 PyYAML
# 亲测可用
pip install apache-airflow --ignore-installed PyYAML
安装成功后查看命令：
[root@quant ~]# airflow -h
usage: airflow [-h] GROUP_OR_COMMAND ...positional arguments:GROUP_OR_COMMANDGroups:celery Celery componentsconfig View configurationconnections Manage connectionsdags Manage DAGsdb Database operationskubernetes Tools to help run the KubernetesExecutorpools Manage poolsproviders Display providersroles Manage rolestasks Manage tasksusers Manage usersvariables Manage variablesCommands:cheat-sheet Display cheat sheetinfo Show information about current Airflow and environmentkerberos Start a kerberos ticket renewerplugins Dump information about loaded pluginsrotate-fernet-keyRotate encrypted connection credentials and variablesscheduler Start a scheduler instancesync-perm Update permissions for existing roles and DAGsversion Show the versionwebserver Start a Airflow webserver instanceoptional arguments:-h, --help show this help message and exit
[root@quant ~]#
2、初始化数据库
# initialize the database
airflow db init
报这样的错误：
ImportError: Something is wrong with the numpy installation. While importing we detected an older version of numpy
解决方案：
如报错信息所说
先卸载numpy：pip uninstall numpy
再卸载numpy，直到卸载到提示信息显示，此时完全已经没有numpy了为止
下载numpy：pip install numpy
此时应该可用；
若不可用，查看python安装目录下的libs文件夹，删除掉其中的另一个dll文件，应该可用。
3、添加用户
airflow users create \--username admin \--firstname Corwien \--lastname Wong \--role Admin \--email 407544577@qq.com
创建的用户密码为：quant
4、启动web服务
# start the web server, default port is 8080
airflow webserver --port 8080a
5、启动定时任务
# start the scheduler
# open a new terminal or else run webserver with ``-D`` option to run it as a daemon
airflow scheduler# visit localhost:8080 in the browser and use the admin account you just
# created to login. Enable the example_bash_operator dag in the home page
