要為 Airflow 程式碼做單元測試,關鍵是在 @task 函式中把商業邏輯抽離成可獨立測試的 Python 函式,這樣就不必實際執行 Airflow。你的團隊已在 dags/data_cleaning.py 中完成這件事,將去除姓名前後空白並將電子郵件轉為小寫的邏輯實作在 clean_record 函式裡。
@task
dags/data_cleaning.py
clean_record
你將為這個函式補上單元測試。
本練習屬於課程
將理論付諸實踐,立即體驗我們的互動練習
你將先認識 Airflow 的元件,使用 TaskFlow API 撰寫你的第一個 Dags,並透過 XCom 在任務之間傳遞資料。
接著,你會用動態任務對映平行執行任務,使用 Assets 以資料為基準排程 Dags,並加入人工核准步驟。
本章將透過重試與回呼處理失敗情況,使用可延後的感測器節省資源,並以三個層級測試你的 Dags。
當前練習
在最後一章,你會在 DuckDB 上建置 SQL ETL 管線,使用 Asset Partitions 加入分割區感知的排程,並內嵌資料品質檢查。