Google Functions中无法访问Cloud Storage存储桶文件怎么办

计划实现一套自动化数据解析清洗ETL任务，流程如下：

• 数据进入BUCKET1时自动触发函数启动
• 对数据完成解析、清洗、ETL处理
• 处理完成的数据保存至BUCKET2

当前已选择配置为BUCKET1变更触发的Google Cloud Functions，但部署后无法访问BUCKET1中的文件，无法定位问题：代码在本地Jupyter Notebook环境运行完全正常，部署到Google Functions后执行时，print(paths)返回空列表，无法加载paths列表，需要明确Google Storage中文件的正确访问方式。

原问题中贴出的存在问题的代码如下：

import pandas as pd
import glob
from google.cloud import storage
storage_client = storage.Client(project='MyProjectName')

paths = []
all_dfs = []

def hello_gcs(event, context):
    #"""Triggered by a change to a Cloud Storage bucket.
    #Args:
    #     event (dict): Event payload.
    #     context (google.cloud.functions.Context): Metadata for the event.
    #"""
    #file = event
    #print(f"Processing file: {file['name']}.")

    for files in glob.glob("gs:/BUCKET1/*/*.csv"):
        paths.append(files)
    print(paths)
    print("testingtesting")
    for i in range(len(paths)):

    temp_list = (paths[i].split("_"))
    date_temp_list = (temp_list[2])
    read_date = (date_temp_list.split("T")[0])

    globals()['table%s' % i] = pd.read_csv('{}' .format(paths[i]), index_col=None, header=0) #create new dfs based on subfolder structure
    globals()['table%s' % i]["Read Date"] = read_date
    all_dfs.append(globals()['table%s' % i])

output_df = pd.concat(all_dfs, axis=0, ignore_index=True)
output_df.to_csv("gs:/BUCKET2/Filename.csv")

本地Jupyter能跑通、Cloud Functions里返回空列表，核心原因有3个：

1. glob模块无法识别GCS路径：glob是Python标准库中用来扫描本地文件系统的工具，本身不支持gs://开头的对象存储路径。本地环境能跑是因为本地大概率做了GCS桶挂载、或者装了路径适配插件，Cloud Functions的沙箱运行时没有这类适配，glob会直接在实例本地磁盘查找对应路径，自然返回空。
2. GCS路径格式写错：标准GCS路径前缀是gs://（双斜杠），原代码里写的gs:/（单斜杠）本身就是非法路径。
3. 代码存在缩进语法错误：for i in range(len(paths)): 下的逻辑块没有缩进，就算路径能读到也会直接报语法错误无法运行。

额外注意：原代码把paths、all_dfs定义成了全局变量，Cloud Functions会复用冷启动后的实例，全局变量残留会导致多次触发时数据重复、结果错误。

在修改代码前先确认两项配置：

• 给Cloud Functions绑定的运行时服务账号，授予BUCKET1的存储对象读取权限、BUCKET2的存储对象写入权限
• 在函数的requirements.txt依赖文件中加入以下包，确保pandas能正常读写GCS文件：

pandas
gcsfs
google-cloud-storage

不要用glob扫描GCS文件，改用官方Cloud Storage客户端提供的列表接口枚举符合规则的文件，修正路径格式和缩进问题，移除有污染风险的全局变量：

import pandas as pd
from google.cloud import storage

storage_client = storage.Client(project='MyProjectName')
SRC_BUCKET = "BUCKET1"
DST_BUCKET = "BUCKET2"
OUTPUT_FILENAME = "Filename.csv"

def hello_gcs(event, context):
    """Triggered by a change to a Cloud Storage bucket.
    Args:
         event (dict): Event payload.
         context (google.cloud.functions.Context): Metadata for the event.
    """
    # 可直接从事件中拿到触发本次运行的新增文件名，不需要每次全量扫桶
    print(f"触发本次执行的文件: {event['name']}")

    # 用官方客户端枚举源桶下两级目录内所有csv文件
    src_bucket = storage_client.bucket(SRC_BUCKET)
    matched_blobs = src_bucket.list_blobs(match_glob="*/*.csv")
    file_paths = [f"gs://{SRC_BUCKET}/{blob.name}" for blob in matched_blobs]
    print(f"待处理文件列表: {file_paths}")

    all_dfs = []
    for path in file_paths:
        # 从文件名提取日期字段
        temp_segs = path.split("_")
        date_seg = temp_segs[2]
        read_date = date_seg.split("T")[0]

        # 读取文件、追加日期列
        df = pd.read_csv(path, index_col=None, header=0)
        df["Read Date"] = read_date
        all_dfs.append(df)

    # 合并所有数据写入目标桶
    final_df = pd.concat(all_dfs, axis=0, ignore_index=True)
    output_path = f"gs://{DST_BUCKET}/{OUTPUT_FILENAME}"
    final_df.to_csv(output_path, index=False)
    print(f"处理完成，结果已保存至: {output_path}")

如果业务逻辑不需要每次触发都处理桶内所有历史csv文件，完全可以去掉枚举全量文件的逻辑，直接处理event['name']对应的单个新增文件，运行速度更快，也不会重复处理历史数据。

内容的提问来源于stack exchange，提问作者Martin Walczyński