你將先認識 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

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

自動暫停失敗中的 Dag

選對失敗處理策略

可延後執行的運算元

為正式環境設定感測器

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

何時使用 poke、reschedule 或 deferrable

測試 Airflow 程式碼

撰寫 Dag 完整性測試

為任務函式撰寫單元測試

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

用 Task Groups 組織複雜的 Dags

將相關任務分組

使用顯示名稱巢狀化 Task Group

依領域整理任務

為 Dags 做好上線準備

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

執行 SQL 工作負載

from airflow.sdk import dag, task
# Import SQLExecuteQueryOperator
from airflow.providers.common.sql.operators.sql import ____


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

    # Aggregate raw orders into the daily summary table
    aggregate = ____(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="""
            INSERT INTO daily_summary (order_date, total_orders, total_revenue)
            SELECT order_date, COUNT(*), SUM(revenue)
            FROM raw_orders
            GROUP BY order_date
        """,
    )

    # Read back the results to verify
    read_results = SQLExecuteQueryOperator(
        task_id="____",
        conn_id="____",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    aggregate >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline()

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


from airflow.sdk import dag, task
# Import SQLExecuteQueryOperator
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


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

    # Aggregate raw orders into the daily summary table
    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="""
            INSERT INTO daily_summary (order_date, total_orders, total_revenue)
            SELECT order_date, COUNT(*), SUM(revenue)
            FROM raw_orders
            GROUP BY order_date
        """,
    )

    # Read back the results to verify
    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    aggregate >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline()

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


使用 SQL 彙總銷售資料

DELETE FROM daily_summary WHERE order_date = '____';
INSERT INTO daily_summary (order_date, total_orders, total_revenue)
SELECT order_date, COUNT(*), SUM(revenue)
FROM raw_orders
WHERE order_date = '____'
GROUP BY order_date;


from airflow.sdk import dag, task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


# Set the template search path, so Airflow finds SQL files
@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/____")
def sales_pipeline_v2():

    # Reference the external SQL file instead of inline SQL
    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="____.sql",
    )

    # Read back the results to verify
    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    aggregate >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline_v2()

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


DELETE FROM daily_summary WHERE order_date = '{{ ds }}';
INSERT INTO daily_summary (order_date, total_orders, total_revenue)
SELECT order_date, COUNT(*), SUM(revenue)
FROM raw_orders
WHERE order_date = '{{ ds }}'
GROUP BY order_date;


from airflow.sdk import dag, task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


# Set the template search path, so Airflow finds SQL files
@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v2():

    # Reference the external SQL file instead of inline SQL
    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    # Read back the results to verify
    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    aggregate >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline_v2()

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


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

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

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

# Import CronPartitionTimetable and Asset
from airflow.sdk import dag, task, ____, ____
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

# Define the daily_sales asset
daily_sales = ____("daily_sales")


# Schedule with CronPartitionTimetable
@dag(schedule=____("0 0 * * *", ____="UTC"), template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v3():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        # Emit a partitioned asset event
        outlets=[____],
        sql="aggregate_sales.sql",
    )

    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    load >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline_v3()

if __name__ == "__main__":
    # Workaround: dag.test() does not set a partition_key.
    # We inject one so the CronPartitionTimetable Dag can run.
    from unittest.mock import patch
    from airflow.serialization.definitions.dag import SerializedDAG

    _orig_create_dagrun = SerializedDAG.create_dagrun

    def _create_dagrun_with_partition_key(self, *args, **kwargs):
        kwargs["partition_key"] = "2026-04-23T00:00:00"
        return _orig_create_dagrun(self, *args, **kwargs)

    with patch.object(SerializedDAG, "create_dagrun", _create_dagrun_with_partition_key):
        pipeline.test()


# Import CronPartitionTimetable and Asset
from airflow.sdk import dag, task, CronPartitionTimetable, Asset
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

# Define the daily_sales asset
daily_sales = Asset("daily_sales")


# Schedule with CronPartitionTimetable
@dag(schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"), template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v3():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        # Emit a partitioned asset event
        outlets=[daily_sales],
        sql="aggregate_sales.sql",
    )

    read_results = SQLExecuteQueryOperator(
        task_id="read_results",
        conn_id="duckdb_analytics",
        sql="SELECT * FROM daily_summary ORDER BY order_date",
    )

    @task
    def send_report(results):
        for row in results:
            print(f"Date: {row[0]} | Orders: {row[1]} | Revenue: ${row[2]:.2f}")

    load >> read_results >> send_report(read_results.output)


pipeline = sales_pipeline_v3()

if __name__ == "__main__":
    # Workaround: dag.test() does not set a partition_key.
    # We inject one so the CronPartitionTimetable Dag can run.
    from unittest.mock import patch
    from airflow.serialization.definitions.dag import SerializedDAG

    _orig_create_dagrun = SerializedDAG.create_dagrun

    def _create_dagrun_with_partition_key(self, *args, **kwargs):
        kwargs["partition_key"] = "2026-04-23T00:00:00"
        return _orig_create_dagrun(self, *args, **kwargs)

    with patch.object(SerializedDAG, "create_dagrun", _create_dagrun_with_partition_key):
        pipeline.test()


建立分割區的 pipeline

# Import PartitionedAssetTimetable and StartOfDayMapper
from airflow.sdk import dag, task, Asset, CronPartitionTimetable, ____, ____
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

daily_sales = Asset("daily_sales")


# --- Upstream: partitioned sales load (from previous exercise) ---
@dag(schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"), template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v3():
    SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        outlets=[daily_sales],
        sql="aggregate_sales.sql",
    )

upstream = sales_pipeline_v3()


# --- Downstream: report triggered by partitioned asset event ---
# Schedule on partitioned daily_sales events with StartOfDayMapper
@dag(schedule=____(
    assets=____,
    partition_mapper_config={daily_sales: ____()}
))
def sales_report():

    @task
    def generate_report(dag_run=None):
        # Read the partition key from the Dag run
        partition_key = dag_run.____
        print(f"Generating report for partition: {partition_key}")

    generate_report()

downstream = sales_report()


if __name__ == "__main__":
    # Workaround: dag.test() does not set a partition_key.
    # We inject one so the CronPartitionTimetable Dag can run.
    from unittest.mock import patch
    from airflow.serialization.definitions.dag import SerializedDAG

    _orig_create_dagrun = SerializedDAG.create_dagrun

    def _create_dagrun_with_partition_key(self, *args, **kwargs):
        kwargs["partition_key"] = "2026-04-23T00:00:00"
        return _orig_create_dagrun(self, *args, **kwargs)

    with patch.object(SerializedDAG, "create_dagrun", _create_dagrun_with_partition_key):
        upstream.test()


# Import PartitionedAssetTimetable and StartOfDayMapper
from airflow.sdk import dag, task, Asset, CronPartitionTimetable, PartitionedAssetTimetable, StartOfDayMapper
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

daily_sales = Asset("daily_sales")


# --- Upstream: partitioned sales load (from previous exercise) ---
@dag(schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"), template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v3():
    SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        outlets=[daily_sales],
        sql="aggregate_sales.sql",
    )

upstream = sales_pipeline_v3()


# --- Downstream: report triggered by partitioned asset event ---
# Schedule on partitioned daily_sales events with StartOfDayMapper
@dag(schedule=PartitionedAssetTimetable(
    assets=daily_sales,
    partition_mapper_config={daily_sales: StartOfDayMapper()}
))
def sales_report():

    @task
    def generate_report(dag_run=None):
        # Read the partition key from the Dag run
        partition_key = dag_run.partition_key
        print(f"Generating report for partition: {partition_key}")

    generate_report()

downstream = sales_report()


if __name__ == "__main__":
    # Workaround: dag.test() does not set a partition_key.
    # We inject one so the CronPartitionTimetable Dag can run.
    from unittest.mock import patch
    from airflow.serialization.definitions.dag import SerializedDAG

    _orig_create_dagrun = SerializedDAG.create_dagrun

    def _create_dagrun_with_partition_key(self, *args, **kwargs):
        kwargs["partition_key"] = "2026-04-23T00:00:00"
        return _orig_create_dagrun(self, *args, **kwargs)

    with patch.object(SerializedDAG, "create_dagrun", _create_dagrun_with_partition_key):
        upstream.test()


在分割資產上進行排程

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

資料品質檢查

from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import (
    SQLExecuteQueryOperator,
    # Import SQLColumnCheckOperator
    ____,
)


@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v4():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    # Validate column-level data quality after loading
    check_columns = ____(
        task_id="check_columns",
        conn_id="duckdb_analytics",
        table="daily_summary",
        column_mapping={
            "total_revenue": {"____": {"greater_than": 0}},
            "total_orders": {"____": {"equal_to": 0}},
        },
    )

    load >> check_columns


pipeline = sales_pipeline_v4()

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


from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import (
    SQLExecuteQueryOperator,
    # Import SQLColumnCheckOperator
    SQLColumnCheckOperator,
)


@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v4():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    # Validate column-level data quality after loading
    check_columns = SQLColumnCheckOperator(
        task_id="check_columns",
        conn_id="duckdb_analytics",
        table="daily_summary",
        column_mapping={
            "total_revenue": {"min": {"greater_than": 0}},
            "total_orders": {"null_check": {"equal_to": 0}},
        },
    )

    load >> check_columns


pipeline = sales_pipeline_v4()

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


加入欄位層級的品質檢查

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


@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v5():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    check_columns = SQLColumnCheckOperator(
        task_id="check_columns",
        conn_id="duckdb_analytics",
        table="daily_summary",
        column_mapping={
            "total_revenue": {"min": {"greater_than": 0}},
            "total_orders": {"null_check": {"equal_to": 0}},
        },
    )

    # Add table-level quality check
    check_table = ____(
        task_id="check_table",
        conn_id="duckdb_analytics",
        table="daily_summary",
        # Verify that the table has at least one row
        checks={
            "row_count_check": {"check_statement": "COUNT(*) ____ ____"},
        },
    )

    load >> check_columns >> check_table


pipeline = sales_pipeline_v5()

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


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


@dag(schedule="@daily", template_searchpath="/home/repl/workspace/include/sql")
def sales_pipeline_v5():

    load = SQLExecuteQueryOperator(
        task_id="load_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

    check_columns = SQLColumnCheckOperator(
        task_id="check_columns",
        conn_id="duckdb_analytics",
        table="daily_summary",
        column_mapping={
            "total_revenue": {"min": {"greater_than": 0}},
            "total_orders": {"null_check": {"equal_to": 0}},
        },
    )

    # Add table-level quality check
    check_table = SQLTableCheckOperator(
        task_id="check_table",
        conn_id="duckdb_analytics",
        table="daily_summary",
        # Verify that the table has at least one row
        checks={
            "row_count_check": {"check_statement": "COUNT(*) > 0"},
        },
    )

    load >> check_columns >> check_table


pipeline = sales_pipeline_v5()

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


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

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

從開發到上線

從 CLI 觸發一條 pipeline

回填歷史資料

選項

Build

Observe

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


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

Create Your Free Account