Airflow任務調度延時問題分析和優化

作者 | 高光軒

編輯 | 高衛華

出品 | CSDN（ID：CSDNnews）

背景

airflow作為apache基金會的一款開源的優秀調度系統，目前被國內外很多大中型企業使用；其豐富的算子(operator)型別和極易擴展的支持，被很多企業進行相關的自定義改造和二次開發以滿足自身的業務需求，

但是我們不難發現幾個問題，隨著用戶腳本(dag檔案)和工程目錄數量越來越多,我們可能面臨整任務出現了延時調度的現象，

舉個例子說明下，假設你設定了一個任務是每天8:00跑，但是你發現到了調度的時候延時到了8:02或者某個DAG上游節點調度完畢后，下游節點需要等很久才能得到調度，

今天我們就針對這個問題進行相關的分析并提出幾點相關的優化建議，

針對以上存在的問題，我認為可以從以下幾個方面對airflow的調度延時問題進行分析和優化，后面我們將對每個部分進行相關的介紹，

延時調度問題幾個思考的方向

dag的決議邏輯對延時的影響

在定位問題前，我認為了解airflow原始碼dag(有向無環圖,作業流的描述關系)檔案作業原理能夠使我們更好的定位分析問題，在airflow原始碼中有兩個函式和dag檔案的加載掃描有直接關系，

一個函式叫做:list_py_file_paths；一個叫做:process_file，從字面可以看出，list_py_file_path主要是用來收集整個dag目錄下的所有py檔案，process_file主要是用來處理目錄下的檔案，

① 針對list_py_file_path的邏輯，我們只需要關注這幾行(部分原始碼省略):

②我們重點分析下process_file這個函式的邏輯分析，

整個代碼的邏輯并不復雜，由于代碼比較多，這里我只簡單的介紹下它的作業原理，以及相關的處理方案，process_file函式主要用來處理AIRFLOW_HOME(環境變數)目錄下的檔案，

目前對于airflow而言支持處理兩種型別的檔案加載(python檔案和zip檔案)，其他型別會做過濾，在整個處理程序中有兩個地方其實是比較耗時的，如下：

所以通過以上分析我們可以得到兩點優化建議：

① 在撰寫python dag腳本中，應特別注意規范，應避免大量的計算操作,如果存在不合理的耗時邏輯，那么在加載dag檔案的時候，就會執行函式內部執行邏輯運算，

比如我在dag檔案里面寫了一個函式，然后直接呼叫函式，函式內部的邏輯是sleep操作，那么在dag檔案大量加載的時候，一輪的調度掃描下來，將會花費大量的時間，甚至導致超時例外，導致scheduler例外，這點要尤其注意，所以dag撰寫規范特別重要，

② 如果我們的線上dag撰寫全是py型別檔案，在不需要zip檔案的場景的操作下，可以將zip操作邏輯注釋掉，

zip處理其實在整個dag的處理程序中占用了較多的時間，即使你的專案中沒有zip檔案，那么對于process_file函式中也會有這樣的邏輯判斷，這樣的話其實對于掃描dag檔案而言會存在大量的運算，導致整個調度延時，

小結：根據以上的粗略分析，我們后面可以重點針對dag決議原理和dag撰寫規范兩個方向進行調度延時相關的優化，

磁盤的性能問題對調度延時的影響

在搭建部署airflow的時候，很多企業會采用以下兩種方式部署：

物理機部署的方式有其自身的優點但是也存在很多缺點，采用物理機方式在dag掃描和加載速度上要比容器化部署的方式快很多，

本質上的區別是：物理機直接從本地磁盤讀取dag的內容，而容器化部署的方式則需要從網路磁盤讀取內容，網路磁盤則需要走網路的IO，

所以從檔案的加載效率上講，物理機更勝一籌，但是物理機面臨的問題也很多，比如資源的動態擴展，行程監控和重啟機制等都不如容器化方便和高效，

與之對比，就容器化部署方式來說，目前業界最常用的就是基于k8s方式的部署，這種方式橫向擴展airflow特別方便，

只要宿主機的資源比較大，那么對于airflow的web角色、scheduler角色，還是worker角色，擴展機器都是頁面點擊的操作，特別方便；甚至后面可以探究成自動的擴容和縮容機制，

并且在監控和報警層面k8s本身有自身的優勢，包括行程的監控也有其自身獨特的重啟策略和機制，

對于資源的規劃和分配，通過k8s也可以更好的去操作處理，但是k8s方式需要一個全域共享盤，一般來說我們常用的就是一些網路存盤組件，無論是阿里的oss，nas還是亞馬遜的s3等等，它們都有一個共性就是需要走網路IO，同時存在這資料一致性的風險，

網路IO的延時在T+1調度或者小時級調度粒度上也完全沒有太多的問題，但隨著業務量的增加比如10w+調度任務，可能就會存在延時問題，

小結：針對airflow的調度方式，我們可以根據自身的業務需要和任務量來選擇部署方式，如果線上的任務數量很多，單個集群出現了性能瓶頸可以考慮多集群部署方式，來保證橫向擴展解決IO的處理過慢的問題，

airflow的引數配置對調度延時的影響

① 對于dag延時調度在airflow的引數上也需要注意合理話的配置，一般這些影響引數可以包括如下：

②單dag內部task并行度和集群并行度

parallelism：

全域(集群）任務最大并行度,這個值需要根據整體集群的性能配置和單結點worker行程數量進行配置，如果配置的過大，本身集群的性能跟不上也是徒勞的，假設我們的airflow的集群worker節點所啟動的默認worker行程是10個，我們有10臺機器，理論上講最大并行度100左右，假設這個值配置200，最大也只能達到100的并發，

為什么說這個值會影響調度延時呢？因為對于airflow這樣的調度系統，0-24小時都可能有大量的任務在調度，假設某一時刻任務峰值比較大，就會造成其他的任務等待，就會造成延時，這個時候就需要考慮這個引數的配置或者整體集群的擴容了，

dag_concurrency：

這個引數比較容易理解，就是對于一個dag，最多能同時跑多少個task，如果某一時刻同一個dag下的任務啟動的比較多，最多并行跑的數量也只能是dag_concurrency的值,所以就會造成其它任務的等待和延時，當然我們不能把

這個引數配置的過大，要考慮系統中dag之間的影響，如果單個dag的并行度過高，可能會造成其它dag的任務調度延時或者長時間得不到執行，

③ 任務優先級引數對調度延時造成的影響，

任務優先級這個很明顯問題，由于很多任務的優先級設定的比較低，造成優先級高的任務先執行以至于后面的低優先級的任務得不到執行，那么我們先分析下airflow背后的優先級的邏輯，

airflow的優先級調度原始碼如下:

舉個例子，假設我們有一個如下圖圖結構的dag圖，并且每個任務的節點的優先級都是默認值1

根據airflow的優先級計算策略我們可以把它歸為3類:

1.任務本身優先級值的配置作為最終任務優先級的值，任務優先級配置可以在operator引數指定，官方原始碼引數解釋如下：

2. 任務自身的優先級配置值加上所有下游優先級配置值之和，

我們可以這樣解釋，如下下圖描述:

3. 任務自身優先級的配置值加上所有上游優先級配置值之和，

我們可以這樣解釋，如下下圖描述:

小結：任務優先級配置也會對任務的執行造成延時，我們在日常的任務執行計劃中，應當根據自身任務的緊急程度設定不同的優先級策略，

對于airflow而言優先級的默認配置是WeightRule.DOWNSTREAM，這種方式其實需要根據同一時刻有哪些task需要執行，并根據task所在dag的結構圖計算出優先級值，進行相關的邏輯排名，然后進行執行，

因此在以后的生產環境中要根據真實的需要進行相關的優先級值的調整，

更多精彩推薦

?與 Brian Kernighan 一起回憶 Unix 的誕生！
?挑戰 Linux 之父認為的“不可能”：向 M1 Mac 移植 Linux

?TIOBE 12 月編程語言：Python 有望第四次成為年度語言！

?升級版APDrawing，人臉照秒變線條肖像畫，細節呈現驚人

?中科大“九章”歷史性突破，但實作真正的量子霸權還有多遠？

?云原生應用Go語言：你還在考慮的時候，別人已經應用實踐

點分享點點贊點在看