建立生產環境的 pipeline

Python 中的 Apache Airflow 入門

Mike Metzger

Data Engineer

執行 Dags 與任務

從命令列執行特定任務：

airflow tasks test <dag_id> <task_id> <date>

執行整個 Dag：

airflow dags trigger --logical-date <date> <dag_id>
Python 中的 Apache Airflow 入門

任務小抄

  • @task－將指定的 Python 函式標註為 Airflow 任務
  • @task.bash－將 bash 指令的輸出作為 Airflow 任務結果
  • @task.branch－在 Dag 中提供執行時分支選擇的運算子
  • FileSensor－需要 filepath 參數，可能還需 mode 或 poke_interval 屬性
Python 中的 Apache Airflow 入門

樣板重點提醒

  • Airflow 中許多物件可使用樣板
  • 某些欄位可用樣板字串，其他則不行
  • 檢查方式之一是看內建說明文件：
  1. 開啟 python3 直譯器
  2. 匯入必要函式庫（例如：from airflow.sdk import dag, task）
  3. 在提示字元執行 help(<Airflow object>)，例如 help(task)
  4. 找到提到「template_fields」的行。這會列出可用樣板的參數。
Python 中的 Apache Airflow 入門

樣板文件範例

終端機顯示 Airflow 物件的 python3 help 輸出

終端機輸出中標示 Airflow 說明裡的 template_fields 行

Python 中的 Apache Airflow 入門

使用 Airflow

建立、排程與監控工作流程的示意圖

Apache Airflow 標誌

Python 中的 Apache Airflow 入門

一起來練習吧！

Python 中的 Apache Airflow 入門

Preparing Video For Download...