Apache Airflow 系列教程 | 第33课：实战项目 — 构建企业级 ETL 平台

导读（Introduction）

欢迎来到 Apache Airflow 源码深度解析系列的第33课，也是"综合实战篇"的第一课。

在前面32课的学习中，我们系统性地剖析了 Airflow 的各个核心模块——从 DAG 定义到调度器内核，从执行器架构到安全认证，从 Asset 事件驱动到 Object Storage 抽象层。每一课都聚焦于某个特定子系统的深度解析。现在是时候将这些知识融会贯通，应用到真实的企业级场景中了。

本课将带你从零构建一个完整的企业级 ETL（Extract-Transform-Load）平台。这不是一个简单的 “Hello World” 示例，而是一个包含多团队协作、多层数据依赖、数据质量校验、监控告警、生产运维等完整生命周期的真实系统设计。我们将综合运用前面学到的所有核心概念：

• Asset 驱动的多层 DAG 依赖——构建数据仓库的分层流转
• Multi-Team 权限模型——实现多团队之间的资源隔离
• Deadline 与 Callback 机制——建立完善的监控告警体系
• Backfill 系统——处理数据修复与历史重跑
• ObjectStoragePath——统一数据 IO 层

通过本课的实战，你将理解如何将 Airflow 的各个组件编排成一个可靠、可观测、可维护的生产级数据平台。

学习目标（Learning Objectives）

完成本课学习后，你将能够：

1. 设计企业 ETL 平台的整体架构——理解数据仓库分层（ODS/DWD/DWS/ADS）在 Airflow 中的映射方式
2. 掌握多团队协作模式——利用 Team、DAG Bundle、RBAC 实现组织级的权限隔离
3. 构建 Asset 驱动的多层依赖——使用 Asset 表达式实现解耦的跨层级数据触发
4. 集成外部数据源——通过 Hook/Operator/Connection 对接数据库、API、文件系统
5. 实现数据质量检查与告警——结合 Sensor、Callback、Notifier 构建质量门控
6. 制定生产运维策略——包括 Backfill 数据回填、Deadline 超时管理、故障恢复方案

正文内容（Main Content）

1. 企业 ETL 平台的架构设计

1.1 数据仓库分层模型

一个典型的企业级数据仓库采用经典的分层架构，每一层处理不同粒度的数据：

在 Airflow 中，每一层对应一组 DAG，层与层之间通过Asset建立数据依赖关系。上游层的 DAG 完成数据产出后，通过 Asset 事件自动触发下游层的 DAG 运行。

1.2 平台整体架构

┌────────────────────────────────────────────────────────────┐ │ 企业 ETL 平台架构 │ ├────────────────────────────────────────────────────────────┤ │ │ │ 数据源层 ODS层 DWD层 DWS层 ADS层 │ │ ┌──────┐ ┌────────┐ ┌───────┐ ┌──────┐ ┌───┐│ │ │MySQL │─────▶│ods_user│────▶│dwd_ │──▶│dws_ │─▶│ads││ │ │Oracle│─────▶│ods_order────▶│ user │ │ user │ │ ││ │ │API │─────▶│ods_pay │────▶│ order │ │ sales│ │ ││ │ │Files │─────▶│ods_log │ │ pay │ │ │ │ ││ │ └──────┘ └────────┘ └───────┘ └──────┘ └───┘│ │ │ │ │ │ │ │ │ ▼ ▼ ▼ ▼ ▼ │ │ [Connection] [Asset事件] [Asset事件] [Asset] [Asset]│ │ │ ├────────────────────────────────────────────────────────────┤ │ 横切关注点： │ │ • Multi-Team 权限隔离 │ │ • Deadline 超时告警 │ │ • 数据质量门控（Sensor + Callback） │ │ • Backfill 回填策略 │ │ • 统一监控仪表盘 │ └────────────────────────────────────────────────────────────┘

1.3 技术选型与组件映射

2. 多团队协作模式：Team 与 DAG 权限

2.1 Multi-Team 架构原理

在企业环境中，通常有多个团队共享同一套 Airflow 集群。数据工程团队负责 ODS/DWD 层的 ETL 管道，数据分析团队负责 DWS/ADS 层的报表聚合，算法团队可能有独立的模型训练管道。这些团队之间需要资源隔离和权限边界。

Airflow 3.x 引入了原生的 Multi-Team 支持。从源码层面看，其核心数据模型定义在 team.py：

classTeam(Base):""" Contains the list of teams defined in the environment. This table is only used when Airflow is run in multi-team mode. """__tablename__="team"name:Mapped[str]=mapped_column(String(50),primary_key=True)dag_bundles=relationship("DagBundleModel",secondary=dag_bundle_team_association_table,back_populates="teams")

Team 通过一个多对多关联表dag_bundle_team_association_table与 DAG Bundle 建立关联。值得注意的是，团队归属的粒度是DAG Bundle（而非单个 DAG）——一个 Bundle 中的所有 DAG 统一归属同一个 Team：

dag_bundle_team_association_table=Table("dag_bundle_team",Base.metadata,Column("dag_bundle_name",StringID(length=250),ForeignKey("dag_bundle.name",ondelete="CASCADE"),primary_key=True),Column("team_name",String(50),ForeignKey("team.name",ondelete="CASCADE"),primary_key=True),# 每个 Bundle 最多属于一个 TeamIndex("idx_dag_bundle_team_dag_bundle_name","dag_bundle_name",unique=True),)

这意味着权限继承链条为：Task → DAG → DAG Bundle → Team。

2.2 权限模型与资源隔离

Airflow 的资源授权通过BaseAuthManager抽象层实现，每种受保护资源都携带team_name属性用于隔离判断。从 resource_details.py 可以看到：

@dataclassclassDagDetails:"""Represents the details of a Dag."""id:str|None=Noneteam_name:str|None=None@dataclassclassConnectionDetails:"""Represents the details of a connection."""conn_id:str|None=Noneteam_name:str|None=None@dataclassclassVariableDetails:"""Represents the details of a variable."""key:str|None=Noneteam_name:str|None=None@dataclassclassPoolDetails:"""Represents the details of a pool."""name:str|None=Noneteam_name:str|None=None

在 SimpleAuthManager 中，权限判断的核心逻辑非常清晰：

def_is_authorized(self,*,user,team_name,...):ifteam_nameandteam_namenotinuser.teams:returnFalse# 用户不属于该 Team，拒绝访问...

2.3 企业 ETL 平台的团队规划

在我们的实战项目中，按照职能划分以下团队：

# airflow.cfg 配置（SimpleAuthManager 模式）# [core]# multi_team = True# simple_auth_manager_users =# data_eng_lead:admin:data-engineering,# etl_dev_alice:user:data-engineering,# etl_dev_bob:user:data-engineering,# analyst_carol:user:data-analytics,# analyst_dave:viewer:data-analytics,# ml_eng_eve:user:ml-platform# DAG Bundle 归属规划：# ┌────────────────────┬─────────────────────┐# │ DAG Bundle │ Team │# ├────────────────────┼─────────────────────┤# │ ods-extraction │># │ dwd-transformation │># │ dws-aggregation │># │ ads-serving │># │ ml-training │ ml-platform │# └────────────────────┴─────────────────────┘

这样的设计确保了：

• data-engineering团队只能管理 ODS 和 DWD 层的 DAG、Connection、Variable
• data-analytics团队管理 DWS 和 ADS 层
• ml-platform团队有自己独立的 DAG Bundle
• 跨团队的数据流转通过Asset事件实现解耦——数据工程团队的 DAG 产出 Asset 事件，分析团队的 DAG 通过 Asset 调度自动触发

3. Asset 驱动的多层 DAG 依赖设计

3.1 Asset 核心概念回顾

Asset 是 Airflow 3.x 的数据资产抽象，定义在 asset/init.py：

classAsset(os.PathLike,BaseAsset):"""A representation of data asset dependencies between workflows."""name:str# 资产唯一名称uri:str# 资产 URI（如 s3://bucket/path）group:str# 分组（默认为 asset_type）extra:dict# 自定义元数据watchers:list# 外部事件触发器allow_producer_teams:list[str]# 允许产出该资产的团队列表

Asset 驱动调度的工作原理是：

1. 生产者 DAG的 Task 声明outlets=[asset]
2. Task 成功完成后，Airflow 记录一个 AssetEvent
3. 消费者 DAG的schedule=[asset]声明等待 Asset 更新
4. Scheduler 检测到所有依赖的 Asset 都已更新，创建新的 DagRun

3.2 多层依赖的 Asset 表达式

Airflow 支持丰富的 Asset 表达式来描述复杂依赖关系。从 example_assets.py 可以看到各种模式：

# 单一 Asset 依赖：任何一个 Asset 更新即触发schedule=[dag1_asset]# AND 语义：列表中所有 Asset 都必须更新schedule=[dag1_asset,dag2_asset]# 显式 AND 表达式schedule=(dag1_asset&dag2_asset)# OR 表达式：任意一个 Asset 更新即触发schedule=(dag1_asset|dag2_asset)# 复合表达式：dag1 更新 OR (dag2 AND dag3 同时更新)schedule=(dag1_asset|(dag2_asset&dag3_asset))# 混合调度：Asset 事件 + 定时调度schedule=AssetOrTimeSchedule(timetable=CronTriggerTimetable("0 1 * * 3",timezone="UTC"),assets=(dag1_asset&dag2_asset))

3.3 实战：定义数据仓库资产层级

""" enterprise_etl/assets.py 定义企业数据仓库的全部数据资产（Asset） """fromairflow.sdkimportAsset# ============================================================# ODS 层资产 —— 原始数据落地# ============================================================ods_user=Asset(name="ods_user",uri="s3://data-warehouse/ods/user/",group="ods",extra={"source":"mysql_prod","table":"users","load_type":"incremental"},)ods_order=Asset(name="ods_order",uri="s3://data-warehouse/ods/order/",group="ods",extra={"source":"mysql_prod","table":"orders","load_type":"incremental"},)ods_payment=Asset(name="ods_payment",uri="s3://data-warehouse/ods/payment/",group="ods",extra={"source":"pg_billing","table":"payments","load_type":"full"},)ods_user_behavior=Asset(name="ods_user_behavior",uri="s3://data-warehouse/ods/user_behavior/",group="ods",extra={"source":"kafka","topic":"user_events","load_type":"streaming_batch"},)# ============================================================# DWD 层资产 —— 明细清洗数据# ============================================================dwd_user_detail=Asset(name="dwd_user_detail",uri="s3://data-warehouse/dwd/user_detail/",group="dwd",extra={"grain":"user_id","partition":"ds"},)dwd_order_detail=Asset(name="dwd_order_detail",uri="s3://data-warehouse/dwd/order_detail/",group="dwd",extra={"grain":"order_id","partition":"ds"},)dwd_payment_detail=Asset(name="dwd_payment_detail",uri="s3://data-warehouse/dwd/payment_detail/",group="dwd",extra={"grain":"payment_id","partition":"ds"},)# ============================================================# DWS 层资产 —— 汇总宽表# ============================================================dws_user_daily_summary=Asset(name="dws_user_daily_summary",uri="s3://data-warehouse/dws/user_daily_summary/",group="dws",extra={"grain":"user_id,ds","metrics":"order_cnt,pay_amt,active_score"},)dws_sales_daily=Asset(name="dws_sales_daily",uri="s3://data-warehouse/dws/sales_daily/",group="dws",extra={"grain":"ds","metrics":"total_orders,total_revenue,avg_order_value"},)# ============================================================# ADS 层资产 —— 应用服务数据# ============================================================ads_user_portrait=Asset(name="ads_user_portrait",uri="s3://data-warehouse/ads/user_portrait/",group="ads",extra={"consumer":"user_profile_api"},)ads_sales_report=Asset(name="ads_sales_report",uri="s3://data-warehouse/ads/sales_report/",group="ads",extra={"consumer":"bi_dashboard"},)

3.4 实战：ODS 层 — 数据抽取 DAG

""" enterprise_etl/dags/ods_extraction.py ODS 层数据抽取 DAG —— 每日凌晨从各数据源全量/增量抽取 """from__future__importannotationsfromdatetimeimporttimedeltaimportpendulumfromairflow.sdkimportDAG,taskfromairflow.sdkimportAsset,AsyncCallback,DeadlineAlert,DeadlineReferencefromairflow.providers.slack.notifications.slack_webhookimportSlackWebhookNotifier# 导入资产定义fromenterprise_etl.assetsimportods_user,ods_order,ods_payment,ods_user_behaviordefon_ods_failure(context):"""ODS 层失败回调：记录失败信息并发送告警"""ti=context["task_instance"]dag_run=context["dag_run"]print(f"[ALERT] ODS extraction failed:{ti.task_id}in{dag_run.dag_id}")withDAG(dag_id="ods_daily_extraction",description="ODS层每日数据抽取管道",schedule="0 2 * * *",# 每天凌晨2点执行start_date=pendulum.datetime(2024,1,1,tz="Asia/Shanghai"),catchup=False,tags=["ods","extraction","daily"],default_args={"owner":"data-engineering","retries":3,"retry_delay":timedelta(minutes=5),"retry_exponential_backoff":True,"on_failure_callback":on_ods_failure,},# Deadline 告警：超过60分钟发出警告deadline=[DeadlineAlert(reference=DeadlineReference.DAGRUN_QUEUED_AT,interval=timedelta(minutes=60),callback=AsyncCallback(SlackWebhookNotifier,kwargs={"text":"⚠️ ODS extraction running over 60 minutes: { { dag_run.dag_id }}"},),),DeadlineAlert(reference=DeadlineReference.DAGRUN_QUEUED_AT,<