如何解决Apache Airflow中因Worker节点依赖导致的DAG导入错误？

这是Airflow容器化分布式部署里非常典型的痛点——你把依赖拆分到Worker的思路本身没问题，但忽略了一个关键细节：Scheduler和Webserver在加载、解析DAG文件时，会执行DAG文件中的顶层代码，如果你的DAG里直接在顶层导入了crypto这类仅Worker才有的依赖，自然会触发ModuleNotFoundError。

下面是几种实用的解决方案，按推荐优先级排序：

1. 延迟导入（Lazy Import）——最推荐的轻量方案 #

核心思路是把依赖导入从DAG文件的顶层，移到任务执行时才会触发的代码块里，这样Scheduler/Webserver解析DAG时不会触及这些导入逻辑，只有Worker运行任务时才会加载依赖。

举个代码改造的例子：

改造前（会报错） #

# 顶层导入，Scheduler/Webserver解析时会执行
import crypto
from airflow import DAG
from airflow.operators.python import PythonOperator

def process_data():
    crypto.encrypt("some_data")

with DAG(dag_id="crypto_dag", schedule_interval=None) as dag:
    PythonOperator(task_id="encrypt_task", python_callable=process_data)

改造后（正常运行） #

from airflow import DAG
from airflow.operators.python import PythonOperator

def process_data():
    # 只有任务执行时才导入依赖
    import crypto
    crypto.encrypt("some_data")

with DAG(dag_id="crypto_dag", schedule_interval=None) as dag:
    PythonOperator(task_id="encrypt_task", python_callable=process_data)

如果是自定义Operator，把依赖导入放到Operator的execute()方法内部即可，原理完全一致。

2. 为Scheduler/Webserver安装最小必要依赖 #

如果你的DAG顶层代码必须用到某个依赖（比如用crypto生成任务ID、定义常量等），不需要给Scheduler/Webserver安装Worker的全量依赖，只需要安装这个特定的依赖包即可。

比如在Scheduler/Webserver的Dockerfile里添加：

RUN pip install pycryptodome  # 注意：crypto可能是pycryptodome的别名，根据实际包名调整

⚠️ 注意：务必保证Scheduler/Webserver安装的依赖版本和Worker一致，避免出现版本兼容问题。

3. 启用DAG序列化（Airflow 2.x+） #

Airflow 2.x引入了DAG序列化特性，开启后Scheduler会把解析后的DAG序列化为JSON存到元数据库中，Webserver直接从数据库读取序列化后的DAG，不再需要解析原始DAG文件——这意味着Webserver完全不需要DAG的依赖。

开启方法： #

修改airflow.cfg配置文件：

[core]
dag_serialization_enabled = True

⚠️ 注意：Scheduler仍然需要解析原始DAG文件来完成序列化，所以如果DAG顶层有依赖导入，Scheduler还是需要安装对应的依赖；但Webserver可以彻底摆脱DAG依赖。这个方案适合大规模集群部署，能显著降低Webserver的资源消耗。

4. 拆分DAG+跨DAG任务依赖 #

如果某个任务的依赖非常特殊（比如需要大量冷门库、特定版本的环境），可以把这个任务拆分到单独的DAG中，专门部署对应依赖的Worker来执行它，然后用ExternalTaskSensor让主DAG等待这个子DAG的任务完成。

这种方案会增加DAG的管理复杂度，适合依赖极端特殊、无法用前三种方案解决的场景。

内容的提问来源于stack exchange，提问作者Marco Miduri