【airflow實戰系列】 基於 python 的調度和監控工做流的平臺

簡介

airflow 是一個使用python語言編寫的data pipeline調度和監控工做流的平臺。Airflow被Airbnb內部用來建立、監控和調整數據管道。任何工做流均可以在這個使用Python來編寫的平臺上運行。

Airflow是一種容許工做流開發人員輕鬆建立、維護和週期性地調度運行工做流（即有向無環圖或成爲DAGs）的工具。在Airbnb中，這些工做流包括瞭如數據存儲、增加分析、Email發送、A/B測試等等這些跨越多部門的用例。

這個平臺擁有和 Hive、Presto、MySQL、HDFS、Postgres和S3交互的能力，而且提供了鉤子使得系統擁有很好地擴展性。除了一個命令行界面，該工具還提供了一個基於Web的用戶界面讓您能夠可視化管道的依賴關係、監控進度、觸發任務等。

傳統Workflow一般使用TextFiles(json,xml/etc)來定義DAG,而後Scheduler解析這些DAG文件造成具體的TaskObject執行；Airflow沒這麼幹，它直接用Python寫DAGdefinition,一會兒突破了文本文件表達能力的侷限，定義DAG變得簡單。

Airflow的架構

在一個可擴展的生產環境中，Airflow含有如下組件：

• 一個元數據庫（MySQL或Postgres）

• 一組Airflow工做節點

• 一個調節器（Redis或RabbitMQ）

• 一個Airflow Web服務器

全部這些組件能夠在一個機器上隨意擴展運行。若是使用LocalExcuter來適度的安裝則能夠得到至關多的額外性能。

優勢

• python腳本實現DAG，很是容易擴展

• 工做流依賴可視化

• no XML

• 可測試

• 可做爲crontab的替代

• 可實現複雜的依賴規則

• Pools

• CLI和Web UI

功能簡介

常見命令

• initdb，初始化元數據DB，元數據包括了DAG自己的信息、運行信息等；

• resetdb，清空元數據DB；

• list_dags，列出全部DAG；

• list_tasks，列出某DAG的全部task；

• test，測試某task的運行情況；

• backfill，測試某DAG在設定的日期區間的運行情況；

• webserver，開啓webserver服務；

• scheduler，用於監控與觸發DAG。

ETL

ETL，是英文 Extract-Transform-Load 的縮寫，用來描述將數據歷來源端通過抽取（extract）、轉換（transform）、加載（load）至目的端的過程。ETL一詞較經常使用在數據倉庫，但其對象並不限於數據倉庫。

Airflow設計時，只是爲了很好的處理ETL任務而已，可是其精良的設計，正好能夠用來解決任務的各類依賴問題。

任務依賴

一般，在一個運維繫統，數據分析系統，或測試系統等大型系統中，咱們會有各類各樣的依賴需求。好比：

• 時間依賴：任務須要等待某一個時間點觸發

• 外部系統依賴：任務依賴Mysql中的數據，HDFS中的數據等等，這些不一樣的外部系統須要調用接口去訪問

• 機器依賴：任務的執行只能在特定的某一臺機器的環境中，可能這臺機器內存比較大，也可能只有那臺機器上有特殊的庫文件

• 任務間依賴：任務A須要在任務B完成後啓動，兩個任務互相間會產生影響

• 資源依賴：任務消耗資源很是多，使用同一個資源的任務須要被限制，好比跑個數據轉換任務要10個G，機器一共就30個G，最多隻能跑兩個，我但願相似的任務排個隊

• 權限依賴：某種任務只能由某個權限的用戶啓動

也許你們會以爲這些是在任務程序中的邏輯須要處理的部分，可是我認爲，這些邏輯能夠抽象爲任務控制邏輯的部分，和實際任務執行邏輯解耦合。

如何理解Crontab

如今讓咱們來看下最經常使用的依賴管理系統，Crontab

在各類系統中，總有些定時任務須要處理，每當在這個時候，咱們第一個想到的老是crontab。

確實，crontab能夠很好的處理定時執行任務的需求，可是對於crontab來講，執行任務，只是調用一個程序如此簡單，而程序中的各類邏輯都不屬於crontab的管轄範圍（很好的遵循了KISS）

因此咱們能夠抽象的認爲：

crontab是一種依賴管理系統，並且只管理時間上的依賴。

Airflow的處理依賴的方式

Airflow的核心概念，是DAG(有向無環圖)，DAG由一個或多個TASK組成，而這個DAG正是解決了上文所說的任務間依賴。Task A 執行完成後才能執行 Task B，多個Task之間的依賴關係能夠很好的用DAG表示完善

Airflow完整的支持crontab表達式，也支持直接使用python的datatime表述時間，還能夠用datatime的delta表述時間差。這樣能夠解決任務的時間依賴問題。

Airflow在CeleryExecuter下可使用不一樣的用戶啓動Worker，不一樣的Worker監聽不一樣的Queue，這樣能夠解決用戶權限依賴問題。Worker也能夠啓動在多個不一樣的機器上，解決機器依賴的問題。

Airflow能夠爲任意一個Task指定一個抽象的Pool，每一個Pool能夠指定一個Slot數。每當一個Task啓動時，就佔用一個Slot，當Slot數佔滿時，其他的任務就處於等待狀態。這樣就解決了資源依賴問題。

Airflow中有Hook機制(其實我以爲不該該叫Hook)，做用時創建一個與外部數據系統之間的鏈接，好比Mysql，HDFS，本地文件系統(文件系統也被認爲是外部系統)等，經過拓展Hook可以接入任意的外部系統的接口進行鏈接，這樣就解決的外部系統依賴問題。

參考

http://wingerted.com/2017/02/20/introduce-to-airflow/

https://www.youtube.com/watch?v=cHATHSB_450

https://www.youtube.com/watch?v=Pr0FrvIIfTU

 

此文已由做者受權騰訊雲技術社區發佈，轉載請註明文章出處