从SequentialExecutor切换到LocalExecutor时出现报错求助

问题说明 #

切换到LocalExecutor后，调度器抛出以下错误，任务持续处于排队状态：

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/spawn.py", line 116, in spawn_main
    exitcode = _main(fd, parent_sentinel)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/spawn.py", line 126, in _main
    self = reduction.pickle.load(from_parent)
AttributeError: 'QueuedLocalWorker' object has no attribute 'wrapper'

使用SequentialExecutor时无异常，环境配置为Airflow 2.10.3 + Python 3.10虚拟环境。

解决方案 #

1. 升级Airflow到最新稳定版 #

该错误是Airflow 2.10.x版本的已知bug，后续补丁版本（如2.10.4及以上）已修复。执行以下命令升级：

pip install --upgrade apache-airflow==2.10.4

升级完成后重启Airflow服务：

airflow scheduler restart
airflow webserver restart

2. 临时应急：修改Airflow核心代码（不推荐长期使用） #

若暂时无法升级，找到虚拟环境中local_executor.py文件（路径示例：{虚拟环境路径}/lib/python3.10/site-packages/airflow/executors/local_executor.py），在QueuedLocalWorker类的__init__方法中添加wrapper属性初始化：

class QueuedLocalWorker(BaseWorker):
    def __init__(self, executor: LocalExecutor, queue: Queue[TaskInstance]):
        super().__init__()
        self.executor = executor
        self.queue = queue
        self.wrapper = None  # 添加此行代码
        self._running = True

保存修改后重启调度器即可临时解决序列化失败问题，但后续升级Airflow会覆盖该修改，因此优先推荐版本升级方案。

3. 校验虚拟环境完整性 #

若上述方案无效，可尝试重新安装Airflow依赖，确保虚拟环境无损坏：

pip install --force-reinstall apache-airflow==2.10.3

若仍有问题，建议创建全新虚拟环境重新部署Airflow。

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