最近工作需要，使用airflow搭建了公司的ETL系統，順帶在公司分享了一次airflow，整理成文，Enjoy！

1. airflow 介紹

1.1 airflow 是什么

Airflow is a platform to programmatically author, schedule and monitor workflows.

airflow 是一個編排、調度和監控workflow的平臺，由Airbnb開源，現在在Apache Software Foundation 孵化。airflow 將workflow編排為tasks組成的DAGs，調度器在一組workers上按照指定的依賴關系執行tasks。同時，airflow 提供了豐富的命令行工具和簡單易用的用戶界面以便用戶查看和操作，并且airflow提供了監控和報警系統。

1.2 airflow 核心概念

DAGs：即有向無環圖(Directed Acyclic Graph)，將所有需要運行的tasks按照依賴關系組織起來，描述的是所有tasks執行的順序。

Operators：可以簡單理解為一個class，描述了DAG中一個具體的task具體要做的事。其中，airflow內置了很多operators，如BashOperator 執行一個bash 命令，PythonOperator 調用任意的Python 函數，EmailOperator 用于發送郵件，HTTPOperator 用于發送HTTP請求， SqlOperator 用于執行SQL命令...同時，用戶可以自定義Operator，這給用戶提供了極大的便利性。

Tasks：Task 是 Operator的一個實例，也就是DAGs中的一個node。

Task Instance：task的一次運行。task instance 有自己的狀態，包括"running", "success", "failed", "skipped", "up for retry"等。

Task Relationships：DAGs中的不同Tasks之間可以有依賴關系，如 TaskA >> TaskB，表明TaskB依賴于TaskA。

通過將DAGs和Operators結合起來，用戶就可以創建各種復雜的 workflow了。

1.3 其它概念

Connections: 管理外部系統的連接信息，如外部MySQL、HTTP服務等，連接信息包括conn_id／hostname ／ login ／ password／schema 等，可以通過界面查看和管理，編排workflow時，使用conn_id 進行使用。

Pools: 用來控制tasks執行的并行數。將一個task賦給一個指定的pool，并且指明priority_weight，可以干涉tasks的執行順序。

XComs：在airflow中，operator一般(not always)是原子的，也就是說，他們一般獨立執行，同時也不需要和其他operator共享信息，如果兩個operators需要共享信息，如filename之類的， 推薦將這兩個operators組合成一個operator。如果實在不能避免，則可以使用XComs (cross-communication)來實現。XComs用來在不同tasks之間交換信息。

Trigger Rules：指task的觸發條件。默認情況下是task的直接上游執行成功后開始執行，airflow允許更復雜的依賴設置，包括all_success(所有的父節點執行成功)，all_failed(所有父節點處于failed或upstream_failed狀態)，all_done(所有父節點執行完成)，one_failed(一旦有一個父節點執行失敗就觸發，不必等所有父節點執行完成)，one_success(一旦有一個父節點執行成功就觸發，不必等所有父節點執行完成)，dummy(依賴關系只是用來查看的，可以任意觸發)。另外，airflow提供了depends_on_past，設置為True時，只有上一次調度成功了，才可以觸發。

2. 示例

先來看一個簡單的DAG。圖中每個節點表示一個task，所有tasks組成一個DAG，各個tasks之間的依賴關系可以根據節點之間的線看出來。

DAGs

2.1 實例化DAG

# -*- coding: UTF-8 -*-## 導入airflow需要的modules from airflow import DAG from datetime import datetime, timedeltadefault_args = {'owner': 'lxwei','depends_on_past': False, # 如上文依賴關系所示'start_date': datetime(2018, 1, 17), # DAGs都有個參數start_date，表示調度器調度的起始時間'email': ['lxwei@github.com'], # 用于alert'email_on_failure': True,'email_on_retry': False,'retries': 3, # 重試策略'retry_delay': timedelta(minutes=5) }dag = DAG('example-dag', default_args=default_args, schedule_interval='0 0 * * *')

在創建DAGs時，我們可以顯示的給每個Task傳遞參數，但通過default_args，我們可以定義一個默認參數用于創建tasks。

注意，schedule_interval 跟官方文檔不一致，官方文檔的方式已經被deprecated。

2.2 定義依賴關系

這個依賴關系是我自己定義的，key表示某個taskId，value里的每個元素也表示一個taskId，其中，key依賴value里的所有task。

"dependencies": {"goods_sale_2": ["goods_sale_1"], # goods_sale_2 依賴 goods_sale1"shop_sale_1_2": ["shop_sale_1_1"],"shop_sale_2_2": ["shop_sale_2_1"],"shop_sale_2_3": ["shop_sale_2_2"],"etl_task": ["shop_info", "shop_sale_2_3", "shop_sale_realtime_1", "goods_sale_2", "shop_sale_1_2"],"goods_sale_1": ["timelySalesCheck", "productDaySalesCheck"],"shop_sale_1_1": ["timelySalesCheck", "productDaySalesCheck"],"shop_sale_realtime_1": ["timelySalesCheck", "productDaySalesCheck"],"shop_sale_2_1": ["timelySalesCheck", "productDaySalesCheck"],"shop_info": ["timelySalesCheck", "productDaySalesCheck"] }

2.3 定義tasks和依賴關系

首先，實例化operators，構造tasks。如代碼所示，其中，EtlTask、MySQLToWebDataTransfer、MySQLSelector 是自定義的三種Operator，根據taskType實例化operator，并存放到taskDict中，便于后期建立tasks之間的依賴關系。

for taskConf in tasksConfs:taskType = taskConf.get("taskType")if taskType == "etlTask":task = EtlTask(task_id=taskConf.get("taskId"),httpConnId=httpConn,etlId=taskConf.get("etlId"),dag=dag)taskDict[taskConf.get("taskId")] = taskelif taskType == "MySQLToWebDataTransfer":task = MySqlToWebdataTransfer(task_id = taskConf.get("taskId"),sql= taskConf.get("sql"),tableName=taskConf.get("tableName"),mysqlConnId =mysqlConn,httpConnId=httpConn,dag=dag)taskDict[taskConf.get("taskId")] = taskelif taskType == "MySQLSelect":task = StatusChecker(task_id = taskConf.get("taskId"),mysqlConnId = mysqlConn,sql = taskConf.get("sql"),dag = dag)taskDict[taskConf.get("taskId")] = taskelse:logging.error("error. TaskType is illegal.")

構建tasks之間的依賴關系，其中，dependencies中定義了上面的依賴關系，A >> B 表示A是B的父節點，相應的，A << B 表示A是B的子節點。

for sourceKey in dependencies:destTask = taskDict.get(sourceKey)sourceTaskKeys = dependencies.get(sourceKey)for key in sourceTaskKeys:sourceTask = taskDict.get(key)if (sourceTask != None and destTask != None):sourceTask >> destTask

3. 常用命令

命令行輸入airflow -h，得到幫助文檔

backfill Run subsections of a DAG for a specified date range list_tasks List the tasks within a DAG clear Clear a set of task instance, as if they never ran pause Pause a DAG unpause Resume a paused DAG trigger_dag Trigger a DAG run pool CRUD operations on pools variables CRUD operations on variables kerberos Start a kerberos ticket renewer render Render a task instance's template(s) run Run a single task instance initdb Initialize the metadata database list_dags List all the DAGs dag_state Get the status of a dag run task_failed_deps Returns the unmet dependencies for a task instancefrom the perspective of the scheduler. In other words,why a task instance doesn't get scheduled and thenqueued by the scheduler, and then run by an executor). task_state Get the status of a task instance serve_logs Serve logs generate by worker test Test a task instance. This will run a task withoutchecking for dependencies or recording it's state inthe database. webserver Start a Airflow webserver instance resetdb Burn down and rebuild the metadata database upgradedb Upgrade the metadata database to latest version scheduler Start a scheduler instance worker Start a Celery worker node flower Start a Celery Flower version Show the version connections List/Add/Delete connections

其中，使用較多的是backfill、run、test、webserver、scheduler。其他操作在web界面操作更方便。另外，initdb 用于初始化metadata，使用一次即可；resetdb會重置metadata，清除掉數據（如connection數據）, 需要慎用。

4. 問題

在使用airflow過程中，曾把DAGs里的task拆分得很細，這樣的話，如果某個task失敗，重跑的代價會比較低。但是，在實踐中發現，tasks太多時，airflow在調度tasks會很低效，airflow一直處于選擇待執行的task的過程中，會長時間沒有具體task在執行，從而整體執行效率大幅降低。

5. 總結

airflow 很好很強大。如果只是簡單的ETL之類的工作，可以很容易的編排。調度靈活，而且監控和報警系統完備，可以很方便的投入生產環節。

6. 參閱

airflow 官網
github

總結

以上是生活随笔為你收集整理的十八、可视化任务调度系统airflow的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。