一个开源的ETL工具Airflow.doc
一个开源的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: administertestemail, email_on_failure: True , email_on_retry: True, retries: 1, retry_delay: timedelta(minutes=1), os.environSPARK_HOME = /usr/local/spark-2.1.1-bin-hadoop2.6 sys.path.append(os.path.join(os.environSPARK_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命令或符号>> 、> country_task 命令指country_task 任务将依赖gender_task 任务先执行完，反之同理2、命令行执行Airflow通过可视化界面的方式实现了调度管理的界面操作，但在测试脚本或界面操作失败的时候，可通过命令行的方式调起任务。下面介绍几个常用命令命令1: airflow list_tasks userprofile该命令用于查看当前DAG任务下的所有task的列表其中userprofile是DAGid，加粗的airflow list_tasks是关键字命令-命令2: airflow test userprofile gender_task 20180601该命令用于单独执行DAG下面的某个task其中userprofile是DAGid，gender_task是要具体某个taskid，20180601是执行日期。加粗部分是关键字命令-命令3：airflow backfill -s 2018-06-01 -e 2018-06-02 userprofile该命令用于调起整个DAG脚本执行其中2018-06-01是执行脚本的开始日期， 2018-06-02是结束日期，userprofile是DAGid，加粗部分是关键字命令。
