你將先認識 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（動態任務對應）

from airflow.sdk import dag, task


@dag
def api_processor():

    @task
    def get_endpoints():
        return ["users", "orders", "products"]

    @task
    def process_endpoint(endpoint):
        print(f"Processing: {endpoint}")
        return f"{endpoint}_done"

    # Get the list of endpoints
    endpoints = ____()
    # Map over the list with expand
    process_endpoint.____(endpoint=____)


pipeline = api_processor()

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


from airflow.sdk import dag, task


@dag
def api_processor():

    @task
    def get_endpoints():
        return ["users", "orders", "products"]

    @task
    def process_endpoint(endpoint):
        print(f"Processing: {endpoint}")
        return f"{endpoint}_done"

    # Get the list of endpoints
    endpoints = get_endpoints()
    # Map over the list with expand
    process_endpoint.expand(endpoint=endpoints)


pipeline = api_processor()

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


用 expand 處理多個端點

from airflow.sdk import dag, task


@dag
def api_processor_v2():

    @task
    def get_endpoints():
        return ["users", "orders", "products"]

    @task
    def process_endpoint(endpoint, base_url):
        print(f"Fetching: {base_url}/{endpoint}")
        return f"{endpoint}_done"

    endpoints = get_endpoints()
    # Fix the shared parameter and expand over endpoints
    process_endpoint.____(base_url="https://api.company.com/v2").____(endpoint=endpoints)


pipeline = api_processor_v2()

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


from airflow.sdk import dag, task


@dag
def api_processor_v2():

    @task
    def get_endpoints():
        return ["users", "orders", "products"]

    @task
    def process_endpoint(endpoint, base_url):
        print(f"Fetching: {base_url}/{endpoint}")
        return f"{endpoint}_done"

    endpoints = get_endpoints()
    # Fix the shared parameter and expand over endpoints
    process_endpoint.partial(base_url="https://api.company.com/v2").expand(endpoint=endpoints)


pipeline = api_processor_v2()

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


用 partial 修正共用參數

使用 zip 配對輸入

使用 Assets 進行資料感知排程

from airflow.sdk import dag, task, Asset


# Define the asset
sales_data = ____("s3://data-lake/sales/daily.csv")

@dag
def sales_producer():

    @task(outlets=[sales_data])
    def write_sales_data():
        print("Writing daily sales data")
        return "sales_data_written"

    write_sales_data()


pipeline = sales_producer()

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


from airflow.sdk import dag, task, Asset


# Define the asset
sales_data = Asset("s3://data-lake/sales/daily.csv")

@dag
def sales_producer():

    @task(outlets=[sales_data])
    def write_sales_data():
        print("Writing daily sales data")
        return "sales_data_written"

    write_sales_data()


pipeline = sales_producer()

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


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

from airflow.sdk import dag, task, Asset


# Define the same asset as the producer
sales_data = ____("s3://data-lake/sales/daily.csv")

# Schedule the Dag on the asset
@dag(____=[sales_data])
def sales_consumer():

    @task
    def analyze_sales():
        print("Analyzing latest sales data")
        return "analysis_complete"

    analyze_sales()


pipeline = sales_consumer()

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


from airflow.sdk import dag, task, Asset


# Define the same asset as the producer
sales_data = Asset("s3://data-lake/sales/daily.csv")

# Schedule the Dag on the asset
@dag(schedule=[sales_data])
def sales_consumer():

    @task
    def analyze_sales():
        print("Analyzing latest sales data")
        return "analysis_complete"

    analyze_sales()


pipeline = sales_consumer()

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


在 Asset 上排程 consumer Dag

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

樣板、冪等性與回填

from airflow.sdk import dag
from airflow.providers.standard.operators.bash import BashOperator


@dag(schedule="@daily")
def daily_export():

    # Replace hardcoded date with Jinja template
    export = BashOperator(
        task_id="export_data",
        bash_command="echo 'Exporting data for {{ ____ }}' > /tmp/export_{{ ____ }}.csv",
    )


pipeline = daily_export()

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


from airflow.sdk import dag
from airflow.providers.standard.operators.bash import BashOperator


@dag(schedule="@daily")
def daily_export():

    # Replace hardcoded date with Jinja template
    export = BashOperator(
        task_id="export_data",
        bash_command="echo 'Exporting data for {{ ds }}' > /tmp/export_{{ ds }}.csv",
    )


pipeline = daily_export()

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


加入具日期感知的樣板

from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import (
    SQLExecuteQueryOperator,
    SQLInsertRowsOperator,
)


@dag(schedule="@daily")
def idempotent_load():

    # Complete the WHERE clause to filter by logical date
    staged_rows = SQLExecuteQueryOperator(
        task_id="get_staged_rows",
        conn_id="duckdb_default",
        sql="SELECT * FROM staging WHERE ____ = '{{ ds }}'",
    )

    # Add a preoperator to delete existing rows before inserting
    load = SQLInsertRowsOperator(
        task_id="load_sales",
        conn_id="duckdb_default",
        table_name="sales",
        columns=["date", "product", "amount"],
        preoperator="____ FROM sales WHERE date = '{{ ds }}';",
        rows=staged_rows.output,
    )


pipeline = idempotent_load()

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


from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import (
    SQLExecuteQueryOperator,
    SQLInsertRowsOperator,
)


@dag(schedule="@daily")
def idempotent_load():

    # Complete the WHERE clause to filter by logical date
    staged_rows = SQLExecuteQueryOperator(
        task_id="get_staged_rows",
        conn_id="duckdb_default",
        sql="SELECT * FROM staging WHERE date = '{{ ds }}'",
    )

    # Add a preoperator to delete existing rows before inserting
    load = SQLInsertRowsOperator(
        task_id="load_sales",
        conn_id="duckdb_default",
        table_name="sales",
        columns=["date", "product", "amount"],
        preoperator="DELETE FROM sales WHERE date = '{{ ds }}';",
        rows=staged_rows.output,
    )


pipeline = idempotent_load()

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


讓 pipeline 具有冪等性（idempotent）

觸發回補（backfill）

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

from airflow.sdk import dag, task, chain, Param
# Import HITLEntryOperator
from airflow.providers.standard.operators.____ import ____


@dag
def ticket_review():

    @task
    def generate_draft():
        # In production, this would call an LLM API to generate the draft
        return (
            "Hi Sarah, thanks for reaching out. I can see your last "
            "invoice was charged twice due to a billing system error. "
            "We have initiated a refund of $49.99 which should appear "
            "within 3-5 business days. Let me know if you need anything else."
        )

    # Complete the HITLEntryOperator with params
    review = ____(
        task_id="human_review",
        subject="Review AI draft response",
        body="{{ ti.xcom_pull(task_ids='generate_draft') }}",
        params={
            "____": ____("", type="string"),
            "____": ____("p3", type="string"),
        },
    )

    @task
    def send_response(hitl_output):
        feedback = hitl_output["params_input"]["feedback"]
        urgency = hitl_output["params_input"]["urgency"]
        print(f"Sending response (urgency: {urgency}): {feedback}")

    draft = generate_draft()
    chain(draft, review)
    send_response(review.output)


pipeline = ticket_review()

if __name__ == "__main__":
    from unittest.mock import patch

    mock_hitl_response = {
        "chosen_options": [],
        "params_input": {"feedback": "Looks good", "urgency": "p3"},
    }

    with patch(
        "airflow.providers.standard.operators.hitl.HITLEntryOperator.execute",
        return_value=mock_hitl_response,
    ):
        pipeline.test()


from airflow.sdk import dag, task, chain, Param
# Import HITLEntryOperator
from airflow.providers.standard.operators.hitl import HITLEntryOperator


@dag
def ticket_review():

    @task
    def generate_draft():
        # In production, this would call an LLM API to generate the draft
        return (
            "Hi Sarah, thanks for reaching out. I can see your last "
            "invoice was charged twice due to a billing system error. "
            "We have initiated a refund of $49.99 which should appear "
            "within 3-5 business days. Let me know if you need anything else."
        )

    # Complete the HITLEntryOperator with params
    review = HITLEntryOperator(
        task_id="human_review",
        subject="Review AI draft response",
        body="{{ ti.xcom_pull(task_ids='generate_draft') }}",
        params={
            "feedback": Param("", type="string"),
            "urgency": Param("p3", type="string"),
        },
    )

    @task
    def send_response(hitl_output):
        feedback = hitl_output["params_input"]["feedback"]
        urgency = hitl_output["params_input"]["urgency"]
        print(f"Sending response (urgency: {urgency}): {feedback}")

    draft = generate_draft()
    chain(draft, review)
    send_response(review.output)


pipeline = ticket_review()

if __name__ == "__main__":
    from unittest.mock import patch

    mock_hitl_response = {
        "chosen_options": [],
        "params_input": {"feedback": "Looks good", "urgency": "p3"},
    }

    with patch(
        "airflow.providers.standard.operators.hitl.HITLEntryOperator.execute",
        return_value=mock_hitl_response,
    ):
        pipeline.test()


使用 HITLEntryOperator 收集表單輸入

在 UI 中回應必要動作

使用 HITLBranchOperator 進行分支

動態且能感知資料的管線

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

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

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

自動暫停失敗中的 Dag

選對失敗處理策略

可延後執行的運算元

為正式環境設定感測器

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

何時使用 poke、reschedule 或 deferrable

測試 Airflow 程式碼

撰寫 Dag 完整性測試

為任務函式撰寫單元測試

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

用 Task Groups 組織複雜的 Dags

將相關任務分組

使用顯示名稱巢狀化 Task Group

依領域整理任務

為 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 流程。



樣板、冪等性與回填

Create Your Free Account