一个开源的ETL工具Airflow

最近在弄画像标签每天ETL的调度事情，这篇文章分享一下一个开源的ETL工具Airflow。

一、基础概念

Airflow是Airbnb内部发起并开源的一个ETL管理平台，使用Python编写实现的任务管理、调度、监控工作流平台。这是其官方文档地址：Apache Airflow (incubating) Documentation，关于airflow产品的使用，里面有详细的介绍。

Airflow的调度依赖于crontab命令，与crontab相比airflow可以直观的看到任务执行情况、任务之间的逻辑依赖关系、可以设定任务出错时邮件提醒、可以查看任务执行日志。

而crontab命令管理的方式存在以下几方面的弊端：

1、在多任务调度执行的情况下，难以理清任务之间的依赖关系；

2、不便于查看当前执行到哪一个任务；

3、任务执行失败时不便于查看执行日志，也即不方便定位报错的任务和错误原因；

4、不便于查看调度流下每个任务执行的起止消耗时间，这对于优化task作业是非常重要的；

5、不便于记录历史调度任务的执行情况，而这对于优化作业和错误排查是很重要的；

Airflow中有两个最基本的概念：DAG和task，下面主要介绍一下。

DAG是什么：

DAG是Directed Acyclic Graph的缩写，即有向无环图。是所有要执行任务脚本(即task)的集合，在这个DAG中定义了各个task的依赖关系、调度时间、失败重启机制等。通过DAGid来标识每个DAG任务

每个DAG是由1到多个task组成

task是什么：

task是具体执行的任务脚本，可以是一个命令行(BashOperator)，也可以是python脚本等。

二、主要功能键介绍

1、DAG管理

在airflow的主页，可以看到当前所有的DAG列表（通俗点说就是所有的调度任务列表），中间“Task by State”那一列显示任务的执行状态。深绿色的表示已执行成功的task，浅绿色的表示当前正在执行的task。

右侧“Links”那一列可以链接查看当前DAG任务的依赖关系、执行时间、执行脚本等情况。

当点击具体某一个DAG任务时，就可以进去查看该DAG的调度依赖、执行时长、调度脚本等具体执行情况

2、调度依赖查看

通过“Graph View”选项可以查看当前调度任务的依赖关系，当调度作业较为复杂时，这种图形化方式展示的依赖关系可以帮助用户迅速理清。

在用户画像的调度管理中，每天需要执行cookieid和userid两个维度的画像脚本，因此可以设定并行执行任务，让cookieid和userid的脚本同时执行调度作业

3、执行状态

通过“Tree View”选项可以查看当前任务的执行状态，包括当前执行到哪一个task，还有哪些task未执行。哪些task执行成功，哪些task执行失败。

也可以查看历史上该DAG下面各task的执行情况。

4、各task执行时间

通过“Gantt”选项可以查看各task任务的执行起止时间的甘特图。

了解各task执行的时间可以有针对性地优化执行时间长的task对应脚本。

5、DAG调度脚本

通过“Code”选项，可以查看当前DAG调度的脚本。脚本里面定义了需要执行的task、执行顺序及依赖、调度时间、失败发送邮件或重调机制等方法

三、脚本实例

在开发过程中，task脚本是需要被调度的脚本，在Airflow中主要需要开发的是DAG脚本，即管理task任务的脚本。通过一个DAG脚本，将各个调度作业脚本串起来，按照业务逻辑去执行。

1、DAG脚本

下面通过一个具体DAG脚本实例来了解一下：

from airflow.operators.bash_operator import BashOperator import airflow from airflow.models import DAG from airflow import operators from airflow.contrib.hooks import SSHHook from airflow.models import BaseOperator from airflow.contrib.operators import SSHExecuteOperator from airflow.operators.latest_only_operator import LatestOnlyOperator import os import sys from datetime import timedelta,date,datetime import pendulum from airflow.utils.trigger_rule import TriggerRule default_args = { 'owner': 'superuserprofile', 'depends_on_past': False, 'start_date': datetime(2018, 06, 01), 'email': ['[e-mail]'], 'email_on_failure': True , 'email_on_retry': True, 'retries': 1, 'retry_delay': timedelta(minutes=1), } os.environ['SPARK_HOME'] = '/usr/local/spark-2.1.1-bin-hadoop2.6' sys.path.append(os.path.join(os.environ['SPARK_HOME'], 'bin'))

该段脚本定义了需要引入的包，以及默认的DAG参数配置，包括task是否依赖上游任务，首次调度时间、任务失败接收邮箱、任务失败是否重新调起等

dag = DAG( 'superuserprofile', default_args=default_args, description='A userprofile test', schedule_interval='00 08 * * *' )

该段脚本实例化了DAG，设置了DAGid，调度执行时间

gender_task = BashOperator( task_id='gender', bash_command=' sudo -E -H -u userprofile spark-submit --master yarn --deploy-mode client --driver-memory 1g --executor-memory 8g --executor-cores 2 --num-executors 200 /airflow/userprofile_gender.py {{ ds_nodash }} ', dag=dag, trigger_rule=TriggerRule.ALL_DONE ) country_task = BashOperator( task_id='country', bash_command=' sudo -E -H -u userprofile spark-submit --master yarn --deploy-mode client --driver-memory 1g --executor-memory 4g --executor-cores 2 --num-executors 200 /airflow/userprofile_country.py {{ ds_nodash }} ', dag=dag, trigger_rule=TriggerRule.ALL_DONE )

该段脚本设置了两个需要执行的task任务(userprofile_gender.py和userprofile_country.py)的实例化。

task直接的调度依赖关系可以通过set_upstream、set_downstream命令或符号>> 、<<来建立。

gender_task .set_upstream(country_task) 命令指gender_task 任务将依赖country_task任务；反之同理

gender_task >> country_task 命令指country_task 任务将依赖gender_task 任务先执行完，反之同理

2、命令行执行

Airflow通过可视化界面的方式实现了调度管理的界面操作，但在测试脚本或界面操作失败的时候，可通过命令行的方式调起任务。下面介绍几个常用命令

命令1:airflow list_tasksuserprofile

该命令用于查看当前DAG任务下的所有task的列表

其中userprofile是DAGid，加粗的airflow list_tasks是关键字命令

-----------------------------------------------------------------------

命令2:airflow testuserprofile gender_task 20180601

该命令用于单独执行DAG下面的某个task

其中userprofile是DAGid，gender_task是要具体某个taskid，20180601是执行日期。加粗部分是关键字命令

-----------------------------------------------------------------------

命令3：airflow backfill -s2018-06-01-e2018-06-02 userprofile

该命令用于调起整个DAG脚本执行

其中2018-06-01是执行脚本的开始日期， 2018-06-02是结束日期，userprofile是DAGid，加粗部分是关键字命令。