应将pandas逻辑封装为可导入模块，用绝对路径和显式导入；设execution_timeout、捕获异常、禁用matplotlib gui；用>>链式声明依赖；在airflow环境pip安装pandas或构建自定义镜像。
Airflow 中如何正确封装 pandas 脚本为可调度任务
直接把本地跑通的 pandas 脚本扔进 Airflow 的 PythonOperator 里，大概率会失败——不是路径找不到 CSV，就是 import 报错，或者内存爆掉。根本原因是 Airflow Worker 运行环境与你本地开发环境隔离，且默认不加载当前 DAG 文件所在目录到 sys.path。
实操建议：
- 把数据处理逻辑抽成独立的、可 import 的 Python 模块（例如
etl/transform.py），放在AIRFLOW_HOME/dags/同级或子目录下，并确保该目录有__init__.py - 在 DAG 文件中用
from etl.transform import clean_sales_data显式导入，别用exec(open(...).read())或runpy动态执行 - 所有文件路径必须用绝对路径或基于
os.path.dirname(__file__)构造，避免硬编码"./data/input.csv" - 如果脚本依赖大文件（如 500MB 的原始 Excel），别在 task 中直接
pd.read_excel()—— 先用BashOperator下载到共享存储（如 S3/NFS），再由PythonOperator加载
为什么 pandas 任务在 Airflow 里总卡在 “running” 状态不结束
这不是 pandas 慢，是 Airflow 没收到进程退出信号。常见于：脚本里调了 input()、用了未关闭的 matplotlib 图形后端、或 pd.read_csv() 遇到损坏文件卡死但没设 timeout 和异常捕获。
排查和修复要点：
- 在 DAG 的
default_args中强制设置"execution_timeout": timedelta(minutes=10)，防止无限挂起 - 所有 I/O 操作包在
try/except Exception as e:里，并显式raise AirflowException(f"ETL failed: {e}")，让 Airflow 正确标记失败而非假死 - 禁用 matplotlib 的 GUI 后端：在 task 函数开头加
import matplotlib; matplotlib.use("Agg") - 检查 pandas 版本兼容性 —— Airflow 2.6+ 默认用 Python 3.8+，但某些旧版
pandas==1.1.5在多线程读取时会 hang，升级到pandas>=1.5.3可缓解
如何让多个 pandas 任务按依赖顺序执行（比如清洗 → 聚合 → 导出）
别靠时间戳或文件轮询来“猜”前序任务是否完成。Airflow 的 DAG 依赖必须显式声明，否则清洗失败后聚合仍可能触发，导致空表入库。
正确写法：
- 用
task1 >> task2 >> task3链式写法，而不是仅靠depends_on_past=True（后者只校验上一次运行状态，不保证上游本次成功） - 如果清洗任务输出中间结果（如 parquet 文件），聚合任务应通过
ExternalTaskSensor监听清洗任务的success状态，而非轮询文件是否存在 - 避免跨 task 共享变量（如全局 DataFrame），每个 task 必须独立加载输入、写出输出；推荐统一用
tempfile.mkstemp()或对象存储 URI（如s3://my-bucket/etl/stage/cleaned_{{ ds }}.parquet）传递数据 - 导出任务（如写入 PostgreSQL）务必启用
autocommit=True并捕获IntegrityError，否则主键冲突会导致整个 DAG 被标为失败但数据已部分写入
DAG 运行时报 “No module named ‘pandas’” 怎么办
Airflow worker 进程用的是它自己的 Python 环境，跟你的 venv 或 conda 环境完全无关。即使你在本地 pip install pandas 成功，Airflow 可能根本没装。
解决路径只有两条：
- 在 Airflow 所在服务器上，用运行
airflow命令的同一用户，执行：pip install pandas openpyxl sqlalchemy（注意：不要用sudo pip，权限错位会导致 worker 找不到包） - 更可靠的做法是构建自定义 Docker 镜像，在
Dockerfile中FROM apache/airflow:2.8.1后追加RUN pip install "pandas>=1.5" "openpyxl>=3.1"，然后部署该镜像 —— 这能彻底避免环境漂移 - 验证是否生效：进 Airflow Web UI → Admin → Plugins → 点 “Trigger DAG” 旁的 “Test” 按钮，选一个最小 task，点 “Run” 查看日志里是否有
import pandas成功输出
真正麻烦的不是写通第一个 DAG，而是当清洗逻辑变复杂（比如要 join 十张表）、数据量从 MB 级涨到 GB 级、或需要对接 Hive/Spark 时，pandas 的单机内存模型和 Airflow 的 task 粒度会迅速成为瓶颈。这时候得提前想好退路：比如用 SparkKubernetesOperator 替换 PythonOperator，或把核心计算下沉到 dbt。但起步阶段，老实用好 pandas + Airflow 的组合，先让流程跑稳，比追求架构“先进”更重要。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
