如何用PySpark/Spark迁移S3存储桶的文件夹、子文件夹及文件至另一桶？

一、是否可行？ #

完全可行。主流云存储（如S3、OSS、GCS）均支持跨存储桶对象复制，结合Spark/PySpark的分布式处理能力，可高效完成带完整目录结构的批量迁移——无需手动创建目录，因为云存储的“文件夹”本质是对象键的前缀，复制时保留前缀即可自动还原结构。

二、PySpark实现思路 #

核心逻辑 #

云存储的目录并非实体，而是对象键的前缀标识。只需遍历源桶A的所有对象，提取每个对象的完整键（路径），将其复制到目标桶B的指定迁移文件夹前缀下，就能自动保留原有的子文件夹层级。

具体实现步骤 #

1. 批量获取源桶对象列表

用云存储SDK（如S3的boto3）递归列举源桶所有对象的键，转为Spark RDD实现分布式处理：

import boto3
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("BucketMigration").getOrCreate()
s3_client = boto3.client('s3')

# 递归列举源桶所有对象键
def list_all_objects(bucket_name, prefix=""):
    paginator = s3_client.get_paginator('list_objects_v2')
    for page in paginator.paginate(Bucket=bucket_name, Prefix=prefix):
        if 'Contents' in page:
            for obj in page['Contents']:
                yield obj['Key']

# 转为RDD做分布式处理
source_obj_keys = spark.sparkContext.parallelize(list(list_all_objects("bucket-a")))

2. 分布式执行跨桶复制

对每个对象键构造目标路径（迁移文件夹前缀+原键），调用云存储的复制接口：

def copy_single_object(source_key):
    # 目标路径示例：将源对象复制到bucket-b的migrated_data文件夹下
    dest_key = f"migrated_data/{source_key}"
    s3_client.copy_object(
        Bucket="bucket-b",
        Key=dest_key,
        CopySource={"Bucket": "bucket-a", "Key": source_key}
    )

# 分布式执行复制任务
source_obj_keys.foreach(copy_single_object)

3. 优化与注意事项

• 批量处理：将对象键分组后批量调用复制接口，减少API请求次数，提升效率。
• 异常重试：给复制逻辑添加重试机制（如用tenacity库），处理网络波动或临时权限问题，同时记录失败的对象键以便后续补迁。
• 权限配置：确保Spark运行的账号/角色同时拥有源桶的读权限和目标桶的写权限。
• 大文件适配：针对超大文件，使用云存储的分段复制（multipart copy）接口，避免单次请求超时。

简化方案（直接用Spark读写API） #

如果无需自定义复制逻辑，可直接用Spark的二进制文件读写API，自动保留目录结构：

# 读取源桶所有文件（**表示递归所有子目录）
df = spark.read.format("binaryFile").load("s3a://bucket-a/**")
# 写入目标桶的migrated_data文件夹
df.write.format("binaryFile").mode("overwrite").save("s3a://bucket-b/migrated_data")

这种方式代码更简洁，Spark会自动处理分布式读写，适合无特殊需求的场景。

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