pythonoperator 应仅作调度胶水，业务逻辑须抽离为独立模块；禁用硬编码连接/变量，参数通过op_kwargs传递；慎用provide_context，函数签名需显式声明**context；重试、连接交由airflow管理；cpu密集型任务应换用bashoperator或kubernetespodoperator。
PythonOperator 里别直接写业务逻辑
函数体太长、混着数据库操作和日志打印，DAG 文件一改就触发重调度，连带所有 task 都得重跑。这不是 Airflow 的设计意图——PythonOperator 只该做「调度层胶水」，不该当业务容器。
- 把核心逻辑抽成独立的 Python 模块（比如
etl/jobs.py），在 DAG 文件里只 import + 调用 - 确保被调用函数能单独测试：不依赖
context、不硬编码 conn_id、不读写本地文件路径 - 如果必须传参，用
op_kwargs或templates_dict，别在函数里硬写Variable.get("xxx")
用 provide_context=True 要想清楚后果
加了这个参数，Airflow 会把一堆运行时上下文（execution_date、task_instance、dag_run 等）打包进函数参数，看着方便，但容易让函数和 Airflow 运行时强耦合，单元测试写起来费劲，迁移出 DAG 也困难。
- 只在真需要动态获取
execution_date或task_instance.xcom_pull()时才开；其他情况优先用params或op_kwargs - 如果开了，函数签名必须显式声明
**context，别用*args, **kwargs模糊兜底——否则下次 Airflow 升级加个新 key，你的函数可能默默忽略它 -
context里的对象不能 pickle，别试图把它存进 XCom（会报TypeError: cannot serialize '_io.TextIOWrapper' object）
别在 PythonOperator 里手动管理连接或重试
常见错误是自己写 try/except 包裹数据库调用，再加 time.sleep(2) 循环重试。这绕过了 Airflow 的重试机制，导致 UI 上看不到失败次数、不会触发 on_failure_callback、XCom 也不清空。
- 用
retries=3和retry_delay=timedelta(seconds=30)让 Airflow 控制重试节奏 - 连接统一走 Airflow Connection：
PostgresHook(postgres_conn_id="my_db")，而不是手写psycopg2.connect(...) - 如果某个步骤必须“失败即跳过”，用
trigger_rule="all_done"+ 单独写个空PythonOperator做 fallback，别在主逻辑里 if-else 分支处理
PythonOperator 性能卡在 GIL 和序列化上
一个 PythonOperator 默认运行在 scheduler 启动的 worker 进程里，共享同一个 Python 解释器。CPU 密集型任务（比如 Pandas 大表计算）会卡住整个 worker；而函数参数、返回值要走 pickle 序列化，含 lambda、嵌套类或文件句柄的对象直接报错。
立即学习“Python免费学习笔记（深入）”；
- CPU 密集型任务换
BashOperator调外部脚本，或用KubernetesPodOperator隔离进程 - 返回大数据结构（如 DataFrame）前先
.to_dict("records")或存 S3，别直接 returndf - 避免在函数里定义嵌套函数或使用闭包——pickle 不认它们，会抛
AttributeError: Can't pickle local object
最常被忽略的是：DAG 文件每次被解析，都会执行一遍 PythonOperator 的 python_callable 定义语句（哪怕没调度）。所以别在函数定义外写耗时初始化，比如 model = load_model("path") —— 它会在 scheduler 内存里重复加载 N 次。
