本章将全面介绍 Apache Airflow 的各个组件，并讲解如何以及为何使用它们。

Apache Airflow 简介

Dag 计数器

运行您的第一个 Dag

Airflow Dags

创建一个简单的 Dag

使用 Dags 与 Airflow Shell

查看 Airflow 命令

在 Airflow 中测试任务

Airflow Web 界面

浏览 Airflow 界面

使用 Airflow UI 检查 Dags

Dag 错误

Airflow 简介

DAG 搞起来！现在是学习实现 Airflow DAG 基础的时候了。通过动手练习，您将学会如何设置并部署 operator、任务和调度。

Airflow operators

为任务添加装饰器

使用 @task.bash 定义 BashOperator

定义任务执行顺序

调度 Dags

通过 Python 调度一个 Dag

解读 Airflow 调度

排查 Dag 运行问题

使用 XCom 在任务之间传递数据

使用 XCom 传递数据

XCom 变量

Airflow 传感器

确定任务的执行顺序

创建一个 FileSensor

感官剥夺

在 Airflow 中构建 DAG

本章将讲解如何使用传感器（sensor）与执行器（executor）等 Airflow 组件为您节省时间，并在监控与排障 Airflow 工作流时发挥作用。

监控、告警与回调

实现回调函数

实现 SmtpNotifier

日志里有什么？

使用 Jinja 进行模板化

使用 Jinja 编写

替换路径

Airflow 变量

答案是 Variable

在 Python 中读取变量

变量不太对劲

在 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()

搭建销售 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. 
  @____
  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 的基础知识，并学习如何在生产环境中实现复杂的数据工程流水线。您还将学习如何使用有向无环图（DAG），自动化数据工程工作流，并以简洁且可复用的方式实现数据工程任务，帮助您更高效、更省心地完成工作。

Intermediate Python

Introduction to Shell

学习使用Airflow和Python中的DAG实现数据工程工作流的调度、错误处理和报告。

Python 中的 Apache Airflow 入门

学习如何实现并调度数据工程工作流。

Airflow Fundamentals

数据工程师 在 Python 中

添加重试

Python 中的 Apache Airflow 入门

练习说明

交互式实操练习