如何在Python脚本中高效用gsutil跨GCS存储桶复制指定文件列表

问题根源 #

你之前的代码错误地将管道命令作为printf的参数传递，且在shell=False模式下管道语法不生效，同时超长的文件列表字符串作为命令行参数触发了系统的Argument list too long限制。正确的做法是利用gsutil cp -I从标准输入读取文件列表的特性，避免将所有文件路径作为命令行参数传递。

高效实现方法 #

方法1：一次性通过stdin传递文件列表

适合文件列表规模中等的场景，直接将所有文件路径按换行分隔后传入gsutil的标准输入：

import subprocess

# 生成目标文件列表（注意：如果是GCS源文件，路径需写全，比如gs://source_bucket/xxx.txt）
file_list = [f'gs://source_bucket/{_}.txt' for _ in range(1000000)]
# 转换为每行一个文件的字节流
input_content = '\n'.join(file_list).encode('utf-8')

# 执行复制命令
try:
    subprocess.run(
        ['gsutil', '-m', 'cp', '-I', 'gs://target_bucket'],
        input=input_content,
        check=True,
        capture_output=True
    )
    print("复制完成")
except subprocess.CalledProcessError as e:
    print(f"复制失败：{e.stderr.decode('utf-8')}")

方法2：逐行写入stdin（低内存占用）

针对超大规模文件列表（如百万级以上），避免一次性加载所有路径到内存，采用逐行写入的方式：

import subprocess

# 打开gsutil进程，通过管道传入数据
with subprocess.Popen(
    ['gsutil', '-m', 'cp', '-I', 'gs://target_bucket'],
    stdin=subprocess.PIPE,
    text=True
) as proc:
    # 逐行写入文件路径（可替换为从文件/数据库读取的逻辑）
    for num in range(1000000):
        proc.stdin.write(f'gs://source_bucket/{num}.txt\n')
    # 关闭输入流，等待进程完成
    proc.stdin.close()
    return_code = proc.wait()
    if return_code == 0:
        print("复制完成")
    else:
        print(f"复制失败，返回码：{return_code}")

关键注意事项 #

• 确保源文件路径是完整的GCS路径（gs://bucket_name/file_path），否则gsutil会尝试读取本地文件。
• -m参数开启多线程复制，能大幅提升大文件/多文件场景下的复制效率。
• 避免使用shell=True执行管道命令，既存在安全风险（如注入），也容易触发参数长度限制。

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