Airflow 2中BashOperator执行conda activate报错CommandNotFoundError求助

问题原因 #

每个BashOperator都是独立子进程运行，t1中source conda.sh的环境变量仅在当前子进程生效，无法传递到后续t2任务。同时Airflow默认使用非交互式shell，而conda activate需要shell加载conda的函数配置才能正常运行。

解决方案 #

方案1：合并命令到单个BashOperator #

将环境初始化、激活和后续命令放在同一个任务中，确保在同一个子进程内完成：

from airflow.decorators import dag
from airflow.operators.bash import BashOperator
from datetime import datetime

@dag(start_date=datetime(2022, 8, 10), schedule_interval='* * * * *', catchup=False)
def dag_bash():
    run_env_task = BashOperator(
        task_id='run_in_conda_env',
        bash_command='source ~/miniconda3/etc/profile.d/conda.sh && conda activate ukjo && echo $CONDA_DEFAULT_ENV'
    )

dag = dag_bash()

方案2：直接调用conda环境的绝对路径 #

无需激活环境，直接使用目标环境下的可执行文件绝对路径执行命令，这是生产环境更可靠的方式：

from airflow.decorators import dag
from airflow.operators.bash import BashOperator
from datetime import datetime

@dag(start_date=datetime(2022, 8, 10), schedule_interval='* * * * *', catchup=False)
def dag_bash():
    check_env_task = BashOperator(
        task_id='check_conda_env',
        bash_command='~/miniconda3/envs/ukjo/bin/echo $CONDA_DEFAULT_ENV'
    )

dag = dag_bash()

如果要执行Python脚本，直接写：~/miniconda3/envs/ukjo/bin/python your_script.py

方案3：使用交互式shell（不推荐） #

通过指定交互式bash运行命令，强制加载conda配置，但会带来额外环境加载开销：

from airflow.decorators import dag
from airflow.operators.bash import BashOperator
from datetime import datetime

@dag(start_date=datetime(2022, 8, 10), schedule_interval='* * * * *', catchup=False)
def dag_bash():
    activate_env_task = BashOperator(
        task_id='activate_conda_env',
        bash_command='bash -i -c "conda activate ukjo && echo $CONDA_DEFAULT_ENV"'
    )

dag = dag_bash()

关键提示 #

Airflow的任务都是独立进程，进程间环境变量不共享，跨任务传递环境变量需要用XCom等持久化方式，不要依赖shell环境的传递。生产环境优先选择方案1或方案2。

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