如何在Cloud Run实例中读取Cloud Storage的CSV文件并转为pandas DataFrame

• 你当前的代码直接将文件下载到当前工作目录，Cloud Run 运行环境的默认工作目录为只读分区，download_blob 执行时就会抛出写入权限错误，后续自然无法读取文件，最终触发内部服务器错误。
• 你没有对下载操作做异常捕获，导致错误没有被正确打印，难以定位问题。

首先调整下载路径到 Cloud Run 唯一允许写入的 /tmp 临时目录即可，你原有下载函数无需改动，仅修改调用逻辑：

import pandas as pd
from google.cloud import storage

def download_blob(bucket_name, source_blob_name, destination_file_name):
    """从GCS存储桶下载文件"""
    # GCS存储桶ID
    # bucket_name = "你的存储桶名称"

    # GCS对象的ID
    # source_blob_name = "存储对象的路径"

    # 文件下载到本地的目标路径
    # destination_file_name = "本地文件路径"

    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    # 构造客户端侧的blob对象，注意Bucket.blob和Bucket.get_blob的区别是前者不会从GCS拉取额外元数据，不需要额外数据时优先使用
    blob = bucket.blob(source_blob_name)
    blob.download_to_filename(destination_file_name)
    print(f"已将存储对象 {source_blob_name} 从存储桶 {bucket_name} 下载到本地文件 {destination_file_name}")

# 调用示例
if __name__ == "__main__":
    bucket_name = "替换为你的存储桶名称"
    source_blob_name = "替换为GCS上CSV文件的路径"
    # 仅修改目标路径到/tmp目录
    local_file_path = "/tmp/testing.csv"
    download_blob(bucket_name, source_blob_name, local_file_path)
    # 正常读取文件
    df = pd.read_csv(local_file_path)
    # 后续处理逻辑

如果不想存储临时文件，可直接将GCS文件内容读取到内存转换为DataFrame，跳过写本地磁盘的步骤，性能更好：

import pandas as pd
from google.cloud import storage
from io import StringIO

def read_gcs_csv_to_df(bucket_name, source_blob_name):
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(source_blob_name)
    # 直接读取文件内容到内存
    content = blob.download_as_text()
    # 转换为DataFrame
    return pd.read_csv(StringIO(content))

# 调用示例
df = read_gcs_csv_to_df("替换为你的存储桶名称", "替换为GCS上CSV文件的路径")

• /tmp 目录的存储容量上限等于你给Cloud Run实例分配内存的一半，若CSV文件超过该上限会触发磁盘不足错误。
• 确保Cloud Run的服务账号拥有对应GCS存储桶的storage.objects.get权限，否则下载会因权限不足报错。
• 建议给下载、读取逻辑添加异常捕获，打印错误日志方便后续排查问题。

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