如何用Python生成器流式上传数据至Google Cloud Storage？

官方推荐的实现方式是利用Google Cloud Storage Python客户端库的upload_from_file方法，配合自定义类文件对象包装生成器，实现无本地磁盘缓冲的流式上传，完全不需要依赖gsutil子进程。

步骤1：安装依赖 #

确保安装官方GCS客户端库：

pip install google-cloud-storage

步骤2：自定义类文件对象包装生成器 #

GCS的upload_from_file方法接受实现了read()方法的类文件对象，我们可以把生成器包装成这样的对象，实现流式读取：

class GeneratorIO:
    def __init__(self, generator):
        self.generator = generator
        self.buffer = b""

    def read(self, size=-1):
        # 填充缓冲区直到达到指定大小，或生成器耗尽
        while len(self.buffer) < size or size == -1:
            try:
                chunk = next(self.generator)
                if not chunk:
                    break
                self.buffer += chunk
            except StopIteration:
                break
        
        # 返回指定大小的数据，剩余部分留在缓冲区
        if size == -1:
            result = self.buffer
            self.buffer = b""
        else:
            result = self.buffer[:size]
            self.buffer = self.buffer[size:]
        return result

步骤3：实现流式上传逻辑 #

使用GCS客户端将包装后的生成器数据上传：

from google.cloud import storage

def stream_to_gcs(bucket_name, target_blob_path, data_generator):
    # 初始化GCS客户端（凭据需提前配置，比如环境变量或gcloud本地配置）
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(target_blob_path)

    # 将生成器转为类文件对象
    stream = GeneratorIO(data_generator)

    # 执行流式上传，chunk_size控制分块大小（建议1-10MB，按需调整）
    blob.upload_from_file(stream, chunk_size=1024*1024)  # 1MB分块

# 示例：模拟生成大量字节数据的生成器
def large_data_generator():
    for i in range(1000):
        yield f"Chunk {i}: Sample streaming data\n".encode("utf-8")

# 调用示例
if __name__ == "__main__":
    stream_to_gcs("your-bucket-name", "streamed-data.txt", large_data_generator())

关键说明 #

• 无内存/磁盘缓冲：GeneratorIO每次仅从生成器读取数据填充缓冲区，不会把全部数据加载到内存或写入本地磁盘。
• 分块上传自动处理：upload_from_file会根据设置的chunk_size自动进行分块上传，GCS支持未知总大小的流式上传。
• 凭据配置：确保GCS客户端能正确获取凭据，可通过设置GOOGLE_APPLICATION_CREDENTIALS环境变量指向服务账号密钥文件，或使用gcloud auth application-default login配置本地凭据。

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