最近在弄畫像標簽每天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': ['administer@testemail.com'], '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，加粗部分是關鍵字命令。