本章将全面介绍 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 故障排查

from airflow.sdk import dag, task
#from airflow.providers.standard.sensors.filesystem import FileSensor
from pendulum import datetime

@dag(
    dag_id="execute_report",
    schedule="0 0 * * *",
    start_date=datetime(2026, 3, 20),
)
def execute_report():

    check_for_datafile = FileSensor(
        task_id="check_for_datafile",
        filepath="salesdata_ready.csv",
        mode="poke",
    )

    @task()
    def generate_report():
        print("Generating report...")

    check_for_datafile >> generate_report()

execute_report()

from airflow.sdk import dag, task
from pendulum import datetime

@dag(
    dag_id="sample_dag",
    schedule="0 0 * * *",
    start_date=datetime(2023, 2, 20),
)
def sample_dag():

    @task()
    def sample():
        print("Running sample task...")

    sample()

sample_dag()

from airflow.sdk import dag, task
from airflow.providers.standard.sensors.filesystem import FileSensor
from pendulum import datetime

@dag(
    dag_id="execute_report",
    schedule="0 0 * * *",
    start_date=datetime(2026, 3, 20),
)
def execute_report():

    check_for_datafile = FileSensor(
        task_id="check_for_datafile",
        filepath="salesdata_ready.csv",
        mode="poke",
    )

    @task()
    def generate_report():
        print("Generating report...")

    check_for_datafile >> generate_report()

execute_report()

缺失的 Dag

维护与监控 Airflow 工作流

将所学融会贯通。在最后一章，您将把全部知识应用到 Airflow 中，构建一个可用于生产的高质量工作流。

触发器与容错性

实现 Trigger Rule

添加重试

触发子 Dag

分支

定义一个 @task.branch

分支故障排查

HITL Operator 简介

添加人工审批

ApprovalOperator 任务

创建生产级流水线

搭建销售 ETL 流水线

添加每月分支逻辑

添加人工审批关卡

恭喜您！

控制 DAG 逻辑

现已更新至 Apache Airflow 3.1.6。按计划交付数据常常需要大量手动操作：您要编写脚本、配置复杂的 cron 任务，并不断调整以满足变化的需求——与团队协作时管理起来更棘手。Apache Airflow 通过为工作流加入调度、错误处理和报告功能，可以帮您省去这些麻烦。在本课程中，您将掌握 Apache Airflow 的基础知识，并学习如何在生产环境中实现复杂的数据工程流水线。您还将学习如何使用有向无环图（DAG），自动化数据工程工作流，并以简洁且可复用的方式实现数据工程任务，帮助您更高效、更省心地完成工作。

Intermediate Python

Introduction to Shell

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

在 Airflow 中进行调试与故障排查

Create Your Free Account