在本章中，你會完整認識 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 疑難排解

遺失的 Dag

維護與監控 Airflow 工作流程

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

觸發器與容錯機制

實作 Trigger Rule

加入重試機制

觸發子 Dag

分支

定義一個 @task.branch

分支疑難排解

HITL Operator

加入人工核准

ApprovalOperator 任務

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

# Imports
from airflow.sdk import ____, ____, ____
from airflow.providers.standard.sensors.filesystem import ____
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='____',
  on_success_callback=____(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.____}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
    
  initialized = initialize()
  initialized >> filewatcher
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
      
  initialized = initialize()
  initialized >> filewatcher 
  
sales_etl()

設定 sales ETL pipeline

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @____
  def validate_data():
    sales_path = Variable.____('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = '____'
    if '-09-' in sales_path:
      next_task = '____'
    return next_task
  
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >> validated
  
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >> validated
  
  
sales_etl()

加入每月分支

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
from airflow.providers.standard.operators.hitl import ____

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  @task
  def regular_monthly_task():
    print(f'Processing data and automatically updating dataset')
    
  yearend_approval_task = ____(
    task_id="____",
    subject="Sales data processing - Approval Required",
    body=(
            "Please review the sales data produced by the *validate_data* task "
            "on {{ds}}."
            "Approve to push to the data warehouse, or Reject to halt the run."
    ),
  )
  
  @task
  def push_to_warehouse():
    print(f'Pushing processed data to warehouse')
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >>validated
  
  validated >> [regular_monthly_task(), ____] >> push_to_warehouse()
  
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
from airflow.providers.standard.operators.hitl import ApprovalOperator

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  @task
  def regular_monthly_task():
    print(f'Processing data and automatically updating dataset')
    
  yearend_approval_task = ApprovalOperator(
    task_id="yearend_approval_task",
    subject="Sales data processing - Approval Required",
    body=(
            "Please review the sales data produced by the *validate_data* task "
            "on {{ds}}."
            "Approve to push to the data warehouse, or Reject to halt the run."
    ),
  )
  
  @task
  def push_to_warehouse():
    print(f'Pushing processed data to warehouse')
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >>validated
  
  validated >> [regular_monthly_task(), yearend_approval_task] >> push_to_warehouse()
  
  
sales_etl()

加入人工審核關卡

恭喜你！

控制 DAG 邏輯

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

Intermediate Python

Introduction to Shell

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

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

Create Your Free Account