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

Airflow 核心概念

# Import the Dag and task decorators
from airflow.sdk import ____, ____


# Apply the Dag decorator
@____
def my_first_pipeline():

    # Apply the task decorator and return the expected string
    @____
    def start():
        ____ "Pipeline started"

    start()


pipeline = my_first_pipeline()

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


# Import the Dag and task decorators
from airflow.sdk import dag, task


# Apply the Dag decorator
@dag
def my_first_pipeline():

    # Apply the task decorator and return the expected string
    @task
    def start():
        return "Pipeline started"

    start()


pipeline = my_first_pipeline()

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


撰寫最小可行的 TaskFlow Dag

選項

編排（Orchestration）

執行（Execution）

介面與儲存（Interface & Storage）

將 Airflow 元件配對到其角色

TaskFlow API

from airflow.sdk import dag, task

# Add the Dag decorator
@____
def etl_pipeline():

    # Add the task decorator
    @____
    def extract_data():
        return {"users": 150, "events": 4200}

    # Add the task decorator
    @____
    def print_summary(data):
        print(f"Users: {data['users']}, Events: {data['events']}")

    # Wire dependencies by passing return values
    data = ____()
    ____(data)


pipeline = etl_pipeline()

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


# Classic version (read-only reference)
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator


def extract_data():
    return {"users": 150, "events": 4200}


def print_summary(data):
    print(f"Users: {data['users']}, Events: {data['events']}")


with DAG(dag_id="etl_pipeline") as dag:
    extract_task = PythonOperator(
        task_id="extract_data",
        python_callable=extract_data,
    )
    summary_task = PythonOperator(
        task_id="print_summary",
        python_callable=print_summary,
        op_args=[extract_task.output],
    )
    extract_task >> summary_task


from airflow.sdk import dag, task

# Add the Dag decorator
@dag
def etl_pipeline():

    # Add the task decorator
    @task
    def extract_data():
        return {"users": 150, "events": 4200}

    # Add the task decorator
    @task
    def print_summary(data):
        print(f"Users: {data['users']}, Events: {data['events']}")

    # Wire dependencies by passing return values
    data = extract_data()
    print_summary(data)


pipeline = etl_pipeline()

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

將傳統 Dag 轉換為 TaskFlow

from airflow.sdk import dag, task

@dag
def user_pipeline():

    @task
    def extract():
        return [
            {"id": 1, "name": "alice"},
            {"id": 2, "name": "bob"},
            {"id": 3, "name": "charlie"},
        ]

    @task
    def transform(raw_data):
        return [{"id": r["id"], "name": r["name"].upper()} for r in raw_data]

    # Add a decorated task function
    @____
    def load(____):
        print(f"Loaded {len(____)} records")

    raw = extract()
    cleaned = transform(raw)
    # Call the new task
    ____(cleaned)


pipeline = user_pipeline()

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


from airflow.sdk import dag, task

@dag
def user_pipeline():

    @task
    def extract():
        return [
            {"id": 1, "name": "alice"},
            {"id": 2, "name": "bob"},
            {"id": 3, "name": "charlie"},
        ]

    @task
    def transform(raw_data):
        return [{"id": r["id"], "name": r["name"].upper()} for r in raw_data]

    # Add a decorated task function
    @task
    def load(records):
        print(f"Loaded {len(records)} records")

    raw = extract()
    cleaned = transform(raw)
    # Call the new task
    load(cleaned)


pipeline = user_pipeline()

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


新增工作並串接相依關係

在 Airflow UI 中探索一個 pipeline

在任務之間分享資料

from airflow.sdk import dag, task

@dag
def xcom_example():

    @task
    def get_config():
        return {"pipeline_name": "daily_etl", "batch_size": 1000}

    @task
    def run_pipeline(config):
        print(f"Running {config['pipeline_name']}")

    # Call the config function and store the result
    config = ____()
    # Pass the variable to wire the dependency
    ____(config)


pipeline = xcom_example()

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


from airflow.sdk import dag, task

@dag
def xcom_example():

    @task
    def get_config():
        return {"pipeline_name": "daily_etl", "batch_size": 1000}

    @task
    def run_pipeline(config):
        print(f"Running {config['pipeline_name']}")

    # Call the config function and store the result
    config = get_config()
    # Pass the variable to wire the dependency
    run_pipeline(config)

pipeline = xcom_example()

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


從 TaskFlow 函式回傳資料

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

@dag
def mixed_pipeline():

    get_timestamp = BashOperator(
        task_id="get_timestamp",
        bash_command="date +%Y-%m-%d",
    )

    @task
    def log_timestamp(ts_value):
        print(f"Timestamp received: {ts_value}")

    # Access the classic operator's result
    ts = get_timestamp.____
    # Pass the output to wire the dependency
    log_timestamp(____)


pipeline = mixed_pipeline()

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


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

@dag
def mixed_pipeline():

    get_timestamp = BashOperator(
        task_id="get_timestamp",
        bash_command="date +%Y-%m-%d",
    )

    @task
    def log_timestamp(ts_value):
        print(f"Timestamp received: {ts_value}")

    # Access the classic operator's result
    ts = get_timestamp.output
    # Pass the output to wire the dependency
    log_timestamp(ts)


pipeline = mixed_pipeline()

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


將經典 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

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

自動暫停失敗中的 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