Apache Airflow 3 DAG Bundles 实战指南：用本地目录、Git、S3 与 GCS 实现 DAG 的版本化交付

Apache Airflow 3 DAG Bundles 实战指南：用本地目录、Git、S3 与 GCS 实现 DAG 的版本化交付

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow

本指南围绕 Apache Airflow 3 引入的Dag Bundle（DAG 包）机制展开，系统讲解它相较于 Airflow 2dags_folder的核心演进、四种内置 Bundle 类型（Local / Git / S3 / GCS）的配置方法、重跑版本行为（rerun_with_latest_version）的控制逻辑，以及如何基于BaseDagBundle编写自定义 Bundle。读完本文，你将能在一套 Airflow 部署中同时挂载多种 DAG 来源，并为 DAG Run 实现"同一份代码跑完整次运行"的可复现交付。

DAG Bundle 是什么，为什么它是 Airflow 3 的基石

Dag Bundle 是一个或多个 DAG 文件及其关联资源（辅助 Python 脚本、配置文件等）的集合，它可以来自本地目录、Git 仓库或 S3/GCS 等外部系统。管理员可以为一套 Airflow 部署定义多个 Bundle，让 DAG 按逻辑单元组织管理；由于 Bundle 处于"包"这一更高抽象层级，天然支持对 DAG 运行所需的一切内容做整体版本化。

Airflow 2 及更早版本中，DAG 必须集中放在本机磁盘的单一dags_folder里，把代码"搬进去"完全依赖部署管理员手工操作。DAG Bundle 与这种Dags folder模式相似但更强大：

• 版本控制（Version Control）：Bundle 支持版本化后，DAG Run 在整个运行期间都使用同一份代码——即使运行中途 DAG 被更新，本次运行也不受影响，重跑时仍可选择原始版本复现问题；
• 可扩展性（Scalability）：把大量 DAG 组织成逻辑单元，Airflow 能更高效地管理它们；
• 灵活性（Flexibility）：无缝对接 Git、对象存储等外部系统来获取 DAG 源码，摆脱"必须放到本机固定目录"的束缚。

从源码结构看，整个机制的核心抽象是 BaseDagBundle 基类，所有 Bundle 类型（无论内置还是自定义）都实现这套接口；DAG Bundle 同时被DAG Processor（用于持续解析并保持 DAG 最新）与Worker（用于按某个特定版本运行任务）两种角色使用，二者使用模式不同：Processor 始终只跟踪最新版本，而单个 Worker 上可能同时存在同一 Bundle 的多个版本副本。

内置 Bundle 类型一览

版本化能力通过基类上的类属性supports_versioning: bool = False声明：LocalDagBundle在 local.py 中显式置为False，GitDagBundle在 git.py 中置为True。

配置 DAG Bundles：dag_bundle_config_list

所有 Bundle 都在配置项[dag_processor] dag_bundle_config_list中声明，可以在此配置一个或多个 Bundle。该配置项是一个 JSON 列表，其中每个元素是一个字典，包含name、classpath、kwargs三个键；默认值为一个指向 Dags 文件夹的LocalDagBundle，以保持与 Airflow 2 一致的开箱行为：

[dag_processor] dag_bundle_config_list = [ { "name": "dags-folder", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": { "path": "/opt/airflow/dags" } } ]

对LocalDagBundle而言，唯一的 kwarg 是path，省略时默认取[core] dags_folder的值。这一定义与 config.yml 中记录的默认配置完全一致：

default: >- [ {{ "name": "dags-folder", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {{}} }} ]

配置加载与合法性校验（源码级）

在 DagBundlesManager.parse_config 中，配置经过严格的解析与校验：

• 配置值必须是一个列表，否则抛出AirflowConfigException；
• 每个元素必须是字典，并通过 Pydantic 模型_ExternalBundleConfig（含name、classpath、kwargs、可选的team_name）校验；
• Bundle 名称不允许重复，重复会直接报配置错误；
• 名称example_dags是保留名称，不可占用；示例 DAG 应通过[core] load_examples配置启用；
• 当[core] load_examples为True时，管理器会自动追加example_dagsBundle，并把每个已安装 Provider 的example_dags目录注册为名为<provider>-example-dags的本地 Bundle（见 _add_provider_example_dags_to_bundle）；
• classpath 通过import_string动态导入类，随后缓存在内存字典中（同一类不会被重复导入）；
• 启动时通过sync_bundles_to_db把配置同步为元数据库中的DagBundleModel记录，从配置中移除的 Bundle 会被标记为active = False（对多 dag-processor 分片部署可传deactivate_missing=False关闭该行为）。

凭据必须走 Connection，严禁内联

Bundle 的kwargs原样存储于[dag_processor] dag_bundle_config_list配置中，而 Airflow 在启用[api] expose_config时会通过 Config API 对外暴露配置。任何有配置读取权限的用户都能逐字读取这些值，因此kwargs中绝不能包含密钥。不要写出这样的配置：

"repo_url": "https://x-access-token:<token>@github.com/org/repo.git" # 严禁！

正确做法是引用 Airflow 的 Connection（Git 用git_conn_id、S3 用aws_conn_id、GCS 用gcp_conn_id），把真实凭据放到 secrets backend 中；Connection 的字段在运行时才解析，不会落盘写入dag_bundle_config_list。Airflow 的 secrets backend 能力覆盖本地文件、环境变量与外部服务等多种实现，具体可参见 secrets backend 文档。

GitDagBundle：版本化 DAG 的 GitOps 方案

Git Bundle 是唯一内置支持版本化的对象存储/Git 类 Bundle，运行每个 DAG Run 时会记录其创建时对应的 Git commit，即使仓库后续已更新，重跑仍使用完全相同的代码。

kwargs 说明

对应参数签名可直接在 GitDagBundle.init中确认。一个完整的配置示例：

[dag_processor] dag_bundle_config_list = [ { "name": "my-git-repo", "classpath": "airflow.providers.git.bundles.git.GitDagBundle", "kwargs": { "git_conn_id": "my_git_conn", "subdir": "dags", "tracking_ref": "main", } } ]

底层实现要点（git.py）：refresh()会根据tracking_ref拉取远端（origin/<tracking_ref>），get_current_version()返回当前 checkout 的 commit SHA，从而把"代码版本"与"运行版本"一一对应起来。Git Bundle 的完整 kwargs 列表与更多示例可参考apache-airflow-providers-git的 bundles 文档（本仓库中对应实现位于 providers/git）。

S3DagBundle 与 GCSDagBundle：从对象存储加载 DAG

两者均以bucket_name为唯一必填 kwarg，支持用prefix把 Bundle 限定到桶内子目录，且均不支持版本化（任务始终跑最新代码）。

S3 配置示例：

[dag_processor] dag_bundle_config_list = [ { "name": "my-s3-dags", "classpath": "airflow.providers.amazon.aws.bundles.s3.S3DagBundle", "kwargs": { "aws_conn_id": "aws_default", "bucket_name": "my-airflow-bucket", "prefix": "dags/" } } ]

aws_conn_id默认值为aws_default。从源码可见（s3.py），初始化时会校验桶与 prefix 是否存在。完整 kwargs 参考apache-airflow-providers-amazon的 bundles 文档（仓库实现位于 providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py）。

GCS 配置示例：

[dag_processor] dag_bundle_config_list = [ { "name": "my-gcs-dags", "classpath": "airflow.providers.google.cloud.bundles.gcs.GCSDagBundle", "kwargs": { "gcp_conn_id": "google_cloud_default", "bucket_name": "my-airflow-bucket", "prefix": "dags/" } } ]

gcp_conn_id默认值为google_cloud_default。对应实现位于 providers/google/src/airflow/providers/google/cloud/bundles/gcs.py。

在同一部署中组合多种 Bundle

不同类型的 Bundle 可以在一个dag_bundle_config_list里自由混用：默认的dags-folder本地 Bundle 既可删除，也可保留与其他 Bundle 并存：

[dag_processor] dag_bundle_config_list = [ { "name": "my_git_bundle", "classpath": "airflow.providers.git.bundles.git.GitDagBundle", "kwargs": {"tracking_ref": "main", "git_conn_id": "my_git_conn"} }, { "name": "dags-folder", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {} } ]

注意：多行值的缩进，尤其是最后一行，必须保持正确，否则 INI 多行值解析会失败——这是 Pythonconfigparser的既有行为（详见其支持的 INI 文件结构文档）。此外配置解析阶段还要求：Bundle 名称在整个列表中唯一，且不能使用保留名example_dags。

为每个 Bundle 单独定制刷新频率

全局刷新由[dag_processor] refresh_interval控制（默认 300 秒），决定 DAG Processor 多久去 Bundle 源里"找新文件"。可以通过 kwargs 传入refresh_interval对单个 Bundle 覆盖该值：

{ "name": "my-git-repo", "classpath": "airflow.providers.git.bundles.git.GitDagBundle", "kwargs": { "subdir": "dags", "tracking_ref": "main", "refresh_interval": 0 } }

在源码层面，BaseDagBundle.__init__的refresh_interval参数默认值即取自conf.getint("dag_processor", "refresh_interval")（见 base.py），config.yml 的示例配置也演示了按 Bundle 设 0（表示尽快刷新）的用法（config.yml）。

自定义视图 URL（view_url_template）

默认情况下每个 Bundle 类型会提供自己的视图 URL（例如 Git Bundle 指向仓库 Web 界面）。如需改用自定义 URL，可在该 Bundle 的kwargs中传view_url_template：

[dag_processor] dag_bundle_config_list = [ { "name": "my_git_repo", "classpath": "airflow.providers.git.bundles.git.GitDagBundle", "kwargs": { "tracking_ref": "main", "git_conn_id": "my_git_conn", "view_url_template": "https://my.custom.git.repo/view/{subdir}", } } ]

规则如下：

• 模板中的占位符（如{subdir}）只能是Bundle 自身拥有的属性，渲染时会替换为对应属性值（manager.py 的_extract_template_params通过正则\{([^}]+)\}提取占位符并取getattr(bundle, placeholder)），不允许使用 Bundle 属性之外的任何占位符；
• 指定自定义 URL 后，它会覆盖该 Bundle 默认提供的 URL；
• URL 会经过安全检查_is_safe_bundle_url（manager.py）：只允许http/httpsscheme、必须有网络位置（netloc）、不得包含控制字符；若判定不安全，该 Bundle 的视图 URL 会被置为None，以防潜在安全问题。

值得一提的是，安全 URL 还会用[core] fernet_key做签名后落库（signed_url_template），用于完整性校验；视图 URL 的生成不需要初始化 Bundle，因此 UI 在展示时无需触发拉取等昂贵操作。旧版view_url方法已标记弃用，请使用view_url_template。

存储布局与磁盘清理：理解底层运行方式

尽管配置是逻辑层面的，但理解 Bundle 在磁盘上的落点有助于排查问题（相关辅助函数见 base.py）：

• 存储根目录由[dag_processor] dag_bundle_storage_path指定（必须为绝对路径）；未设置时默认落到Path(tempfile.gettempdir()) / "airflow" / "dag_bundles"；
• 每个 Bundle 的基础目录为<root>/<bundle_name>，各版本副本存放在<root>/<bundle_name>/versions/<version>；
• 共享磁盘的 Worker 上，Bundle 版本副本会随任务运行不断累积，因此 Airflow 提供了一组清理配置（见 config.yml）：

清理由BundleUsageTrackingManager（base.py）执行：它跟踪每个版本最后被使用的时间戳，删除"使用不活跃且超过阈值"的版本，但总是保留最近使用的 N 个版本；删除前通过文件锁（flock）确保不会误删正在被任务使用的副本——这也呼应了"Worker 上多个版本并存、每个版本按需锁定使用"的设计。

Docker 镜像中安装 Git

从Airflow 3.0.2起，官方基础镜像已预装 git；若你使用更早的版本（或自建镜像），需要自行安装 git 并配置gitpython使用它：

RUN apt-get update && apt-get install -y git ENV GIT_PYTHON_GIT_EXECUTABLE=/usr/bin/git ENV GIT_PYTHON_REFRESH=quiet

其中GIT_PYTHON_GIT_EXECUTABLE告诉 GitPython 去哪里找 git 可执行文件，GIT_PYTHON_REFRESH=quiet抑制 GitPython 每次调用的仓库刷新日志。

在用户模拟（run_as_user）下使用 DAG Bundle

当结合run_as_user（用户模拟）使用 Bundle 时，需要保证被模拟的用户能访问由 Airflow 主进程创建的 Bundle 文件，请按以下两点配置权限：

1. 所有被模拟用户与 Airflow 运行用户属于同一个用户组；
2. 配置合适的 umask（例如umask 0002），使组内成员可读/写新建文件。

注意：这种基于共享组权限的做法只是临时方案。未来的 Airflow 版本将通过 supervisor 化的 Bundle 操作来管理多用户访问，届时无需再依赖共享组权限。

控制重跑使用哪个版本：rerun_with_latest_version

当一个 DAG Run 或 Task Instance 被 clear 时，UI 会弹出一个复选框，询问重跑时使用最新 Bundle 版本还是原运行所用的版本。rerun_with_latest_version设置即控制该复选框的默认勾选状态，团队无需每次手动决策；该设置同样决定通过 API 或 CLI 创建 backfill 时的默认run_on_latest_version行为。

工作原理

• 每个 DAG 都有一个解析版本（DagModel.bundle_version），每次 DAG Processor 重新解析 DAG 文件时更新；每个 DAG Run 则记录它创建时所使用的 Bundle 版本；
• 当rerun_with_latest_version为False时，clear 一个 DAG Run 会保留其原始 Bundle 版本，重跑使用同一份代码——这在排查失败时可复现；为True时，clear 会把 DAG Run 更新到当前解析版本，确保重跑采用最新代码。

解析优先级（从高到低）

1. 显式请求：API 请求体中的run_on_latest_version参数（若提供）；
2. DAG 级：DAG 构造参数rerun_with_latest_version（显式传True/False时）；
3. 全局配置：[core] rerun_with_latest_version选项（若设置）；
4. 调用点回退：clear/rerun 场景默认False，backfill 场景默认True（保留各自的历史默认行为）。

一个例外：自身没有任何版本的 DAG Run（从 Airflow 2 迁移而来，或版本被airflow db clean清理掉）没有可保留的版本，因此 clear 它时无论设置如何，总是使用最新代码与最新 Bundle 版本。

全局配置

[core] rerun_with_latest_version = False # 重跑时使用原始 Bundle 版本 # rerun_with_latest_version = True # 重跑时使用最新 Bundle 版本

未设置时按第 4 条调用点回退规则处理。该配置项在 config.yml 中定义（version_added: 3.2.0）。

DAG 级配置

可按 DAG 覆盖全局默认值：

from datetime import datetime from airflow import DAG from airflow.operators.empty import EmptyOperator # 该 DAG 总是用最新版本重跑 with DAG( dag_id="always_latest_dag", rerun_with_latest_version=True, start_date=datetime(2024, 1, 1), ) as dag: EmptyOperator(task_id="task")

典型使用场景

• 调试失败运行：用False（默认），clear 失败运行后以同一份代码重跑，便于复现与定位问题；
• 总是运行最新代码：团队更希望重跑总是拿到最新代码时（例如原运行后又部署了修复），全局设[core] rerun_with_latest_version = True；
• 混合策略：全局默认设为True，但对关键 DAG 用rerun_with_latest_version=False覆盖，在最需要版本稳定性的地方保持固定版本。

注意：以上行为仅适用于支持版本化的 Bundle 类型（如GitDagBundle）。本地 Bundle 不支持版本化，始终使用最新代码，不受该设置影响。

与 disable_bundle_versioning 的关系

Airflow 提供两个影响 Bundle 版本行为的独立设置，二者目的不同：

• disable_bundle_versioning：彻底关闭版本跟踪。置为True后，DAG Run 上不再记录bundle_version。可同时作为 DAG 参数与全局配置（[dag_processor] disable_bundle_versioning，默认False，见 config.yml）使用，且只对支持版本化的 Bundle 生效；
• rerun_with_latest_version：在保持版本跟踪开启的前提下，控制默认的重跑行为，仅改变呈现给用户的默认选择，版本历史仍被完整记录。

一句话总结：disable_bundle_versioning回答"是否要跟踪版本"，rerun_with_latest_version回答"重跑时默认用哪个版本"。两者相互独立，且当版本化被禁用时，rerun_with_latest_version不产生任何效果。

编写自定义 DAG Bundle

当内置类型无法满足需求（如自建代码托管平台、企业内部对象存储）时，可继承BaseDagBundle实现自定义 Bundle。

必须实现的抽象方法

自定义类必须实现以下三个成员（基类中用@abstractmethod声明，见 base.py）：

• path（属性）：返回一个Path，指向本 Bundle DAG 文件存放的目录。Airflow 依据它定位并解析 DAG 文件；initialize()之后，该路径下应能访问到全部 DAG 文件；
• get_current_version()：返回当前 Bundle 版本的字符串。之后 Airflow 运行任务时会把该版本传回__init__以重新获取同一版本；若不支持版本化则返回None。推荐返回BundleVersion数据类（同时携带版本字符串与可选的结构化元数据，如 S3 manifest）；直接返回裸字符串属于已弃用的旧路径，未来会发出弃用警告（base.py）；
• refresh()：负责从源头刷新 Bundle 内容（例如从远端拉取最新提交），DAG Processor 会周期性调用它以保证 Bundle 不过期。

可选覆写的方法

• __init__：可扩展以接收额外参数（如 Git Bundle 的tracking_ref），必须调用父类__init__保证name、refresh_interval、version、version_data、view_url_template等基础字段正确初始化。避免在本方法中做网络调用等昂贵操作（会拖慢 Bundle 实例化），放到initialize中做；
• initialize()：在 DAG Processor 或 Worker首次真正使用Bundle 内容前被调用，适合执行昂贵的准备工作；它只在 Airflow 需要把 Bundle 文件落到磁盘时触发（纯调用view_url的场景不会触发）。覆写时务必在方法末尾调用super().initialize()，基类实现会校验 Bundle 路径存在并给出告警（base.py）；
• view_url_template()：返回一个字符串模板，用于在 UI 中跳转到外部系统（如 Git Web 界面）查看该 Bundle 某版本；它需要在未调用initialize的情况下也能工作。

一个最小化的"本地目录 + 自定义源"风格示例：

from pathlib import Path from airflow.dag_processing.bundles.base import BaseDagBundle class MyStaticBundle(BaseDagBundle): supports_versioning = False # 或不声明，基类默认即为 False def __init__(self, *, source_path: str, **kwargs) -> None: super().__init__(**kwargs) self.source_path = Path(source_path) @property def path(self) -> Path: return self.source_path def get_current_version(self) -> None: # 不支持版本化 → 返回 None return None def refresh(self) -> None: # 例如：从自定义源同步文件到 self.path ... def view_url_template(self) -> str | None: return "https://example.com/browse/{version}"

随后即可在dag_bundle_config_list中通过classpath指向它。

其他关键注意事项

• 版本化 Bundle：若你的 Bundle 支持版本化，请确保initialize、get_current_version、refresh都正确处理版本相关的逻辑（按版本检出/保留副本）；
• 并发安全：Worker 可能同时创建大量 Bundle 对象，Airflow不会序列化对 Bundle 对象的调用。如果底层技术对此敏感（例如多个进程同时 clone 同一 git 仓库），Bundle 类必须自行加锁保证同一时刻只有一个对象在克隆。基类提供了可复用的lock上下文管理器（基于fcntl.flock的排他文件锁，见 base.py）；
• Triggerer 限制：DAG Bundle 不会在 triggerer 组件中被初始化，因此 trigger 代码不能来自 DAG Bundle——triggerer 不处理随时间变化的 trigger 代码，一切都在主进程中发生。若需要自定义 trigger，请确保它位于 Python 环境sys.path中（而不是从 Bundle 获取）。

升级与迁移要点

• DAG Bundle 是 Airflow 3.0 的核心新能力，元数据库中的dag_bundle表由迁移脚本引入（见 0050_3_0_0_add_dagbundlemodel.py）；后续版本又加入了 URL/模板参数、bundle_name非空约束等演进；
• 默认dags-folderBundle 的存在保证了从 Airflow 2 平滑升级：旧 DAG 仍然从dags_folder加载。若部署使用了自定义 Bundle，历史遗留 DAG 行（bundle_name指向未配置 Bundle 或relative_fileloc为空）会在解析生命周期中自愈，必要时可执行airflow dags reserialize强制重写bundle_name与relative_fileloc；
• 从 Airflow 2 迁移过来、本身不带 Bundle 版本的 DAG Run，clear 时总是采用最新版本（无旧版本可保留）。

通过将 DAG 交付升级为"版本化的 Bundle"，Airflow 3 让大规模团队协作下的 DAG 生命周期管理走向了 GitOps 化：代码进仓库、运行按版本、重跑可复现、多源可并存。本文涉及的配置、类与源码入口都可在本仓库对应路径中进一步展开研读。

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow