如何在Airflow中通过编程设置全局环境变量并跨Operator访问？

用户提供的DAG代码如下：

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago
from airflow.operators.python import PythonOperator

def returnSum():
    BashOperator(task_id="running",
    bash_command="export value={var};".format(var=3))

with DAG(
    dag_id="test_python",
    schedule_interval=None,
    catchup=False,
    start_date=days_ago(1)) as dag:

    pyth=PythonOperator(
        task_id="read_from_aws_sm",
        python_callable=returnSum,
        provide_context=True
    )
    ba =  BashOperator(task_id="running_bash", bash_command="echo $value;")

    pyth >> ba   

用户尝试在Python方法中通过BashOperator设置环境变量，并在另一个BashOperator中访问该值，但输出为空，需要解决这个问题。

问题原因 #

1. 任务进程隔离：Airflow的每个任务都是独立运行的进程，进程间的环境变量完全隔离，无法直接共享。
2. 代码逻辑错误：在PythonOperator的returnSum函数中仅创建了BashOperator实例，并未实际执行该任务；就算执行了，其设置的环境变量也只存在于该进程内部，无法传递到后续的running_bash任务。

解决方案：使用XCom实现任务间数据传递 #

Airflow内置的XCom机制是任务间传递小量数据的标准方式，以下是两种可行的实现方式：

方式1：通过XCom传递值，直接在Bash命令中引用 #

修改代码，让PythonOperator的可调用函数返回要传递的值，Airflow会自动将该值推送到XCom，后续BashOperator通过模板语法获取：

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from airflow.operators.python import PythonOperator

def returnSum():
    # 返回要传递的值，自动推送到XCom
    return 3

with DAG(
    dag_id="test_python",
    schedule_interval=None,
    catchup=False,
    start_date=days_ago(1)) as dag:

    pyth = PythonOperator(
        task_id="read_from_aws_sm",
        python_callable=returnSum
    )
    # 通过模板语法从XCom拉取值并输出
    ba = BashOperator(
        task_id="running_bash",
        bash_command="echo {{ ti.xcom_pull(task_ids='read_from_aws_sm') }};"
    )

    pyth >> ba

方式2：将XCom值设置为BashOperator的环境变量 #

如果需要通过环境变量的方式访问，可以将XCom获取到的值赋值给BashOperator的env参数：

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from airflow.operators.python import PythonOperator

def returnSum(**context):
    # 手动推送带key的XCom，避免值混淆
    value = 3
    context['ti'].xcom_push(key='target_value', value=value)

with DAG(
    dag_id="test_python",
    schedule_interval=None,
    catchup=False,
    start_date=days_ago(1)) as dag:

    pyth = PythonOperator(
        task_id="read_from_aws_sm",
        python_callable=returnSum,
        provide_context=True  # 需要上下文来访问ti对象
    )
    # 将XCom值设置为环境变量MY_VALUE，在bash命令中访问
    ba = BashOperator(
        task_id="running_bash",
        bash_command="echo $MY_VALUE;",
        env={
            "MY_VALUE": "{{ ti.xcom_pull(task_ids='read_from_aws_sm', key='target_value') }}"
        }
    )

    pyth >> ba

注意事项 #

• XCom适合传递小量数据（如配置值、计算结果），不要用它传递大文件或大量数据。
• 如果多个任务都推送XCom，建议指定唯一的key来区分，避免拉取错误的值。

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