Airflow DAG 编排：数据依赖图比 cron 表达式更可靠

Airflow DAG 编排数据依赖图比 cron 表达式更可靠如果你还在用 cron 表达式调度数据任务大概率已经遇到过这些噩梦任务 A 还没跑完任务 B 就启动了拿到的是半成品数据任务 C 失败了但下游 D/E/F 全都照跑不误浪费了 3 小时计算资源想临时重跑某一天的分区发现要手动改 6 个 cron 时间参数。Airflow 的 DAGDirected Acyclic Graph有向无环图编排模式从根本上解决了这些问题。它不是更高级的 cron而是完全不同的任务编排哲学。一、DAG 的核心逻辑依赖声明而不是时间声明cron 的思路是时间驱动每个任务独立声明自己的运行时间互相之间没有关联。DAG 的思路是依赖驱动任务不声明自己什么时候跑只声明自己依赖谁Airflow 自动推导执行顺序。打个比方cron 就像每个人都有自己的闹钟但没有人确认你起床了才出门。DAG 就像接力赛上一棒交棒了下一棒才起跑没交棒就等着。graph LR A[extract_user] -- B[clean_user] A -- C[extract_order] C -- D[clean_order] B -- E[join_data] D -- E E -- F[compute_metrics] F -- G[load_to_dw] G -- H[notify_team] style A fill:#e1f5fe style E fill:#fff3e0 style H fill:#e8f5e9这个 DAG 的含义是extract_user和extract_order可以并行它们没有互相依赖join_data必须等两个 clean 任务都完成多上游依赖notify_team是末端通知只有全链路成功才触发如果用 cron 来实现同样的逻辑你需要给每个任务设定不同的时间保证上游比下游早跑在每个任务脚本里手动检查上游任务的输出文件是否存在失败时手动排查是哪一步出了问题而 DAG 只需要你声明依赖关系Airflow 自动处理执行顺序、等待、失败检测和重试。二、写一个完整的 ETL DAG我们来写一个从抽取到加载的完整 DAG覆盖常见的数据工程场景from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator from airflow.sensors.filesystem import FileSensor from datetime import datetime, timedelta # ---- 默认参数 ---- # 为什么把默认参数单独定义而不是写在每个任务里 # 因为这些参数是全链路的通用策略比如重试次数和超时时间 # 单独定义方便统一调整避免某个任务忘了设重试导致全链路卡死 default_args { owner: data-team, retries: 2, # 失败重试2次——为什么不是更多 # 因为数据任务通常是上游数据问题重试再多也没用 # 2次是给偶发性网络抖动留的容错空间 retry_delay: timedelta(minutes5), # 重试间隔5分钟——为什么不是1分钟 # 因为上游数据源可能有临时故障1分钟太短还没恢复 # 5分钟是经验值覆盖大多数短暂故障的恢复窗口 execution_timeout: timedelta(hours1), # 单任务超时1小时 } # ---- DAG 定义 ---- # schedule_interval0 8 * * * 表示每天8点触发 # 但这不是任务在8点跑而是DAG在8点创建一个执行实例 # 实际任务启动时间取决于上游依赖是否完成 # catchupFalse 表示不补跑历史日期——为什么 # 因为老数据通常已经过了时效补跑只会浪费资源 # 只有明确需要回溯时才手动触发特定日期 with DAG( dag_iduser_order_etl, default_argsdefault_args, schedule_interval0 8 * * *, start_datedatetime(2025, 1, 1), catchupFalse, tags[etl, daily], ) as dag: # ---- 任务1等待源数据文件就绪 ---- # FileSensor 是传感器——它不执行逻辑只检查条件是否满足 # 为什么用传感器而不是在extract里直接读 # 因为源数据由外部系统产出我们无法控制它的生成时间 # 传感器会每分钟检查一次文件出现才放行避免读到不完整数据 wait_user_file FileSensor( task_idwait_user_file, filepath/data/inbox/user_export_{{ ds }}.csv, # {{ ds }} 是 Airflow 的执行日期宏——为什么用宏而不是固定日期 # 因为每个 DAG 实例对应一个逻辑日期宏自动替换成当前实例的日期 # 这样同一个 DAG 定义可以跑任意日期不用手动改参数 poke_interval60, # 每60秒检查一次 timeout3600, # 最长等待1小时超时则失败 ) # ---- 任务2抽取用户数据 ---- def extract_user(**context): import pandas as pd # 从执行日期拿到文件路径——context[ds] 是 Airflow 注入的逻辑日期 # 为什么从 context 而不是硬编码路径 # 因为 DAG 可能被手动触发任意日期硬编码只能跑当天 ds context[ds] df pd.read_csv(f/data/inbox/user_export_{ds}.csv) # 基本清洗去重和空值处理 # 为什么在抽取阶段就做去重 # 因为源系统可能有重复导出不去重会导致下游 join 产生笛卡尔积 df df.drop_duplicates(subset[user_id]) df.to_parquet(f/data/staging/user_{ds}.parquet) return df.shape[0] # Xcom 返回行数供下游任务参考 task_extract_user PythonOperator( task_idextract_user, python_callableextract_user, ) # ---- 任务3抽取订单数据从数据库 ---- def extract_order(**context): from sqlalchemy import create_engine ds context[ds] engine create_engine(mysqlpymysql://user:passhost:3306/orders) # 为什么用 WHERE date {{ ds }} 而不是全量查询 # 因为订单表可能有千万行数据全量查询会炸内存 # 按日期分区抽取每次只处理一天的数据量 query fSELECT * FROM orders WHERE date {ds} df pd.read_sql(query, engine) df.to_parquet(f/data/staging/order_{ds}.parquet) return df.shape[0] task_extract_order PythonOperator( task_idextract_order, python_callableextract_order, ) # ---- 任务4关联用户和订单 ---- def join_data(**context): import pandas as pd ds context[ds] users pd.read_parquet(f/data/staging/user_{ds}.parquet) orders pd.read_parquet(f/data/staging/order_{ds}.parquet) # left join——为什么不用 inner join # 因为业务需要知道有用户但没有订单的情况 # inner join 会丢掉这些用户导致用户维度分析不完整 merged orders.merge(users, onuser_id, howleft) merged.to_parquet(f/data/staging/joined_{ds}.parquet) return merged.shape[0] task_join PythonOperator( task_idjoin_data, python_callablejoin_data, ) # ---- 任务5加载到数据仓库 ---- # BashOperator 执行 shell 命令——为什么不用 PythonOperator # 因为 ClickHouse 的导入用 clickhouse-client 更方便 # Python 的 clickhouse-driver 需要额外装包shell 命令更轻量 task_load BashOperator( task_idload_to_dw, bash_command( clickhouse-client --queryINSERT INTO analytics.user_order SELECT * FROM input() --input_formatParquet /data/staging/joined_{{ ds }}.parquet ), ) # ---- 任务6通知团队 ---- def notify(**context): # 从 XCom 获取上游任务的返回值——为什么 # 因为通知内容需要包含数据量信息方便团队确认数据是否正常 # 比如某天只有 3 条订单数据明显有问题通知里能看到 user_count context[ti].xcom_pull(task_idsextract_user) order_count context[ti].xcom_pull(task_idsextract_order) joined_count context[ti].xcom_pull(task_idsjoin_data) print(fETL 完成: 用户{user_count}条, 订单{order_count}条, 关联{joined_count}条) task_notify PythonOperator( task_idnotify_team, python_callablenotify, ) # ---- 依赖声明 ---- # 这就是 DAG 的核心——只声明依赖不声明时间 # Airflow 自动推导wait → extract → join → load → notify # parallel: extract_user 和 extract_order 可以同时跑 [wait_user_file, task_extract_user] task_extract_order task_join task_load task_notify三、DAG 调度的三种高级模式日常使用中除了基本的依赖链还有三种常见的高级调度需求1. 条件分支根据数据量决定走哪条路from airflow.operators.python import BranchPythonOperator def decide_path(**context): # 从上游拿到数据量决定走正常加载还是告警分支 # 为什么需要分支因为数据量异常时直接加载会导致仓库数据波动 # 小于阈值走告警流程人工确认后再加载 row_count context[ti].xcom_pull(task_idsjoin_data) if row_count 1000: return alert_low_volume else: return load_to_dw branch BranchPythonOperator( task_iddecide_path, python_callabledecide_path, )2. 子DAG复用通用逻辑如果你的团队有 5 个 DAG 都需要抽取 → 清洗 → 加载的基础流程把这段抽成子 DAG避免重复代码。子 DAG 像函数一样被调用改一处所有 DAG 都生效。3. 动态任务根据上游结果决定任务数量# 为什么需要动态任务 # 比如上游按省份拆分了数据下游需要对每个省份独立处理 # 省份列表不是固定的每个月可能有新增 # 静态定义任务无法覆盖动态变化的分区数 def generate_tasks(**context): provinces context[ti].xcom_pull(task_idssplit_by_province) for province in provinces: PythonOperator( task_idfprocess_{province}, python_callableprocess_province, op_args[province], )四、Airflow 运维的三个坑和解决办法坑1DAG 解析慢调度延迟Airflow 每隔 30 秒重新解析所有 DAG 文件。如果你的 DAG 目录里有 200 个文件每个文件 500 行解析就是沉重的 CPU 开销。解决办法把 DAG 拆成小文件每个 DAG 只包含核心依赖声明具体逻辑用 Python 包外部化。DAG 文件只做胶水不做业务。坑2XCom 大数据传递XCom 默认存在数据库里大小限制约 48KB。如果你把一个 DataFrame 直接 push 到 XCom大概率会报错。解决办法XCom 只传元数据行数、文件路径实际数据走文件系统或对象存储。这是传信不传物的原则。坑3时区陷阱Airflow 的execution_date是逻辑时间不是物理时间。UTC8 的 8 点 DAGexecution_date是 UTC 的 0 点。所有基于日期的文件路径和分区查询都要用ds宏而不是datetime.now()否则时区偏移会让数据对不上。五、总结Airflow DAG 的核心理念是依赖声明取代时间声明你不告诉任务什么时候跑只告诉它等谁跑完再跑。这让多任务编排从靠时间估算变成靠依赖推导从根本上避免了上游没完成下游就启动的竞态问题。写好 DAG 要注意默认参数统一设置重试策略FileSensor 处理外部依赖XCom 只传元数据不传大数据日期用ds宏不用硬编码。运维上要关注DAG 文件保持轻量避免解析慢时区问题用逻辑日期而不是物理时间条件分支和动态任务覆盖复杂场景。从 cron 迁移到 Airflow 不是换了个调度工具而是换了一套编排哲学——从每个任务各自为政到全链路有据可依。