Apache Airflow 解析：面向初学者的指南

介绍

现代以数据为驱动的公司都依赖数据管道，从一个地方获取、转换、丰富并移动信息到另一个地方。这些数据管道可能涉及许多步骤（提取原始数据、清理数据、训练机器学习模型、构建仪表板等）。此外，它们必须按照非常特定的顺序运行。工作流编排工具，如 Apache Airflow，帮助确保所有这些步骤在正确的时间和正确的顺序执行。它们还使管道易于监控和管理。

Airflow 于 2014 年在 Airbnb 开发，已成为最受欢迎的开源工作流编排平台之一。成千上万的组织使用它来自动化和监控批处理管道。在本文中，我们将从基础介绍 Apache Airflow。我们将涵盖的主题包括：

• Airflow 是什么，它的功能，以及数据团队为什么使用它。
• Airflow 的核心概念和组件，以及它们是如何协同工作的。
• 管道如何被定义为有向无环图（DAG），以及一个简单DAG运行示例的演练。
• 常见的使用场景，如 ETL/ELT、MLOps 等。

我们希望这将成为面向工程师、学生和刚开始使用工作流编排的团队的一个实用、以示例为驱动的 Apache Airflow 入门介绍。

主要要点

• Apache Airflow 是一个基于 Python 的工作流编排平台，它通过使用 DAG（有向无环图）来管理、调度和监控复杂的数据管道。
• DAG 定义任务及其依赖关系，允许 Airflow 按正确顺序执行每个步骤，并提供内置的重试、日志记录和可观测性功能。
• Airflow 的架构（调度器、执行器、工作节点、元数据数据库、Web 界面） 支持可扩展的分布式执行和对流水线的实时监控。
• Airflow 在批处理工作流（如 ETL/ELT 任务、机器学习训练管道、数据质量检查和定期分析）中表现出色——但 不适合实时或超高频工作负载。
• 类似 Prefect、Dagster、Luigi、Argo Workflows 和 Mage AI 的替代方案 可能更适合，具体取决于需求，例如事件驱动的工作流、原生 Kubernetes 执行或现代开发者友好的界面。

什么是 Apache Airflow？

Apache Airflow 是一个用于创建、调度和监控工作流的开源系统。Airflow 工作流是一个 DAG（有向无环图），它是一系列具有明确定义关系（依赖性）且没有任何循环的任务（任务在上游不能重复执行）。Airflow 的核心理念是“配置即代码”，因此不使用带有拖放界面的 UI，而是编写 Python 脚本来表示工作流。这种方法为工程师提供了显著的灵活性：你可以通过 Airflow 的操作器和 Python 库连接几乎任何技术。同时，你也可以将软件工程的最佳实践（版本控制、测试等）应用到你的管道定义中。

为什么工作流程编排很重要？

关键点在于，在任何数据工作流程中，通常有多个步骤必须按顺序执行。一个简单的例子是 ETL（提取/转换/加载）管道。随着添加更多的任务、依赖关系、分支等等，管道可能很快变得更加复杂，同时还需要按固定的周期运行。在这些情况下，手动干预或简单的调度工具往往不够用。Airflow 通过充当中央协调器来解决这个问题。它有一个任务（节点）及其依赖关系（边）的有向无环图（DAG）。它知道何时执行哪项任务（在上游依赖完成后再运行），包括调度（周期性执行）、错误处理（自动重试、警报）和日志记录。

如果我们把整个数据管道看作一个食谱，那么每个单独的任务就是其中的一个子步骤（切菜、煮、装盘等）。Airflow 就像厨房经理，熟悉所有食谱，并监督每个厨房中进行的不同子步骤，确保所有事情按照正确的顺序和时间发生。

没有工作流编排，你可能有所有的厨师和原材料，但它们没有协调。Airflow 提供了这种协调，使得能够生成所需的“菜肴”（工作流运行）。

数据团队为什么使用 Airflow？

Airflow 在数据工程师、机器学习工程师和 DevOps 团队中变得很受欢迎，因为它能够满足复杂数据管道的多种需求。以下是团队选择 Airflow 的几个主要原因：

• 协调复杂的工作流和依赖关系： Airflow 专为具有大量相互依赖步骤的工作流设计。Airflow 可以轻松声明上游和下游的需求，确保任务按正确的顺序执行。如果任务 C 依赖于任务 A 和 B，Airflow 调度器会识别这些关系，并确保在运行 C 之前，A 和 B 已经执行（可能是并行执行）。Airflow 可以处理复杂的依赖关系、分支逻辑等。
• 集中调度（增强版 Cron）： 它提供了一个调度器，可以让你按照特定的时间表执行任务。你可以使用 Airflow 的调度器定义数十甚至数百个管道，在不同的时间（每小时、每天、每月）触发任务。Airflow 理解任务之间的关系，能够智能地调度和触发任务运行。中央调度器让你无需在不同服务器上编写不同的脚本。

• 可扩展性与分布式执行：Airflow 还可以随着您的工作流进行扩展。它可以配置为在多个工作进程或机器上并行执行任务。它还提供了多种执行器（本文后面将讨论），可以在分布式系统上运行任务，例如在一组节点的 Celery 或 Kubernetes 上。这意味着您可以并行处理大型任务（大数据处理），并让多个管道同时运行。
• 监控、日志记录和警报：它配备了一个丰富的网页界面，可提供对工作流的实时可见性。您可以查看哪些任务已成功完成，哪些任务失败，并查看每个任务的日志输出。您甚至可以通过网页界面手动触发和/或重试任务。

Airflow 的关键组成部分

下面是 Apache Airflow 核心构建模块的简要概述。使用此表作为快速参考，以了解每个组件的作用及功能。

Airflow 的工作原理（架构）

以下是Airflow 如何处理工作流（DAG）的逐步概述：

• 创建 DAG：DAG 作者编写一个 Python 脚本来定义 DAG。这个脚本包括调度设置（例如，“每天上午 9 点运行”）或手动触发设置。它还包含任务的定义（通过操作符或 @task 函数）以及任务依赖关系（例如，task1 >> task2（task2 等待 task1 完成））。Python 文件会保存到 Airflow 的 DAGs 文件夹中，以便 Airflow 能够找到它。Airflow 会定期读取（解析）该文件以加载 DAG 定义。
• 调度和DAG解析：Airflow调度器是一个持续在后台运行的进程。它会检查每个DAG是否应根据其计划或触发器运行。例如，一个计划为“每日”的DAG可以被调度器拾取，并检测到必须在午夜安排一次新的运行。Airflow调度器还会检查DAG中的任务是否准备好运行——意味着其所有上游依赖项已完成且预定时间（逻辑日期）已到。
• 执行任务：每个要执行的任务由调度器交给执行器。执行器决定任务的运行位置和方式。任务可以在本地执行，也可以发送到工作队列。在分布式架构下，任务会被添加到队列中，然后由执行任务的工作进程选择执行。
• 状态更新与错误处理：每个任务实例在完成后会将其状态（成功、失败等）报告回调度器，调度器会使用任务状态更新其元数据数据库。
  状态跟踪使调度器能够知道下一步该做什么——例如，如果任务A已成功完成，它就知道可以调度任务B；如果任务A失败且仍有重试机会，它会安排重试；如果任务A失败且没有剩余重试次数，则将DAG运行标记为失败。如果任务失败，Airflow有逻辑可以在配置了重试的情况下自动重试任务。如果任务在重试后仍继续失败，Airflow可以发送失败警报（例如发送电子邮件或Slack消息等）。所有这些都可以作为DAG或任务参数进行配置（例如重试次数、重试间隔、警报邮件接收人）。
• 监控与干预：Airflow Web UI 提供了上述流程的实时界面。您可以看到一个 DAG 运行的显示（例如，Tree 视图中今天运行的新行），并在任务执行过程中查看任务状态的变化。Graph 视图可视化 DAG 的任务，并显示哪些任务已完成或发生错误。
• 清理和下次运行： 当一个 DAG 运行完成后（所有任务都已成功或被跳过/失败），Airflow 会进入下一个计划运行。关于过去运行的所有元数据仍会保存在数据库中，您可以随时查看历史记录。调度器将继续根据计划或手动触发触发新的运行。此过程会重复进行，直到 Airflow 进程被终止。

理解 Airflow 中的 DAG

Airflow 中的 DAG 定义了执行工作流所需的调度、任务和依赖关系。DAG 不需要知道每个任务的具体内容；它只需要定义这些任务何时以及以何种顺序运行。

任务、操作员和传感器

任务是 Airflow 中的基本工作单元。在 Airflow 中，任务是操作器的实例，操作器是预定义操作的模板。Airflow 提供了多种核心操作器。BashOperator 允许你运行 shell 命令，而 PythonOperator 允许你调用 Python 函数。Airflow 还提供了一个 @task 装饰器，它可以将普通的 Python 函数转换为任务。

除了操作符，Airflow 还提供传感器。传感器是等待某个条件或事件发生后才会成功的任务。它们可以用于等待文件出现或表格被填充等情况。

任务依赖关系

要声明任务之间的依赖关系，你可以使用 >> 和 << 运算符（推荐）或像 set_upstream 和 set_downstream 这样的方法。也可以将任务链接起来，创建交叉下游依赖，甚至动态生成任务列表。一个没有任何依赖关系的 DAG 只是一个独立的任务集合。

DAG 设计的最佳实践

下表列出了一些在设计 DAG 时需要注意的指导原则。

示例：一个简单的 Airflow DAG

没有什么比例子更能巩固概念了。让我们通过一个简单的 Airflow DAG 代码来讲解其各个部分。这个例子将创建一个包含两个顺序任务的小型工作流：

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator

# Define the DAG and its schedule
with DAG(
    dag_id="example_dag",
    description="A simple example DAG",
    start_date=datetime(2025, 1, 1),
    schedule_interval="@daily",  # runs daily
    catchup=False
) as dag:
    # Task 1: Print current date
    task1 = BashOperator(
        task_id="print_date",
        bash_command="date"
    )
    # Task 2: Echo a message
    task2 = BashOperator(
        task_id="echo_hello",
        bash_command="echo 'Hello, Airflow!'"
    )
    # Define dependency: task1 must run before task2
    task1 >> task2

让我们来分解一下这里发生的情况：

• 导入所需的类 DAG（用于定义 DAG）和 BashOperator（用于在任务中运行 shell/bash 命令）。我们还导入 datetime 来指定 DAG 的开始日期。
• 我们使用上下文管理器创建一个 DAG 对象（with DAG(…) as dag:）。为 DAG 提供一个唯一的 ID “example_dag”（在 Airflow 环境中必须唯一）、一个描述（可选，用于帮助人类阅读）、一个 start_date 和一个 schedule_interval。在这里，我们设置了内置的 schedule_interval 预设 (@daily)，以及 start_date=datetime(2025,1,1)。我们还设置了 catchup=False，以防止在 start_date 到今天之间的任何缺失运行被回填（我们只希望它从现在开始每日运行）。整个块创建了一个 DAG，使 Airflow 可以每天在一天开始时运行它。
• 然后我们在 DAG 上下文中定义两个任务：
• task1 使用 BashOperator 定义，用于执行一个 bash 命令（date）。task_id 为 “print_date”，将在 UI 和日志中用于标识此任务。当它运行时，该任务将简单地打印当前的日期/时间（到任务的日志中）。
• task2 也 使用了一个 BashOperator，task_id=“echo_hello”，命令为 echo ‘Hello, Airflow!’。这将打印一条漂亮的消息。这些任务现在只是被定义，还没有被链接。
• 最后一行（task1 >> task2）指定了一个依赖关系（流程）：task1 流向 task2。这意味着 task1 将在 task2 可以考虑调度运行之前运行并成功完成。在 DAG 图中，图形上会有一条从 task1 指向 task2 的箭头。

在这个例子中需要注意几件事：

• 我们只是为了简便起见使用了 BashOperator，但我们也可以使用包含多种类型操作符的 DAG。我们可以使用 PythonOperator 调用 Python 函数，或者使用 SimpleHttpOperator 访问 API，等等。
• 使用 DAG(…) as dag: 的上下文模式是创建 DAG 实例的常用习惯用法。所有在其中创建的任务都会自动与该 DAG 关联。
• catchup=False 是一个参数，常用于许多 start_date 在过去的 DAG 中，当你不希望它为所有过去的日期运行时使用。如果我们保持 catchup=True（默认值），并且今天是 2025 年 1 月 10 日，Airflow 会尝试运行 DAG 来覆盖 1 月 1、2、3 … 一直到 1 月 10 日的所有日期，补上之前错过的日子。在这里我们为了简化起见关闭了它。
• 此示例使用了 BashOperator 的相对导入路径。在 Airflow 2 中，许多操作符已被移到 airflow.providers 下。此示例假设这些提供程序已安装（BashOperator 本身就在 Airflow 核心中）。

这个简单的示例可以很容易地扩展——你可以添加更多任务（比如先进行 API 调用，然后将结果加载到数据库，再发送通知）。你还可以添加更复杂的逻辑（比如分支——Airflow 支持使用 BranchOperator 进行 if/else 风格的分支）。

随着您的 DAG 变得越来越大，Airflow 的用户界面和日志记录在了解发生了什么方面变得非常有价值。在这个例子中，print_date 的日志将包含打印的系统日期，而 echo_hello 的日志将包含“Hello, Airflow!”。如果 print_date 失败了（例如在假设的场景中找不到 date 命令），那么 echo_hello 将永远不会被执行（由于依赖关系），并且该 DAG 运行将被标记为失败。然后，您可以检查日志，修复问题，并重新运行该任务或整个 DAG。

Airflow的常见使用场景

下面是常见 Apache Airflow 使用案例的结构化总结。表格突出显示了每个使用案例涉及的内容以及 Airflow 在该情境中如何提供价值。

何时不使用 Airflow

虽然 Airflow 功能强大，但它并不适合所有情况。了解它的局限性可以帮助你选择正确的编排方法：

• 实时或流式工作负载：Airflow 是面向批处理的，具有明确的开始和结束时间。它不适合长时间运行、事件驱动或流式工作负载。如果你正在接收连续的事件流（例如用户点击、物联网传感器等）并需要以非常低的延迟实时处理这些事件，Airflow 并不是最佳选择。
• 高频或许多小任务： 这是 Airflow 可能无法正常工作，或者至少效率不高的另一种情况。Airflow 任务存在一些开销（例如在数据库跟踪等方面），因此如果你需要每隔几秒触发一次任务或有成千上万的微小任务，Airflow 可能无法很好地扩展。
• 纯事件驱动的工作流： 这类似于流处理的用例，但用于应由事件触发的情况。“每当事件 X 发生时，执行 Y” 是 Airflow 的一个完全有效的用例，但如果唯一触发条件是某个事件（例如，“每当文件到达 S3 存储桶时，执行 X”），那么 Airflow 在这种情况下可能不是最轻量的解决方案。Airflow 通常使用传感器（例如 S3KeySensor）来“感知”文件或事件的存在。传感器有其缺点，因为它们依赖轮询。传感器的问题包括可能会对已有文件触发，并且未必能及时响应事件（例如，它们按间隔时间轮询）。

总之，Airflow 适合周期性、批量处理或定义和管理较为复杂的工作流。如果你的使用场景需要实时或连续处理、超高频作业，或者非常简单、不需要 Airflow 额外开销的工作流，那么 Airflow 可能不太适合。在这种情况下，你可能需要考虑其他选项或简化你的解决方案。

Airflow 的替代方案

近年来，工作流编排领域迅速发展，虽然 Airflow 是领导者，但还有几个值得了解的替代工具：

Airflow 并不是唯一的选择，最佳方案取决于你的具体使用场景。Luigi、Prefect 和 Dagster 通常被列为同一领域（基于 Python 的工作流编排器）中的主要开源备选方案。如果你觉得 Airflow 老旧的界面或其他限制令人不便，应考虑评估这些工具。Prefect 试图成为一个“更好”或“更简单”的 Airflow，Dagster 则努力成为一个更“结构化”的编排器，重点关注数据资产，而 Luigi 是 Airflow 的一个更简单的前身。另一方面，如果你的使用场景差异很大（例如实时流处理或完全云原生），你可能根本不使用 Airflow，而是直接考虑流处理平台或托管编排服务。

最终思考

Apache Airflow 是一个开源平台，用于编排复杂的计算工作流和数据处理管道。它最初由 Airbnb 开发，在数据工程和机器学习社区中获得了广泛的认可。Airflow 的组件包括调度器、执行器、工作节点、元数据数据库和基于网页的用户界面。这使团队能够轻松运行和维护可投入生产的 ETL、MLOps 和基础设施管道。Airflow 的 DAG（有向无环图）抽象使工作流保持清晰、可测试并且可长期维护。该平台的社区构建的操作器和提供者生态系统也简化了依赖任何技术或工具的工作流编排。

不过，这并不是一刀切的解决方案。它最适合批量或微批量管道，并且用户必须熟悉 Python。对于流处理或高频率工作流，有比 Airflow 更适合的其他解决方案。对于数据科学家或更喜欢编写声明式管道的团队，可以考虑 Airflow 的替代工具。

Airflow 仍在积极开发中，并且随着时间推移不断添加新功能，例如事件驱动调度、数据集和资产。本指南介绍的概念和最佳实践应能帮助你，无论背景如何，都能理解工作流编排的工作原理，以及什么时候应使用（或不使用）Airflow。希望这能成为你进一步探索这个丰富世界的垫脚石。

参考文献

• Apache Airflow
• 什么是 Airflow？
• Apache Airflow：使用案例、架构及成功的6个技巧
• 使用 Docker 入门 Airflow —— 云端数据管道（I）
• Airflow 的 8 个替代方案及使用场景
• 架构概览
• 调度程序

感谢您与 Linux-Console.net 社区一起学习。