在本章中，你會完整認識 Apache Airflow 的組成元件，並了解該如何、以及為什麼要使用它們。

Apache Airflow 入門

Dag 計數器

執行你的第一個 Dag

Airflow Dags

建立一個簡單的 Dag

操作 Dags 與 Airflow shell

檢視 Airflow 指令

在 Airflow 測試一個任務

Airflow 網頁介面

瀏覽 Airflow UI

使用 Airflow UI 檢視 Dags

Dag 錯誤

Airflow 入門

DAG 準備好了嗎？現在要來學習實作 Airflow DAGs 的基礎。透過動手操作，你會學到如何設定與部署 operators、tasks，以及排程。

Airflow operators

為任務加上裝飾器

使用 @task.bash 定義 BashOperator

定義任務的執行順序

排程 Dags

透過 Python 排程一個 Dag

解讀 Airflow 的排程

排解 Dag 執行問題

使用 XCom 在任務之間傳遞資料

用 XCom 傳遞資料

XCom 變數

Airflow 感測器

判定工作執行順序

建立 FileSensor

感官剝奪

在 Airflow 中建立 DAGs

在本章中，你會學到如何運用 Airflow 的元件（例如 sensors 與 executors）來節省時間，同時監控並疑難排解 Airflow 工作流程。

監控、警示與回呼

實作回呼函式

實作 SmtpNotifier

日誌裡有什麼？

使用 Jinja 模板

使用 Jinja 撰寫

替換參數

Airflow 變數

答案是 Variable

在 Python 中讀取變數

Variables 有點不對勁

在 Airflow 進行除錯與疑難排解

Dag 疑難排解

from airflow.sdk import dag, task
#from airflow.providers.standard.sensors.filesystem import FileSensor
from pendulum import datetime

@dag(
    dag_id="execute_report",
    schedule="0 0 * * *",
    start_date=datetime(2026, 3, 20),
)
def execute_report():

    check_for_datafile = FileSensor(
        task_id="check_for_datafile",
        filepath="salesdata_ready.csv",
        mode="poke",
    )

    @task()
    def generate_report():
        print("Generating report...")

    check_for_datafile >> generate_report()

execute_report()

from airflow.sdk import dag, task
from pendulum import datetime

@dag(
    dag_id="sample_dag",
    schedule="0 0 * * *",
    start_date=datetime(2023, 2, 20),
)
def sample_dag():

    @task()
    def sample():
        print("Running sample task...")

    sample()

sample_dag()

from airflow.sdk import dag, task
from airflow.providers.standard.sensors.filesystem import FileSensor
from pendulum import datetime

@dag(
    dag_id="execute_report",
    schedule="0 0 * * *",
    start_date=datetime(2026, 3, 20),
)
def execute_report():

    check_for_datafile = FileSensor(
        task_id="check_for_datafile",
        filepath="salesdata_ready.csv",
        mode="poke",
    )

    @task()
    def generate_report():
        print("Generating report...")

    check_for_datafile >> generate_report()

execute_report()

遺失的 Dag

維護與監控 Airflow 工作流程

把所學統整起來。在最後一章中，你會綜合應用先前學到的內容，在 Airflow 中打造可供正式環境使用的工作流程。

觸發器與容錯機制

實作 Trigger Rule

加入重試機制

觸發子 Dag

分支

定義一個 @task.branch

分支疑難排解

HITL Operator

加入人工核准

ApprovalOperator 任務

建立可供正式環境使用的管線

設定 sales ETL pipeline

加入每月分支

加入人工審核關卡

恭喜你！

控制 DAG 邏輯

已更新至 Apache Airflow 3.1.6——按排程交付資料常常得手動處理。你會寫腳本、加上複雜的 cron 工作，還得用各種方法去滿足不斷變動的需求——而且和同事協作時要管理這一切更麻煩。Apache Airflow 能透過為你的工作流程加入排程、錯誤處理與回報，減輕這些負擔。在本課程中，你會學會 Apache Airflow 的基礎，並了解如何在正式環境中實作複雜的資料工程 pipeline。你也會學到如何使用有向無環圖（DAGs）、自動化資料工程工作流程，以及以簡單且可重複的方式實作資料工程任務，讓你更輕鬆維護與擴充。

Intermediate Python

Introduction to Shell

學習使用Airflow和Python中的DAG實現數據工程工作流程，添加調度、錯誤處理和報告功能。

Python 中的 Apache Airflow 入門

學習如何實作並排程資料工程工作流程。

Airflow 基礎

資料工程師 使用 Python

替換參數

Python 中的 Apache Airflow 入門

動手互動練習