Airflow(一個Airbnb的開源項目)

Airflow 是一個 Airbnb 的 Workflow 開源項目，在Github 上已經有超過兩千星。

基本介紹

• 軟體名稱：Airflow
• 開發商：Airbnb
• 軟體版本：1.6.1
• 軟體語言：Python 2/3

概述

Airflow 使用 Python 寫的，支持 Python 2/3 兩個版本。 傳統 Workflow 通常使用 Text Files (json, xml / etc) 來定義 DAG, 然後 Scheduler 解析這些 DAG 檔案形成具體的 Task Object 執行；Airflow 沒這么乾，它直接用 Python 寫 DAG definition, 一下子突破了文本檔案表達能力的局限，定義 DAG 變得簡單。 另外，Airflow 的許可權設計、限流設計、以及 Hook/Plugin 的設計都挺有意思，功能性、擴展性良好。當然，項目里的代碼質量感覺比較一般，很多地方函式名和實現不太一致，造成理解障礙；也有很多 Flag 和重複出現的定義，顯然是當初沒有設計好、後面沒有精力 Refactor 轉而 Hack 的結果。但總體上，可讀性中上，系統的擴展性非常好。 但我們想說的是，Airflow 真的是一個可以拿來即用、而且相當好用的東西。坊間傳聞說，Airflow 作者當初在 FB 的時候搞過非常類似的系統，跳槽之後，可能覺得重來一遍沒啥意思，順手開源。於是社區歡欣鼓舞，“不花錢，還能用上比 FB 裡面還好用的 Workflow System, 真是 Why Not 啊”.

Airflow 中的技巧和陷阱

Airbnb的數據工程師 Maxime Beauchemin 激動地表示道：Airflow 是一個我們正在用的工作流調度器，現在的版本已經更新到1.6.1了，並且引入了一些列調度引擎的改革。我們喜歡它是因為它寫代碼太容易了，也便於調試和維護。我們也喜歡全都用他來寫代碼，而不是像xml那樣的配置檔案用來描述DAG。更不用說，我們顯然不用再學習太多東西。

在一個分散式環境中，宕機是時有發生的。Airflow通過自動重啟任務來適應這一變化。到目前為止一切安好。當我們有一系列你想去重置狀態的任務時，你就會發現這個功能簡直是救世主。為了解決這個問題，我們的策略是建立子DAG。這個子DAG任務將自動重試自己的那一部分，因此，如果你以子DAG設定任務為永不重試，那么憑藉子DAG操作你就可以得到整個DAG成敗的結果。如果這個重置是DAG的第一個任務設定子DAG的策略就會非常有效，對於有一個相對複雜的依賴關係結構設定子DAG是非常棒的做法。注意到子DAG操作任務不會正確地標記失敗任務，除非你從GitHub用了最新版本的Airflow。解決這個問題的另外一個策略是使用重試柄：

def make_spooq_exporter(table, schema, task_id, dag):     return SpooqExportOperator(        jdbc_url=('jdbc:mysql://%s/%s?user=user&password=pasta'                    % (TARGET_DB_HOST,TARGET_DB_NAME)),        target_table=table,        hive_table='%s.%s' % (schema, table),        dag=dag,        on_retry_callback=truncate_db,        task_id=task_id)    def truncate_db(context):    hook = MySqlHook('clean_db_export')    hook.run(        'truncate `%s`'%context['task_instance'].task.target_table,        autocommit=False,        parameters=None)

這樣你的重試柄就可以將任務隔離，每次執行某個特定的任務。