Airflow 诞生的背景

Airflow 的诞生是为了解决管理多个数据管道和工作流的复杂性。在 Airflow 出现之前，许多组织依赖于 cron 任务、自定义脚本以及其他低效的方法来处理由数百万用户频繁生成的大数据。然而，这些解决方案难以维护、缺乏灵活性，并且由于无法可视化正在运行的工作流状态、监控失败点和调试错误，使得管理变得更加困难。

Apache Airflow（简称 Airflow）最初由 Maxime Beauchemin 于 2014 年 10 月在 Airbnb 创建。从一开始，它就是一个开源项目，并于 2015 年 6 月正式托管在 Airbnb 的 GitHub 账号下。到了 2016 年 3 月，该项目进入 Apache 软件基金会（Apache Software Foundation, ASF） 的孵化计划，并最终成为我们今天熟知的 Apache Airflow。

谁在使用 Apache Airflow？

目前，大多数数据专业人士（如数据工程师、机器学习工程师）以及顶级公司（如 Airbnb、Netflix）都在日常工作中使用 Apache Airflow。因此，在本文中，你将学习如何安装和使用 Apache Airflow。

前置要求

要充分理解本教程，你需要具备 Python 编程语言 的基础知识，因为代码示例和 Airflow 框架都是基于 Python 编写的。本教程将帮助你熟悉 Apache Airflow 平台，并指导你完成安装及执行简单任务。

什么是 Apache Airflow？

根据 Apache Airflow 官方文档，它被定义为：

“一个用于开发、调度和监控批处理工作流的开源平台。”

由于 Airflow 的核心基于 Python 框架，用户可以使用它来构建可与几乎所有技术连接的工作流。Airflow 可以部署为一个独立单元（如在本地笔记本电脑上运行），也可以部署在分布式系统上，以支持超大规模的工作流。

Airflow 设计的核心理念是 “编程化”，即所有工作流均以 Python 代码 的形式表示。

Apache Airflow 的核心组件

1. DAG（有向无环图）

DAG（Directed Acyclic Graph，有向无环图）是 一组任务的集合，并以一种显示它们关系和依赖性的方式组织起来。它代表了 工作流的图结构，其中：

• 节点（Node） 代表要执行的任务
• 边（Edge） 表示任务之间的依赖关系

“有向（Directed）”表示任务按固定顺序执行，而“无环（Acyclic）”确保任务不会形成循环，防止重复执行。DAG 以 Python 脚本的形式编写，并存储在 Airflow 的 DAG_FOLDER 中。

2. 任务（Tasks）

任务是 DAG 中的 基本执行单元，它代表单个活动或工作单元。例如：

• 运行 SQL 查询
• 从数据库读取数据
• 进行数据转换等

3. 操作符（Operators）

操作符用于创建 DAG 内的具体任务。每个操作符都会定义任务的类型，例如：

• BashOperator：执行 Bash 命令
• EmailOperator：发送邮件
• PythonOperator：调用 Python 函数

DAG 主要用于 组织任务的执行顺序，而操作符则是 定义任务的具体操作。

4. 调度（Scheduling）

Airflow 通过 调度器（Scheduler） 实现任务调度。调度器会：

• 监控所有可用任务和 DAG
• 在任务的所有依赖项满足时触发任务实例

调度器会在后台运行，不断检查活动任务，以决定它们是否可以被触发。

5. XComs（跨任务通信）

XComs（Cross-Communication 的缩写）用于在任务之间传递数据。它们存储 键（Key）、值（Value）、时间戳（Timestamp） 以及创建 XCom 的任务/DAG。

6. Hooks（钩子）

钩子是 外部平台或资源的接口层，它允许任务轻松连接到外部系统，而无需手动进行身份验证或复杂的通信过程。

7. Web UI

Airflow 提供 Web UI（用户界面），用于可视化监控数据管道和排查问题。

如何在本地运行 Apache Airflow？

在本地安装并运行 Apache Airflow 主要涉及：

• 设置 Airflow 环境
• 初始化数据库
• 启动 Airflow Web 服务器
• 运行 Airflow 任务调度器

按照以下步骤操作：

Step 1：创建 Python 虚拟环境

python3 -m venv airflow_tutorial

Step 2：激活虚拟环境

• Mac/Linux
• 
  
• source airflow_tutorial/bin/activate
• Windows
• 
  
• airflow_tutorial\Scripts\activate

Step 3：安装 Apache Airflow

在激活的虚拟环境中运行：

pip install apache-airflow

Step 4：初始化 Airflow 数据库

airflow db init

这将默认在 ~/airflow 目录下生成 所需的表和配置。

Step 5：创建 Airflow 管理员用户

airflow users create \
    --username admin \
    --firstname FirstName \
    --lastname LastName \
    --role Admin \
    --email admin@example.com

执行此命令后，系统会提示你 输入管理员密码。

Step 6：启动 Airflow Web 服务器

airflow webserver --port 8080

然后，在浏览器中打开 终端显示的 URL，并使用 Step 5 创建的账号登录。

Step 7：启动 Airflow 调度器

在新终端窗口中 重新激活虚拟环境（同 Step 2），然后运行：

airflow scheduler

Step 8：创建并运行自定义 DAG

在 ~/airflow/dags/ 目录下创建 dags_tutorial.py，并写入以下代码：

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


# 创建 DAG
with DAG(dag_id="demo", start_date=datetime(2025, 1, 5), schedule="0 0 * * *") as dag:
    # 定义任务
    hello = BashOperator(task_id="hello", bash_command="echo hello")


    @task()
    def airflow():
        print("airflow")


    # 设置任务依赖
    hello >> airflow()

运行后，DAG 将自动出现在 Airflow Web UI 中。

结论

Apache Airflow 是一个 强大且灵活的开源平台，能够高效管理多个工作流和数据管道。它不仅提供 编程化的开发体验，还具备 强大的 UI 界面，可用于 监控和调试任务。

在本教程中，我们学习了：

• Apache Airflow 的基本概念
• 如何安装和配置 Airflow
• 如何创建和运行一个简单的 DAG

建议你将 Airflow 纳入日常开发实践，以快速熟悉该技术！