Airflow实战指南：如何从零开始编写你的第一个DAG

2026-02-15 11:15:11 0条评论 1次阅读 0人点赞

在数据工程的世界里，你是否曾经为管理无数杂乱无章的数据处理脚本而感到头疼？随着业务逻辑的日益复杂，仅仅依靠简单的定时任务往往无法满足我们对依赖关系管理、错误重试以及监控告警的需求。这就是为什么我们需要 Apache Airflow —— 一个强大的工作流管理平台。

在 Airflow 的核心概念中，最重要的莫过于 DAG（有向无环图）。简单来说，DAG 就是我们数据管道的蓝图，它将所有任务按照特定的顺序和依赖关系组织在一起。在这篇文章中，我们将作为探索者，深入探讨如何在 Airflow 中从零开始创建我们的第一个 DAG。我们不仅要编写代码，更要理解其背后的设计哲学，以便你能将这种能力应用到实际的数据工程项目中去。

核心概念解析：什么是 DAG？

在开始敲代码之前，让我们先达成一个共识。在 Airflow 中，DAG（Directed Acyclic Graph，有向无环图） 是指包含了一组按特定顺序执行的任务的集合。你可以把它想象成一条数据流水线，数据从一端流入，经过各个工序的处理，最后从另一端流出。在图论中，这种结构具有两个关键特征：

没有循环： 任务执行完毕后不能回头再次执行自己（除非是重新调度），这保证了流程最终能够结束。
边是有方向的： 连接任务的箭头具有方向性，指明了数据的流向和执行的先后顺序。

为了更好地与 Airflow 交互，我们需要掌握以下几个核心术语：

Operator（操作符）： 这是 DAG 中任务的具体实现模板。你可以把它看作是一个“类”，它定义了任务具体要做什么。例如，INLINECODEde9ce2ea 用于执行 bash 命令，INLINECODE4f83fce4 用于运行 Python 函数。在 DAG 文件中，当我们实例化一个 Operator 时，我们就得到了一个具体的任务。
Task（任务）： 任务是 Operator 的一个实例。它是 DAG 中的节点，代表了一个具体的工作单元。正如我们在下文中会看到的，当我们调用 DummyOperator(task_id=‘start‘) 时，我们就创建了一个名为 ‘start‘ 的任务。
Task Instance（任务实例）： 这是任务在某个特定时间点（Logical Date）的一次运行。如果任务是“类”，那么任务实例就是“对象”。任务实例包含运行状态，如“成功”、“失败”或“重试中”等。这是 Airflow 调度器实际处理的最小单元。
Dependencies（依赖关系）： 这是指任务之间的关系，通常通过位运算符 INLINECODEa8eb7a29 或 INLINECODE0d89bcb8 来定义。它们构成了 DAG 的边，决定了任务执行的先后顺序。

准备工作：创建 DAG 文件的五个步骤

编写一个 DAG 文件本质上就是编写一个 Python 脚本。无论我们的业务逻辑多么复杂，构建 DAG 的骨架通常遵循以下五个标准步骤。一旦你掌握了这个流程，创建复杂的数据管道将变得轻而易举。

导入模块： 导入构建 DAG 所需的 Python 库和 Airflow 类。
定义默认参数： 创建一个字典，配置适用于 DAG 中所有任务的元数据（如所有者、开始时间等）。
实例化 DAG： 使用 DAG 上下文管理器或对象来定义 DAG 的全局属性（如 ID、调度频率）。
定义任务： 初始化 Operators 以创建具体的任务实例。
设置依赖： 建立任务之间的连接关系，确定执行流程。

下面，让我们一步步通过实际代码来实现这个流程。

#### 步骤 1：导入必要的模块

在创建 DAG 时，正确的导入至关重要。这不仅是为了让代码运行，更是为了确保我们拥有构建结构所需的所有组件。首先，我们需要从 INLINECODE2ed88c12 包中导入 INLINECODEc8002793 对象。此外，为了处理时间相关的调度逻辑，我们需要 INLINECODEdc55cd93 和 INLINECODEae3b001e。最后，我们需要导入具体的操作符。在这个示例中，我们将使用 DummyOperator，它是一个“空”操作符，不做任何实际工作，但在测试和构建 DAG 结构时非常有用。

# 导入 DAG 对象，它是我们构建工作流的基础
from airflow import DAG
# 导入 datetime 和 timedelta 模块，用于设置调度时间
from datetime import timedelta, datetime
# 导入 DummyOperator，用于创建占位任务
from airflow.operators.dummy_operator import DummyOperator

#### 步骤 2：配置默认参数

最佳实践是创建一个名为 default_args 的字典。这个字典作为 DAG 的元数据容器，允许我们将相同的参数批量应用到所有任务上，而无需为每个任务单独设置。这不仅减少了代码重复，还使得全局修改（如修改所有者的名字）变得非常容易。

# 初始化默认参数字典
default_args = {
    ‘owner‘: ‘data_team‘,  # 指定 DAG 的所有者，这对于权限管理和通知非常重要
    ‘start_date‘: datetime(2022, 11, 12),  # DAG 的开始日期，调度器会依据此日期计算第一次运行时间
    ‘retries‘: 1,  # 如果任务失败，自动重试的次数（这是一个非常实用的参数）
    ‘retry_delay‘: timedelta(minutes=5),  # 两次重试之间的等待时间
}

关键见解： INLINECODE7029de85 可能是 Airflow 中最让人困惑的概念之一。你需要理解，Airflow 根据 INLINECODE69081077 加上 schedule_interval 来计算数据周期，而不是单纯的任务触发时间。务必设置一个过去的日期，以确保 Airflow 能够立即开始调度。

#### 步骤 3：实例化 DAG 对象

有了参数之后，我们就可以创建 DAG 对象了。我们需要给它一个唯一的 INLINECODE4068b94c。这个 ID 将在 Airflow 的 Web UI 中显示，所以最好起一个有意义的名字。此外，我们可以指定 INLINECODEebf728a8 和 catchup 参数。

# 创建 DAG 对象
dag = DAG(
    dag_id=‘my_first_dag‘,  # DAG 的唯一标识符，请确保在整个 Airflow 环境中是唯一的
    default_args=default_args,  # 应用我们刚才定义的默认参数
    schedule_interval=‘@once‘,  # 调度间隔：‘@once‘ 表示只运行一次，不进行循环调度
    catchup=False,  # 关闭回填：防止 Airflow 尝试运行自 start_date 以来的所有历史任务
    tags=[‘tutorial‘, ‘demo‘]  # 添加标签有助于在 UI 中对 DAG 进行分类和过滤
)

深入理解 INLINECODE4bdf0678： 这是一个至关重要的设置。默认情况下，Airflow 的 INLINECODE0de784a3 为 INLINECODE324dbaa4。如果你今天写了一个 DAG，并把 INLINECODE4d2f804c 设为去年，Airflow 会试图立即补跑过去这一年的所有数据！对于初学者来说，通常建议将 INLINECODEe9a86463 设置为 INLINECODE88aa453a，或者明确理解你的业务是否需要历史数据回填。

#### 步骤 4：创建任务

任务是 Operator 的具体实例。每个任务必须有一个唯一的 task_id。在构建数据管道时，通常会定义一个开始任务和一个结束任务，或者中间的各个处理步骤。让我们创建两个简单的任务：

# 创建第一个任务：start
# 这个任务标志着整个工作流的开始
start = DummyOperator(
    task_id=‘start‘,  # 任务在 DAG 中的唯一标识
    dag=dag  # 将任务挂载到我们创建的 DAG 对象上
)

# 创建第二个任务：end
# 这个任务标志着工作流的结束
end = DummyOperator(
    task_id=‘end‘,
    dag=dag
)

如果我们此时刷新 Airflow 的 Web UI 并进入“Graph View”，我们会看到这两个任务独立地悬浮在画布上，因为它们之间还没有建立联系。

#### 步骤 5：设置任务依赖关系

这是 DAG 设计中最有趣的部分。我们需要告诉 Airflow 哪个任务先运行，哪个后运行。Airflow 提供了两种非常直观的方式来设置依赖：

使用位移运算符： 这是 Python 特有的语法糖，简洁明了。
使用 INLINECODEf64ae00c / INLINECODEea11083d 方法： 这是传统的面向对象写法。

# 设置依赖关系：start >> end
# 这意味着 start 是上游任务，end 是下游任务
# 也就是说，start 必须成功完成后，end 才会开始运行
start >> end

这段代码等同于 INLINECODE7b137542 或 INLINECODE50c997d1。一旦执行了这一步，在 UI 中你会看到一条连接 INLINECODE08350475 和 INLINECODE182f5640 的箭头，这标志着你的第一个 DAG 已经成功创建并定义完毕。

进阶技巧与最佳实践

既然我们已经了解了基础，让我们来看看如何让我们的 DAG 更加健壮和专业。

#### 1. 利用 Context Manager（上下文管理器）简化代码

在上面的例子中，我们必须显式地将 INLINECODEc18a9673 传递给每个任务。这不仅繁琐，还容易遗漏。Airflow 提供了一种更优雅的写法，使用 INLINECODE51f567c3 语句。在 with 代码块内部创建的所有任务都会自动挂载到该 DAG 上。

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime

default_args = {
    ‘owner‘: ‘airflow‘,
    ‘start_date‘: datetime(2023, 1, 1),
}

# 使用 with 语句创建 DAG
with DAG(
    dag_id=‘context_manager_dag‘,
    default_args=default_args,
    schedule_interval=‘@daily‘
) as dag:
    # 注意这里不需要再传递 dag=dag 参数！
    task_1 = DummyOperator(task_id=‘task_1‘)
    task_2 = DummyOperator(task_id=‘task_2‘)
    
    task_1 >> task_2

#### 2. 定义复杂的多路径依赖

在实际的数据工程场景中，依赖关系很少是一条直线。我们通常会有并行处理任务，最后再汇聚。让我们看一个更复杂的例子：提取、加载和转换（ETL）流程。

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime

with DAG(
    dag_id=‘etl_pipeline_dag‘,
    start_date=datetime(2023, 1, 1),
    schedule_interval=‘@daily‘,
    catchup=False
) as dag:

    # 0. 开始任务
    start = DummyOperator(task_id=‘start‘)

    # 1. 并行执行的提取任务
    extract_sales = DummyOperator(task_id=‘extract_sales_data‘)
    extract_logs = DummyOperator(task_id=‘extract_user_logs‘)
    
    # 2. 数据清洗（依赖于所有提取任务）
    clean_data = DummyOperator(task_id=‘clean_data‘)

    # 3. 加载数据
    load_to_warehouse = DummyOperator(task_id=‘load_to_warehouse‘)
    
    # 4. 报告生成
    generate_report = DummyOperator(task_id=‘generate_report‘)

    # 5. 结束任务
    end = DummyOperator(task_id=‘end‘)

    # --- 定义依赖关系 ---
    # start 必须最先执行
    # extract_sales 和 extract_logs 并行执行
    start >> [extract_sales, extract_logs]
    
    # 清洗工作必须等待所有提取完成
    [extract_sales, extract_logs] >> clean_data
    
    # 清洗完成后，加载和报告可以并行（假设它们不冲突）
    clean_data >> [load_to_warehouse, generate_report]
    
    # 最后汇聚到结束任务
    [load_to_warehouse, generate_report] >> end

在这个例子中，我们演示了如何拆分和合并任务流。使用列表 [task_a, task_b] >> task_c 是 Airflow 中非常强大的功能，它允许你表达“当且仅当 A 和 B 都完成后，才运行 C”的逻辑。

常见错误与排查指南

在使用 Airflow 开发 DAG 的过程中，你可能会遇到一些常见的问题。这里有几个专业的建议来帮助你避免陷阱：

循环依赖陷阱： 切记不能创建循环依赖（例如 A >> B >> A）。Airflow 会拒绝加载这样的 DAG，并在调度器日志中报错。如果你的 DAG 没有出现在 UI 列表中，首先应该检查是否存在循环。
不要使用动态 DAG ID： 不要使用 INLINECODEf27b9181 来生成 INLINECODEd716b57b。每次 Airflow 解析 Python 文件时，都会生成一个新的 DAG ID，这会迅速污染你的数据库并导致性能问题。
执行间隔的理解： 许多人误以为 INLINECODE1e46c1e8（每天午夜运行）意味着“作业在午夜结束”。实际上，根据 Airflow 的 INLINECODE7afecc0e 逻辑，这个作业通常是在第二天午夜触发的，处理的是前一个周期的数据。理解这一点对于正确配置窗口时间至关重要。

总结与下一步

通过这篇文章，我们不仅学习了如何在 Airflow 中创建第一个 DAG，还深入探讨了其背后的核心概念：DAG、Operators 和 Task Instances。我们从最基础的代码结构开始，逐步学习了如何设置参数、定义依赖关系，甚至进阶到了复杂的多路径 ETL 流程设计。

你现在拥有的能力不仅仅是写一个 Python 脚本，而是设计一个可扩展、可监控且健壮的数据自动化系统。你可以尝试将 INLINECODE3cbc15dd 替换为实际的 INLINECODE0970f4b5（运行脚本）或 PythonOperator（运行数据清洗函数），来处理真实的业务数据。

让我们总结一下关键点：

DAG 是数据管道的结构定义，必须无环且有序。
Operator 是任务的具体实现，是工作的执行者。
Default Args 帮助我们统一管理任务属性，减少冗余。
Dependencies 定义了业务逻辑的流向，使用 INLINECODE7e23c3c8 和 INLINECODE6c5d3d5a 是最直观的表达方式。
Context Manager (with 语句) 是编写清晰 DAG 的最佳实践。

在接下来的学习旅程中，我建议你深入研究 Airflow 的 XCom（跨任务通信机制）以及 Variables（全局变量配置），这将帮助你在任务之间传递数据，从而构建真正复杂的数据处理应用。祝你编码愉快！

投稿给我们	如何建站？
vps是什么？	如何安装宝塔？
如何通过博客赚钱？	便宜wordpress托管方案
免费wordpress主题	这些都是免费方案

豆丁博客