一個(gè)開(kāi)源的ETL工具Airflow

最近在弄畫(huà)像標(biāo)簽每天ETL的調(diào)度事情，這篇文章分享一下一個(gè)開(kāi)源的ETL工具Airflow。

一、基礎(chǔ)概念

Airflow是Airbnb內(nèi)部發(fā)起并開(kāi)源的一個(gè)ETL管理平臺(tái)，使用Python編寫(xiě)實(shí)現(xiàn)的任務(wù)管理、調(diào)度、監(jiān)控工作流平臺(tái)。這是其官方文檔地址：Apache Airflow (incubating) Documentation，關(guān)于airflow產(chǎn)品的使用，里面有詳細(xì)的介紹。

Airflow的調(diào)度依賴于crontab命令，與crontab相比airflow可以直觀的看到任務(wù)執(zhí)行情況、任務(wù)之間的邏輯依賴關(guān)系、可以設(shè)定任務(wù)出錯(cuò)時(shí)郵件提醒、可以查看任務(wù)執(zhí)行日志。

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

1、在多任務(wù)調(diào)度執(zhí)行的情況下，難以理清任務(wù)之間的依賴關(guān)系；

2、不便于查看當(dāng)前執(zhí)行到哪一個(gè)任務(wù)；

3、任務(wù)執(zhí)行失敗時(shí)不便于查看執(zhí)行日志，也即不方便定位報(bào)錯(cuò)的任務(wù)和錯(cuò)誤原因；

4、不便于查看調(diào)度流下每個(gè)任務(wù)執(zhí)行的起止消耗時(shí)間，這對(duì)于優(yōu)化task作業(yè)是非常重要的；

5、不便于記錄歷史調(diào)度任務(wù)的執(zhí)行情況，而這對(duì)于優(yōu)化作業(yè)和錯(cuò)誤排查是很重要的；

Airflow中有兩個(gè)最基本的概念：DAG和task，下面主要介紹一下。

DAG是什么：

DAG是Directed Acyclic Graph的縮寫(xiě)，即有向無(wú)環(huán)圖。是所有要執(zhí)行任務(wù)腳本(即task)的集合，在這個(gè)DAG中定義了各個(gè)task的依賴關(guān)系、調(diào)度時(shí)間、失敗重啟機(jī)制等。通過(guò)DAGid來(lái)標(biāo)識(shí)每個(gè)DAG任務(wù)

每個(gè)DAG是由1到多個(gè)task組成

task是什么：

task是具體執(zhí)行的任務(wù)腳本，可以是一個(gè)命令行(BashOperator)，也可以是python腳本等。

二、主要功能鍵介紹

1、DAG管理

在airflow的主頁(yè)，可以看到當(dāng)前所有的DAG列表（通俗點(diǎn)說(shuō)就是所有的調(diào)度任務(wù)列表），中間“Task by State”那一列顯示任務(wù)的執(zhí)行狀態(tài)。深綠色的表示已執(zhí)行成功的task，淺綠色的表示當(dāng)前正在執(zhí)行的task。

右側(cè)“Links”那一列可以鏈接查看當(dāng)前DAG任務(wù)的依賴關(guān)系、執(zhí)行時(shí)間、執(zhí)行腳本等情況。

當(dāng)點(diǎn)擊具體某一個(gè)DAG任務(wù)時(shí)，就可以進(jìn)去查看該DAG的調(diào)度依賴、執(zhí)行時(shí)長(zhǎng)、調(diào)度腳本等具體執(zhí)行情況

2、調(diào)度依賴查看

通過(guò)“Graph View”選項(xiàng)可以查看當(dāng)前調(diào)度任務(wù)的依賴關(guān)系，當(dāng)調(diào)度作業(yè)較為復(fù)雜時(shí)，這種圖形化方式展示的依賴關(guān)系可以幫助用戶迅速理清。

在用戶畫(huà)像的調(diào)度管理中，每天需要執(zhí)行cookieid和userid兩個(gè)維度的畫(huà)像腳本，因此可以設(shè)定并行執(zhí)行任務(wù)，讓cookieid和userid的腳本同時(shí)執(zhí)行調(diào)度作業(yè)

3、執(zhí)行狀態(tài)

通過(guò)“Tree View”選項(xiàng)可以查看當(dāng)前任務(wù)的執(zhí)行狀態(tài)，包括當(dāng)前執(zhí)行到哪一個(gè)task，還有哪些task未執(zhí)行。哪些task執(zhí)行成功，哪些task執(zhí)行失敗。

也可以查看歷史上該DAG下面各task的執(zhí)行情況。

4、各task執(zhí)行時(shí)間

通過(guò)“Gantt”選項(xiàng)可以查看各task任務(wù)的執(zhí)行起止時(shí)間的甘特圖。

了解各task執(zhí)行的時(shí)間可以有針對(duì)性地優(yōu)化執(zhí)行時(shí)間長(zhǎng)的task對(duì)應(yīng)腳本。

5、DAG調(diào)度腳本

通過(guò)“Code”選項(xiàng)，可以查看當(dāng)前DAG調(diào)度的腳本。腳本里面定義了需要執(zhí)行的task、執(zhí)行順序及依賴、調(diào)度時(shí)間、失敗發(fā)送郵件或重調(diào)機(jī)制等方法

三、腳本實(shí)例

在開(kāi)發(fā)過(guò)程中，task腳本是需要被調(diào)度的腳本，在Airflow中主要需要開(kāi)發(fā)的是DAG腳本，即管理task任務(wù)的腳本。通過(guò)一個(gè)DAG腳本，將各個(gè)調(diào)度作業(yè)腳本串起來(lái)，按照業(yè)務(wù)邏輯去執(zhí)行。

1、DAG腳本

下面通過(guò)一個(gè)具體DAG腳本實(shí)例來(lái)了解一下：

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'))

該段腳本定義了需要引入的包，以及默認(rèn)的DAG參數(shù)配置，包括task是否依賴上游任務(wù)，首次調(diào)度時(shí)間、任務(wù)失敗接收郵箱、任務(wù)失敗是否重新調(diào)起等

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

該段腳本實(shí)例化了DAG，設(shè)置了DAGid，調(diào)度執(zhí)行時(shí)間

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 )

該段腳本設(shè)置了兩個(gè)需要執(zhí)行的task任務(wù)(userprofile_gender.py和userprofile_country.py)的實(shí)例化。

task直接的調(diào)度依賴關(guān)系可以通過(guò)set_upstream、set_downstream命令或符號(hào)>> 、<<來(lái)建立。

gender_task .set_upstream(country_task) 命令指gender_task 任務(wù)將依賴country_task任務(wù)；反之同理

gender_task >> country_task 命令指country_task 任務(wù)將依賴gender_task 任務(wù)先執(zhí)行完，反之同理

2、命令行執(zhí)行

Airflow通過(guò)可視化界面的方式實(shí)現(xiàn)了調(diào)度管理的界面操作，但在測(cè)試腳本或界面操作失敗的時(shí)候，可通過(guò)命令行的方式調(diào)起任務(wù)。下面介紹幾個(gè)常用命令

命令1:airflow list_tasksuserprofile

該命令用于查看當(dāng)前DAG任務(wù)下的所有task的列表

其中userprofile是DAGid，加粗的airflow list_tasks是關(guān)鍵字命令

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

命令2:airflow testuserprofile gender_task 20180601

該命令用于單獨(dú)執(zhí)行DAG下面的某個(gè)task

其中userprofile是DAGid，gender_task是要具體某個(gè)taskid，20180601是執(zhí)行日期。加粗部分是關(guān)鍵字命令

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

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

該命令用于調(diào)起整個(gè)DAG腳本執(zhí)行

其中2018-06-01是執(zhí)行腳本的開(kāi)始日期， 2018-06-02是結(jié)束日期，userprofile是DAGid，加粗部分是關(guān)鍵字命令。