Airflow 出自 Airbnb，是一款強大的工作流管理平臺。在這個平臺下 ，可以很快速的構建一個 Data Pipeline。然而，如果想保證 Airflow 在生產環境中較高的可用性，還是需要做一些工作提升系統的容錯能力。
可以將 DAG 視做一組相對獨立的的任務集合，DAG 定義了任務集合的執行方式，一個好的 Data Pipeline 需要可讀、易運維，更需要保證災難恢復的時效。一個完整的工作流執行需要 5 個小時執行，如果執行中途偶遇故障，重新執行需要多久？對這個時間的保證，應該在 DAG 設計之初成爲底線，貫穿始終。我們來看下面的例子：
1、Extract.Feature 完成基礎特徵抽取，每天凌晨 3 點 30 執行。
2、Extract.Click 完成點擊分析，Schedule 爲 None。
3、EXtract.Download 將數據從 ODPS 下載到本地，轉存爲 CSV 文件，Schedule 爲 None。
上述三個 DAG 其實是一個相關的業務組，在實現中根據業務相關性，將其拆分 ，並通過 TriggerDagRunOperator 管理依賴關係。
先看一個 DAG 的代碼定義，及其 Graph View：
這個 DAG 有一定的複雜度，Airflow 調度多臺機器協同計算，完成一個機器學習項目的離線推理過程，Airflow 部署在其中一個機器，或通過 SSH 的方式向其他機器發送命令，或在本地處理。經過層層優化，所有機器加起來，剛好在規定時間內完成計算任務。
圖中三個失敗的 sort_task 讓這張圖顯得不夠完美，但事發之時，我也沒有絲毫擔心，因爲 sort_task 的位置決定了他的狀態無障大局，使用 Airflow 提供的命令行工具可以輕鬆回補。
以 Scoring.Run.train.0 爲例查看路徑 [scoreing_run_task，sort_task，raw_to_odps_task] 上各節點的用時佔比：
從時間上看，scoreing_run_task 無疑是重點，而所有的 scoreing_run_task 的實例則是 DAG 的核心路徑：
在任務編排時，將 sort_task（在上圖中是：Scoring.Run.Sort.train.4.6）抽離到 Airflow 所在節點，時間上與 scoreing_run_task 的後序實例（在上圖是：Scoring.Run.train.4.7）保持併發，這種編排在最大程度上保證任務用時，同時避免 sort_task 產生意外殃及池魚。
在上一小節的示例 DAG 中，有三個處於錯誤狀態的任務，錯誤的原因也很巧合，與本例有關：
大概看了一眼便明白，sort_task 併發量過高（取決於前序任務的併發量）導致內存不足，另外，同時在運行的另外一個 DAG 也有一類消耗內存的任務在多併發執行。在代碼層級，我們很難以一種經濟的手段去控制不同任務的的併發量，這種情況就像明星無法避免在紅毯上與人尷尬的撞衫……
瞭解之後發現 Airflow 在 BaseOperator 層實現了併發控制。 打開 UI，在 Admin >> Pools 菜單中，添加一個 key，設置一個合適的 Slots：
接下來分別在不同的 DAG 中修改任務。事實上，Airflow（非集羣模式下）能做的也只是併發量的控制，至於每個函數實現到底用多少資源，還是需要自己控制。但對我而言，內存不足的問題已經完美解決。
「除非確定，否則別輕易關閉」，這些參數可以放到全局的 default_args 中，郵件通知第一時間讓我獲得的異常信息。如果手速快，在處理完異常之後，系統可能已經自愈了。
關於重試，需要提一下 SubDag，如果 SubDag 中的某個任務發生了錯誤，那麼重試不是從當前錯誤的任務開始。他會從頭再來，從本質上看，SubDag 其實也是一個 Task。
exit(0) 代表的唯一意義是 業務層的成功，在使用 BashOperator 時最容易遭遇 exit(0) 陷阱，直接的後果是所有的任務看起來成功，但事實上結果不對。
覈對日誌會發現，在不起眼的地方，產生了不起眼的異常，而這個異常被不正確的 exit(0) 掩蓋掉了，避免的方式也很簡單：
1、command1 && command2 && ... ，
2、在 bash 頭部添加： set -e
我們使用 Airflow 的經驗有限， 相關總結難免有疏漏和不足之處， 也希望和大家一起學習成長。
作者介紹
雲腦科技高級軟件研發工程師 田間
飛信社區 BI 團隊負責人， 負責搭建商業智能平臺， 爲飛信社區數據化運營提供決策支持， 完成飛信好友推薦系統的孵化開發。
翻東西數據團隊架構師， 負責電商數據的抓取及挖掘。
前融雲架構師，負責公司大數據平臺及DevOps系統。
在大數據及分佈式系統領域有着豐富的實戰經驗。