在Google Cloud DAG中运行Python脚本失败：文件找不到或导入错误求助

核心问题说明 #

Airflow的BashOperator运行在worker节点的容器内，直接使用GCS桶的路径（如/europe-west8-cellxpert-blablabla-bucket/...）是无效的，因为容器本地没有挂载这个GCS路径。另外DAGS_FOLDER环境变量如果未在Airflow配置中显式设置，会触发KeyError。

可行解决方案 #

方案1：先下载脚本到worker本地再执行 #

用GoogleCloudStorageToLocalOperator把GCS里的脚本同步到worker本地目录，再通过BashOperator执行：

from airflow.providers.google.cloud.transfers.gcs_to_local import GoogleCloudStorageToLocalOperator
from airflow.operators.bash import BashOperator

# 下载GCS脚本到worker临时目录
download_script = GoogleCloudStorageToLocalOperator(
    task_id="download_python_script",
    bucket="europe-west8-cellxpert-blablabla-bucket",
    object_name="dags/scripts/cellxpert_transactions_script.py",
    filename="/tmp/cellxpert_transactions_script.py",
    dag=dag
)

# 执行本地脚本
execute_script = BashOperator(
    task_id="execute_python_code",
    bash_command="python /tmp/cellxpert_transactions_script.py",
    dag=dag
)

# 设置任务依赖
download_script >> execute_script

方案2：用PythonOperator直接加载GCS脚本执行（推荐） #

如果脚本逻辑不依赖本地环境，可通过gcsfs直接读取脚本内容并执行，省去文件下载步骤：

from airflow.operators.python import PythonOperator
import gcsfs

def run_gcs_script():
    fs = gcsfs.GCSFileSystem(project="你的GCP项目ID")
    with fs.open("europe-west8-cellxpert-blablabla-bucket/dags/scripts/cellxpert_transactions_script.py", "r") as f:
        script_content = f.read()
    exec(script_content)

execute_script = PythonOperator(
    task_id="execute_python_code",
    python_callable=run_gcs_script,
    dag=dag
)

注意：需确保Airflow worker安装了gcsfs包，可通过requirements.txt添加依赖。

方案3：挂载GCS桶到Airflow worker（持久化配置） #

如果需要频繁访问GCS中的脚本，可通过Cloud Storage FUSE将GCS桶挂载到worker本地目录：

1. 在Airflow worker节点安装Cloud Storage FUSE
2. 将桶挂载到指定路径，比如/mnt/gcs-bucket
3. 然后在BashOperator中使用挂载后的路径：

execute_script = BashOperator(
    task_id="execute_python_code",
    bash_command="python /mnt/gcs-bucket/dags/scripts/cellxpert_transactions_script.py",
    dag=dag
)

此方法需维护worker节点的挂载配置，适合长期稳定的使用场景。

环境变量使用补充 #

如果一定要用DAGS_FOLDER，需在Airflow部署时显式注入环境变量，比如设置AIRFLOW__CORE__DAGS_FOLDER=/path/to/dags，之后在DAG中安全获取：

import os
DAGS_FOLDER = os.getenv("AIRFLOW__CORE__DAGS_FOLDER", "/default/dags/path")

注意：该路径是worker本地的DAG目录，不是GCS桶路径，仍需先同步GCS文件到该目录才能执行。

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