Airflow Variables中是否支持将JSON键作为变量引用到另一JSON值中

Airflow原生Variables机制不直接支持JSON结构内部的自引用，无需提前对JSON文件做字符串替换，有两种简便的实现方案：

方案1：借助Airflow原生Jinja模板能力 #

你可以直接在JSON的值里写Airflow内置的Jinja变量引用表达式，调整后的合法JSON配置如下：

{
  "my_variables":
   {
    "s3_bucket": "s3://my_prod_bucket",
    "dataset_1": "{{ var.json.my_variables.s3_bucket }}/dataset_1_path/",
    "dataset_2": "{{ var.json.my_variables.s3_bucket }}/dataset_2_path/"
    }
}

• 在DAG支持Jinja模板的参数位置（比如各类Operator的模板字段）直接调用该变量，Airflow会自动完成嵌套解析，无需额外处理。
• 如果是在不支持Jinja的普通Python代码中使用，可以手动调用Airflow的渲染接口完成解析，示例代码如下：

from airflow.models import Variable
from airflow.template.templater import Templater

# 读取原始JSON变量
my_vars = Variable.get("my_variables", deserialize_json=True)
# 手动渲染目标字段
rendered_path = Templater().render_string(my_vars["dataset_1"], context={})

方案2：封装统一的变量解析工具类 #

如果不想在每个调用位置单独处理渲染，可以封装一个轻量的解析工具，读取变量时自动递归完成所有内部引用的替换，示例实现如下：

import re
from airflow.models import Variable

# 匹配变量引用表达式的正则
VAR_REF_PATTERN = re.compile(r"{{\s*var\.json\.([a-zA-Z0-9_.]+)\s*}}")

def get_parsed_json_var(var_key: str):
    raw_var = Variable.get(var_key, deserialize_json=True)
    
    def _parse_val(val):
        if isinstance(val, str):
            match = VAR_REF_PATTERN.search(val)
            if match:
                # 按路径查找引用的变量值
                ref_path = match.group(1).split(".")
                ref_val = raw_var
                for k in ref_path:
                    ref_val = ref_val[k]
                return val.replace(match.group(0), ref_val)
        elif isinstance(val, dict):
            return {k: _parse_val(v) for k, v in val.items()}
        return val
    
    return _parse_val(raw_var)

调用时直接执行get_parsed_json_var("my_variables")即可拿到所有引用都替换完成的字典。

注意事项 #

• 不要配置循环引用（比如a引用b、b同时引用a），会导致解析报错
• 该方案同样适配Variables存储在第三方Secret Manager的场景，无需调整存储逻辑

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