Airflow 快速学习入门-技术圈

作者：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_COMMAND

    Groups:
      celery         Celery components
      config         View configuration
      connections    Manage connections
      dags           Manage DAGs
      db             Database operations
      kubernetes     Tools to help run the KubernetesExecutor
      pools          Manage pools
      providers      Display providers
      roles          Manage roles
      tasks          Manage tasks
      users          Manage users
      variables      Manage variables

    Commands:
      cheat-sheet    Display cheat sheet
      info           Show information about current Airflow and environment
      kerberos       Start a kerberos ticket renewer
      plugins        Dump information about loaded plugins
      rotate-fernet-key
                     Rotate encrypted connection credentials and variables
      scheduler      Start a scheduler instance
      sync-perm      Update permissions for existing roles and DAGs
      version        Show the version
      webserver      Start a Airflow webserver instance

optional 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

点击左下角阅读原文，到 SegmentFault 思否社区 和文章作者展开更多互动和交流，扫描下方”二维码“或在“公众号后台“回复“ 入群 ”即可加入我们的技术交流群，收获更多的技术文章~

- END -