你將先認識 Airflow 的元件，使用 TaskFlow API 撰寫你的第一個 Dags，並透過 XCom 在任務之間傳遞資料。

Airflow 核心概念

撰寫最小可行的 TaskFlow Dag

將 Airflow 元件配對到其角色

TaskFlow API

將傳統 Dag 轉換為 TaskFlow

新增工作並串接相依關係

在 Airflow UI 中探索一個 pipeline

在任務之間分享資料

從 TaskFlow 函式回傳資料

將經典 operator 連接到 TaskFlow 的 task

了解 XCom 的限制

使用 TaskFlow 與 XCom 撰寫 Dags

接著，你會用動態任務對映平行執行任務，使用 Assets 以資料為基準排程 Dags，並加入人工核准步驟。

Dynamic Task Mapping（動態任務對應）

用 expand 處理多個端點

用 partial 修正共用參數

使用 zip 配對輸入

使用 Assets 進行資料感知排程

用 Asset outlet 發出資料可用的訊號

在 Asset 上排程 consumer Dag

實際觀察資產為基礎的排程

樣板、冪等性與回填

加入具日期感知的樣板

讓 pipeline 具有冪等性（idempotent）

觸發回補（backfill）

有人參與的工作流程（Human-in-the-loop）

使用 HITLEntryOperator 收集表單輸入

在 UI 中回應必要動作

使用 HITLBranchOperator 進行分支

動態且能感知資料的管線

本章將透過重試與回呼處理失敗情況，使用可延後的感測器節省資源，並以三個層級測試你的 Dags。

具備失敗處理與重試的穩健 Dags

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


@dag
def weather_pipeline():

    # Configure retry strategy
    @task(
        retries=____,
        retry_delay=timedelta(minutes=____),
        retry_exponential_backoff=____,
    )
    def fetch_weather():
        import random
        if random.random() < 0.3:
            raise Exception("API timeout")
        return {"temp": 22, "humidity": 65}

    @task
    def store_forecast(data):
        print(f"Stored forecast: temp={data['temp']}C, humidity={data['humidity']}%")

    weather = fetch_weather()
    store_forecast(weather)


pipeline = weather_pipeline()

if __name__ == "__main__":
    pipeline.test()


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


@dag
def weather_pipeline():

    # Configure retry strategy
    @task(
        retries=3,
        retry_delay=timedelta(minutes=2),
        retry_exponential_backoff=2.0,
    )
    def fetch_weather():
        import random
        if random.random() < 0.3:
            raise Exception("API timeout")
        return {"temp": 22, "humidity": 65}

    @task
    def store_forecast(data):
        print(f"Stored forecast: temp={data['temp']}C, humidity={data['humidity']}%")

    weather = fetch_weather()
    store_forecast(weather)


pipeline = weather_pipeline()

if __name__ == "__main__":
    pipeline.test()


使用指數退避設定重試機制

from airflow.sdk import dag, task


def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["ti"].task_id
    print(f"ALERT: {task_id} in {dag_id} failed!")


# Set the Dag to auto-pause after consecutive failures
@dag(max_consecutive_failed_dag_runs=____)
def monitoring_pipeline():

    # Attach the failure callback to the task
    @task(on_failure_callback=____)
    def check_api_health():
        raise Exception("API unreachable: connection refused")

    check_api_health()


pipeline = monitoring_pipeline()

if __name__ == "__main__":
    pipeline.test()


from airflow.sdk import dag, task


def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["ti"].task_id
    print(f"ALERT: {task_id} in {dag_id} failed!")


# Set the Dag to auto-pause after consecutive failures
@dag(max_consecutive_failed_dag_runs=3)
def monitoring_pipeline():

    # Attach the failure callback to the task
    @task(on_failure_callback=alert_on_failure)
    def check_api_health():
        raise Exception("API unreachable: connection refused")

    check_api_health()


pipeline = monitoring_pipeline()

if __name__ == "__main__":
    pipeline.test()


自動暫停失敗中的 Dag

選對失敗處理策略

可延後執行的運算元

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


@dag
def file_processor():

    # Configure the sensor for production use
    wait_for_data = FileSensor(
        task_id="wait_for_data",
        filepath="/tmp/incoming/report.csv",
        mode="____",
        poke_interval=____,
        timeout=____,
    )

    @task
    def process_report():
        print("Processing incoming report")
        return "report_processed"

    chain(wait_for_data, process_report())


pipeline = file_processor()

if __name__ == "__main__":
    pipeline.test()


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


@dag
def file_processor():

    # Configure the sensor for production use
    wait_for_data = FileSensor(
        task_id="wait_for_data",
        filepath="/tmp/incoming/report.csv",
        mode="reschedule",
        poke_interval=30,
        timeout=300,
    )

    @task
    def process_report():
        print("Processing incoming report")
        return "report_processed"

    chain(wait_for_data, process_report())


pipeline = file_processor()

if __name__ == "__main__":
    pipeline.test()


為正式環境設定感測器

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


@dag
def deferred_file_processor():

    # Enable deferrable mode and set timeout
    wait_for_data = FileSensor(
        task_id="wait_for_data",
        filepath="/tmp/incoming/sales.csv",
        deferrable=____,
        timeout=____,
    )

    @task
    def process_sales():
        print("Processing sales data")
        return "sales_processed"

    chain(wait_for_data, process_sales())


pipeline = deferred_file_processor()

if __name__ == "__main__":
    pipeline.test()


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


@dag
def deferred_file_processor():

    # Enable deferrable mode and set timeout
    wait_for_data = FileSensor(
        task_id="wait_for_data",
        filepath="/tmp/incoming/sales.csv",
        deferrable=True,
        timeout=3600,
    )

    @task
    def process_sales():
        print("Processing sales data")
        return "sales_processed"

    chain(wait_for_data, process_sales())


pipeline = deferred_file_processor()

if __name__ == "__main__":
    pipeline.test()


讓感測器支援延後執行（deferrable）

何時使用 poke、reschedule 或 deferrable

測試 Airflow 程式碼

import pytest
from airflow.models import DagBag

dag_bag = DagBag(include_examples=False)


def test_no_import_errors():
    # Assert that there are no import errors
    assert len(dag_bag.____) == 0, f"Import errors: {dag_bag.import_errors}"


def test_dag_loaded():
    # Assert the expected Dag is present
    assert "____" in dag_bag.____, "Expected Dag not found in DagBag"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


import pytest
from airflow.models import DagBag

dag_bag = DagBag(include_examples=False)


def test_no_import_errors():
    # Assert that there are no import errors
    assert len(dag_bag.import_errors) == 0, f"Import errors: {dag_bag.import_errors}"


def test_dag_loaded():
    # Assert the expected Dag is present
    assert "daily_etl" in dag_bag.dags, "Expected Dag not found in DagBag"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


撰寫 Dag 完整性測試

import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import pytest
from dags.data_cleaning import clean_record


def test_strips_whitespace():
    result = clean_record({"name": "  Alice  ", "email": "alice@test.com"})
    # Verify whitespace is stripped from the name
    assert result["name"] == "____", "Name should have whitespace stripped"


def test_lowercases_email():
    result = clean_record({"name": "Bob", "email": "BOB@Example.COM"})
    # Verify the email is lowercased
    assert result["email"] == "____", "Email should be lowercased"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


from airflow.sdk import dag, task


def clean_record(record):
    """Clean a single record by stripping whitespace and lowercasing email."""
    return {
        "name": record["name"].strip(),
        "email": record["email"].lower(),
    }


@dag
def cleaning_pipeline():

    @task
    def extract():
        return [
            {"name": "  Alice  ", "email": "ALICE@Example.COM"},
            {"name": "Bob  ", "email": "bob@EXAMPLE.com"},
        ]

    @task
    def transform(records):
        return [clean_record(r) for r in records]

    @task
    def load(cleaned):
        for r in cleaned:
            print(f"Loaded: {r['name']} ({r['email']})")

    raw = extract()
    cleaned = transform(raw)
    load(cleaned)


pipeline = cleaning_pipeline()

if __name__ == "__main__":
    pipeline.test()


import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import pytest
from dags.data_cleaning import clean_record


def test_strips_whitespace():
    result = clean_record({"name": "  Alice  ", "email": "alice@test.com"})
    # Verify whitespace is stripped from the name
    assert result["name"] == "Alice", "Name should have whitespace stripped"


def test_lowercases_email():
    result = clean_record({"name": "Bob", "email": "BOB@Example.COM"})
    # Verify the email is lowercased
    assert result["email"] == "bob@example.com", "Email should be lowercased"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


為任務函式撰寫單元測試

import json
from pathlib import Path
import pytest
from airflow.models import DagBag
from pendulum import datetime

dag_bag = DagBag(include_examples=False)


def test_etl_pipeline():
    # Load the Dag from the DagBag
    dag = dag_bag.get_dag("____")
    assert dag is not None, "Dag not found in DagBag"

    # Run the full pipeline as an integration test
    dag.____(logical_date=datetime(2026, 1, 15))

    # Verify the output file was created
    output = Path("____")
    assert output.exists(), "Pipeline did not produce output"

    # Verify the output contains the expected number of results
    results = json.loads(output.read_text())
    assert len(results) == ____, "Expected 2 results"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


import json
from pathlib import Path
import pytest
from airflow.models import DagBag
from pendulum import datetime

dag_bag = DagBag(include_examples=False)


def test_etl_pipeline():
    # Load the Dag from the DagBag
    dag = dag_bag.get_dag("etl_output")
    assert dag is not None, "Dag not found in DagBag"

    # Run the full pipeline as an integration test
    dag.test(logical_date=datetime(2026, 1, 15))

    # Verify the output file was created
    output = Path("/tmp/etl_results.json")
    assert output.exists(), "Pipeline did not produce output"

    # Verify the output contains the expected number of results
    results = json.loads(output.read_text())
    assert len(results) == 2, "Expected 2 results"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-p", "no:cacheprovider", "--no-header", "-rN", "--disable-warnings"])


使用 dag.test() 進行整合測試

用 Task Groups 組織複雜的 Dags

# Import a task group
from airflow.sdk import dag, task, ____


@dag
def order_processing():

    # Set a custom group_id and default_args for the group
    @task_group(group_id="____", default_args={"retries": ____})
    def process_orders():

        @task
        def extract_orders():
            return [{"id": 1, "amount": 99.99}, {"id": 2, "amount": 45.50}]

        @task
        def transform_orders(orders):
            return [{"id": o["id"], "total": round(o["amount"] * 1.08, 2)} for o in orders]
          
		# Call extract and pass to transform
        raw = ____()
        return ____(raw)

    @task
    def load(results):
        print(f"Loaded {len(results)} orders to warehouse")

    orders = process_orders()
    load(orders)


pipeline = order_processing()

if __name__ == "__main__":
    pipeline.test()


# Import a task group
from airflow.sdk import dag, task, task_group


@dag
def order_processing():

    # Set a custom group_id and default_args for the group
    @task_group(group_id="ingest_orders", default_args={"retries": 3})
    def process_orders():

        @task
        def extract_orders():
            return [{"id": 1, "amount": 99.99}, {"id": 2, "amount": 45.50}]

        @task
        def transform_orders(orders):
            return [{"id": o["id"], "total": round(o["amount"] * 1.08, 2)} for o in orders]
          
		# Call extract and pass to transform
        raw = extract_orders()
        return transform_orders(raw)

    @task
    def load(results):
        print(f"Loaded {len(results)} orders to warehouse")

    orders = process_orders()
    load(orders)


pipeline = order_processing()

if __name__ == "__main__":
    pipeline.test()


將相關任務分組

from airflow.sdk import dag, task, task_group


@dag
def multi_stream_pipeline():

    # Set a display name for the parent group
    @task_group(group_display_name="____")
    def process_all():

        @task_group(group_display_name="Ingest Orders")
        def orders():
            @task
            def extract():
                return [{"id": 1, "amount": 99.99}]

            @task
            def transform(data):
                return [{"id": o["id"], "total": o["amount"] * 1.08} for o in data]

            return transform(extract())

        # Add a nested task group for returns with a display name
        @task_group(group_display_name="____")
        def returns():
            @task
            def extract():
                return [{"id": 1, "refund": 25.00}]

            @task
            def transform(data):
                return [{"id": r["id"], "credit": r["refund"] * -1} for r in data]

            return transform(extract())
          
		# Include both groups in the result
        return {"orders": orders(), "returns": ____()}

    @task
    def generate_report(data):
        print(f"Orders: {len(data['orders'])}, Returns: {len(data['returns'])}")

    results = process_all()
    generate_report(results)


pipeline = multi_stream_pipeline()

if __name__ == "__main__":
    pipeline.test()


from airflow.sdk import dag, task, task_group


@dag
def multi_stream_pipeline():

    # Set a display name for the parent group
    @task_group(group_display_name="Process All")
    def process_all():

        @task_group(group_display_name="Ingest Orders")
        def orders():
            @task
            def extract():
                return [{"id": 1, "amount": 99.99}]

            @task
            def transform(data):
                return [{"id": o["id"], "total": o["amount"] * 1.08} for o in data]

            return transform(extract())

        # Add a nested task group for returns with a display name
        @task_group(group_display_name="Process Returns")
        def returns():
            @task
            def extract():
                return [{"id": 1, "refund": 25.00}]

            @task
            def transform(data):
                return [{"id": r["id"], "credit": r["refund"] * -1} for r in data]

            return transform(extract())
          
		# Include both groups in the result
        return {"orders": orders(), "returns": returns()}

    @task
    def generate_report(data):
        print(f"Orders: {len(data['orders'])}, Returns: {len(data['returns'])}")

    results = process_all()
    generate_report(results)


pipeline = multi_stream_pipeline()

if __name__ == "__main__":
    pipeline.test()


使用顯示名稱巢狀化 Task Group

選項

依領域

依 operator 類型

依領域整理任務

為 Dags 做好上線準備

在最後一章，你會在 DuckDB 上建置 SQL ETL 管線，使用 Asset Partitions 加入分割區感知的排程，並內嵌資料品質檢查。

執行 SQL 工作負載

使用 SQL 彙總銷售資料

將 SQL 參數化並抽取到檔案中

為使用者輸入選擇安全的作法

使用 Asset Partitions 的分割區感知排程

建立分割區的 pipeline

在分割資產上進行排程

觀察分割區資產觸發下游 Dag

資料品質檢查

加入欄位層級的品質檢查

加入資料表層級的品質檢查

用正確的運算子攔截資料問題

從開發到上線

從 CLI 觸發一條 pipeline

回填歷史資料

Airflow 的正式環境挑戰

恭喜！

打造可上線的 SQL ETL 管線

想要更進一步精通 Airflow 嗎？本課程採用 Airflow 3.2，帶你學會最新的實作方式。你會使用 TaskFlow API 撰寫 Dags，搭配 Assets 與全新的 Asset Partitions 依資料排程，並透過重試、回呼與測試提升可靠度。

在最後一章，你會在 DuckDB 上打造端到端的 SQL ETL 管線，並直接在 Airflow 中加入資料品質檢查，完全不需第三方函式庫，確保管線產出的資料可信。課程結束時，你將能把原型管線提升為可在正式環境中執行的版本。

Introduction to Apache Airflow in Python

使用TaskFlow API、基於資產的排程和可延遲感測器，運行端到端SQL ETL管道並進行質量檢查。

使用 Airflow 建置資料管線

掌握 TaskFlow API、資產型排程與可延遲感測器，運行具品質檢查的端到端 SQL ETL 流程。



具備失敗處理與重試的穩健 Dags

Create Your Free Account