Airflow如何用PythonOperator执行gcsfuse挂载的Python脚本

不需要导入目标脚本内的函数，也不需要额外修改Airflow导入路径配置，直接通过Python原生逻辑执行gcsfuse挂载目录下的目标文件即可，以下两种实现均适配你的目录场景，同时能解决BashOperator因Python解释器环境不匹配导致的库函数报错问题。

方法1：通过exec直接加载执行脚本（无额外进程开销，环境完全对齐） #

该方式直接读取脚本内容在PythonOperator的当前运行上下文中执行，和直接运行脚本的效果一致，完全复用Airflow worker的Python依赖环境。

from airflow.operators.python import PythonOperator
import os

def execute_target_script(**context):
    script_full_path = "/home/airflow/gcsfuse/dags/external/projectXYZ/test.py"
    # 提前校验文件存在性，避免运行时抛出模糊的找不到文件错误
    if not os.path.isfile(script_full_path):
        raise FileNotFoundError(f"待执行脚本路径不存在: {script_full_path}")
    
    # Python3 环境下执行指定路径脚本的标准写法
    with open(script_full_path, "r", encoding="utf-8") as f:
        exec(compile(f.read(), script_full_path, "exec"))

run_python = PythonOperator(
    task_id="run_python",
    python_callable=execute_target_script
    # 若需要给脚本传参，可通过op_kwargs配置后在context中获取
)

方法2：通过subprocess调用当前解释器执行（和原脚本运行逻辑完全一致） #

如果担心exec执行时的全局变量上下文和直接启动脚本运行有差异，可以通过sys.executable获取当前Airflow worker使用的Python解释器路径，启动子进程执行脚本，彻底规避BashOperator硬写python命令指向错误解释器的问题。

from airflow.operators.python import PythonOperator
import sys
import subprocess
import os

def execute_target_script(**context):
    script_full_path = "/home/airflow/gcsfuse/dags/external/projectXYZ/test.py"
    if not os.path.isfile(script_full_path):
        raise FileNotFoundError(f"待执行脚本路径不存在: {script_full_path}")
    
    # sys.executable自动获取当前运行环境的Python解释器路径，不会错用系统全局Python
    exec_result = subprocess.run(
        [sys.executable, script_full_path],
        capture_output=True,
        text=True,
        check=True
    )
    # 打印输出方便在Airflow日志中排查问题
    if exec_result.stdout:
        print("脚本运行输出:\n", exec_result.stdout)
    if exec_result.stderr:
        print("脚本运行错误信息:\n", exec_result.stderr)

run_python = PythonOperator(
    task_id="run_python",
    python_callable=execute_target_script
)

排查说明：此前BashOperator出现库函数报错，核心原因是Bash执行时未加载Airflow运行的虚拟环境变量，硬写的python命令指向了系统全局Python解释器，该解释器下没有安装你需要的依赖库。以上两种写法均直接复用Airflow worker的Python环境，可直接验证是否为环境不匹配导致的错误。如果需要确认解释器是否正确，可以在test.py中添加print(sys.executable)打印解释器路径对比即可。

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