Airflow任务组无法访问Python全局变量的问题及报错咨询

我用Python Operator调用的Python函数里定义了全局变量cluster_key和new_key，函数内通过Variable.set把这两个变量对应的内容存成Airflow变量。但在TaskGroup里尝试用Variable.get(cluster_key)获取对应变量时，出现KeyError: 'Variable does not exist'；如果在TaskGroup作用域内声明global cluster_key，又会触发SyntaxError: 'name 'cluster_key' is assigned to before global declaration'。

相关代码如下：

初始函数代码 #

cluster_key = ""
new_key =""

def dynamic_list(e_run_id):
    global cluster_key
    cluster_key = 'key_cluster_'+e_run_id+''
    global new_key
    new_key = 'new_key_'+e_run_id+''
    # ... 其他业务逻辑
    Variable.set(cluster_key, keys)
    Variable.set(new_key, new_values)

DAG及TaskGroup代码 #

with DAG(
        dag_id=JOB_NAME,
        default_args=default_args,
        start_date=yesterday,
)as main_dag:   
    
    groups = []
    with TaskGroup(group_id='Dynamic_dataproc_Processing') as dataprocs_jobs:
        start = BashOperator(task_id="start", bash_command="sleep 1m")

        sub_groups = []
        with TaskGroup('dataproc_create_cluster', prefix_group_id=False) as dataproc_create_clusters:
           for i in list(Variable.get(cluster_key)):
               dynmaic_create_cluster = DataprocCreateClusterOperator(
                    task_id="create_cluster_{0}".format(str(i)),
                    project_id='{0}'.format(project_id1),
                    cluster_config=CLUSTER_GENERATOR_CONFIG,
                    region='{0}'.format(REGION),
                    cluster_name="dataproc-clustersrc-{0}-src-{1}".format(SRC_GROUP, (i)),
                    sla=timedelta(minutes=5)
                )

1. DAG解析时机冲突：Airflow在启动或刷新时会提前解析DAG文件，这时候cluster_key还是初始的空字符串，Variable.get(cluster_key)实际是在查找空字符串对应的Airflow变量，自然不存在，所以报KeyError。
2. 全局变量作用域误解：cluster_key是在Python Operator运行时才会被赋值，但DAG解析阶段这个变量还没被修改；而且在TaskGroup里声明global时，因为DAG解析时已经读取过cluster_key的空值，所以触发语法错误。

方案1：用XCom传递变量名，延迟任务生成 #

把cluster_key通过XCom从Python Operator传递到后续任务，再用PythonOperator动态生成TaskGroup内的任务（避免在DAG解析阶段直接循环）：

1. 修改Python函数，将变量名推送到XCom：

def dynamic_list(e_run_id, **context):
    cluster_key = 'key_cluster_' + e_run_id
    new_key = 'new_key_' + e_run_id
    # ... 其他业务逻辑
    Variable.set(cluster_key, keys)
    Variable.set(new_key, new_values)
    # 把变量名推送到XCom供后续任务获取
    context['ti'].xcom_push(key='cluster_key', value=cluster_key)

2. 新增PythonOperator动态生成集群任务：

def generate_cluster_tasks(**context):
    # 从XCom获取之前生成的cluster_key
    cluster_key = context['ti'].xcom_pull(task_ids='dynamic_list_task', key='cluster_key')
    # 获取对应的Airflow变量值
    cluster_values = Variable.get(cluster_key)
    
    # 动态创建TaskGroup和任务
    sub_group = TaskGroup('dataproc_create_cluster', prefix_group_id=False)
    with sub_group:
        for i in list(cluster_values):
            DataprocCreateClusterOperator(
                task_id=f"create_cluster_{str(i)}",
                project_id=project_id1,
                cluster_config=CLUSTER_GENERATOR_CONFIG,
                region=REGION,
                cluster_name=f"dataproc-clustersrc-{SRC_GROUP}-src-{i}",
                sla=timedelta(minutes=5)
            )
    return sub_group

with DAG(
    dag_id=JOB_NAME,
    default_args=default_args,
    start_date=yesterday,
) as main_dag:   
    # 先执行生成变量的Python任务
    dynamic_task = PythonOperator(
        task_id='dynamic_list_task',
        python_callable=dynamic_list,
        op_kwargs={'e_run_id': '{{ run_id }}'},
        provide_context=True
    )

    with TaskGroup(group_id='Dynamic_dataproc_Processing') as dataprocs_jobs:
        start = BashOperator(task_id="start", bash_command="sleep 1m")
        # 调用动态生成任务的函数
        generate_clusters = PythonOperator(
            task_id='generate_cluster_tasks',
            python_callable=generate_cluster_tasks,
            provide_context=True,
            do_xcom_push=False
        )
        start >> generate_clusters

    # 设置任务依赖：先生成变量，再执行后续集群任务
    dynamic_task >> dataprocs_jobs

方案2：DAG解析阶段直接构造变量名（适用于固定e_run_id） #

如果e_run_id是固定值（比如固定字符串、可提前计算的日期），可以直接在DAG解析阶段构造cluster_key，不需要全局变量：

# 直接构造cluster_key（比如用固定的run_id或提前计算的值）
e_run_id = "固定值"  # 或者用 '{{ run_id }}' 但注意模板渲染时机
cluster_key = 'key_cluster_' + e_run_id
new_key = 'new_key_' + e_run_id

def dynamic_list():
    # ... 其他业务逻辑
    Variable.set(cluster_key, keys)
    Variable.set(new_key, new_values)

with DAG(...) as main_dag:
    # ... 其他代码
    with TaskGroup('dataproc_create_cluster', prefix_group_id=False) as dataproc_create_clusters:
        # 加异常处理避免解析时变量不存在报错
        try:
            cluster_values = Variable.get(cluster_key)
            for i in list(cluster_values):
                DataprocCreateClusterOperator(
                    task_id=f"create_cluster_{str(i)}",
                    project_id=project_id1,
                    cluster_config=CLUSTER_GENERATOR_CONFIG,
                    region=REGION,
                    cluster_name=f"dataproc-clustersrc-{SRC_GROUP}-src-{i}",
                    sla=timedelta(minutes=5)
                )
        except KeyError:
            # 变量未创建时添加占位任务
            BashOperator(task_id='variable_not_ready', bash_command='echo "变量尚未生成"')

核心注意事项 #

• Airflow的DAG解析是离线执行的，在任务运行前就会完成，所以不能依赖运行时才赋值的变量来生成任务。
• 动态任务必须放在运行时执行的PythonOperator中，或者确保所需变量在DAG解析时已经存在。

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