Airflow初始化阶段如何注入补丁替换dataclass.asdict解决defaultdict兼容问题

Airflow有多个官方支持的初始化入口可以用来挂载这个全局补丁，以下是几种常用的可靠方式：

1. 使用airflow_local_settings.py（最推荐） #

这是Airflow官方提供的自定义初始化配置入口，Airflow的所有进程（Scheduler、Worker、Webserver、DAG解析器）启动时都会优先加载这个文件，完美满足"所有DAG运行前加载"的要求。
操作步骤：

• 在你的$AIRFLOW_HOME目录下创建airflow_local_settings.py文件
• 将你的补丁代码完整写入该文件即可
• 重启所有Airflow服务，补丁会自动全局生效

2. 放在DAG目录顶层的__init__.py #

Airflow在解析所有DAG文件之前，会先加载DAG根目录下的__init__.py，如果你的补丁只需要在DAG解析和任务运行阶段生效，可以用这个方案：

• 找到你配置的Airflow DAG根目录（默认是$AIRFLOW_HOME/dags）
• 编辑根目录下的__init__.py（没有就新建），把补丁代码写入该文件

3. 封装为独立补丁模块显式导入 #

如果你不想做全局替换，可以把补丁封装为独立模块，在所有用到asdict的业务代码开头导入即可：

• 新建dataclass_patch.py文件，写入你的补丁代码：

import copy
from collections import defaultdict
from dataclasses import _is_dataclass_instance, fields, asdict

def my_asdict(obj, dict_factory=dict):
    if _is_dataclass_instance(obj):
        result = []
        for f in fields(obj):
            value = my_asdict(getattr(obj, f.name), dict_factory)
            result.append((f.name, value))
        return dict_factory(result)
    elif isinstance(obj, (list, tuple)):
        return type(obj)(my_asdict(v, dict_factory) for v in obj)
    elif isinstance(obj, defaultdict):
        # This is the patch          
        obj = dict(obj)
    if isinstance(obj, dict):
        return type(obj)((my_asdict(k, dict_factory), my_asdict(v, dict_factory))
                         for k, v in obj.items())
    else:
        return copy.deepcopy(obj)

asdict = my_asdict

• 在需要用到asdict的代码最开头加上from dataclass_patch import asdict，替换原生的from dataclasses import asdict导入即可。

• 必须保证补丁代码的执行时机早于任何导入dataclasses.asdict的业务代码，否则已经提前导入的原生asdict不会被替换。
• 如果使用分布式执行器（CeleryExecutor、KubernetesExecutor等），需要保证补丁文件同步部署到所有Worker节点的对应路径下，确保Worker进程启动时也能加载补丁。
• 要是担心全局替换会影响Airflow自身的逻辑，可以选择第三种显式导入自定义my_asdict的方案，影响范围可控。

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