Amazon S3百万级对象复制与文件夹重命名的Java实现方案咨询

1. S3 DistCP（优先推荐） #

• 是AWS官方推出的大规模S3数据批量迁移工具，底层基于并行计算框架实现，原生支持TB级数据、百万级对象的批量复制，自动处理重试、断点续传、API限流规避，效率远高于逐对象同步调用
• 支持自定义键名转换规则，可通过Java实现KeyTransformer接口完成重命名逻辑，直接将源路径中{数字}-{名称}/前缀替换为{数字}/，完全匹配你的需求
• 无需依赖EMR集群，可直接在Java项目中引入依赖后在本地/EC2/ECS环境运行，150万对象5TB数据通常可在2-4小时内完成复制，无需手动管理任务状态和API调用

核心重命名逻辑示例：

import com.amazonaws.services.s3distcp.KeyTransformer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class CustomKeyTransformer implements KeyTransformer {
    private static final Pattern PREFIX_PATTERN = Pattern.compile("^(\\d+)-[^/]+/(.*)$");
    @Override
    public String transform(String sourceKey) {
        Matcher matcher = PREFIX_PATTERN.matcher(sourceKey);
        if (matcher.matches()) {
            return matcher.group(1) + "/" + matcher.group(2);
        }
        // 不匹配规则的对象原样保留
        return sourceKey;
    }
}

2. AWS SDK for Java 2.x 异步批量客户端 #

如果不想引入额外工具，可直接使用官方SDK优化异步调用实现：

• 采用S3AsyncClient配合线程池批量提交复制请求，内置连接池和重试策略，可自动规避S3 API速率限制
• 配合ListObjectsV2分页查询，每次拉取1000个对象，控制并发数在100-200区间，实测150万对象可在8小时内完成复制

核心实现示例：

import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.*;
import software.amazon.awssdk.core.retry.RetryPolicy;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class S3BatchCopy {
    private static final Pattern PREFIX_PATTERN = Pattern.compile("^(\\d+)-[^/]+/(.*)$");
    private static final int CONCURRENCY = 150;
    private static final String BUCKET = "your-bucket-name";
    private static final String SOURCE_PREFIX = "your-source-dir/";

    public void runCopy() {
        S3AsyncClient s3Async = S3AsyncClient.builder()
                .overrideConfiguration(c -> c.retryPolicy(RetryPolicy.builder().numRetries(3).build()))
                .build();
        ExecutorService executor = Executors.newFixedThreadPool(CONCURRENCY);

        ListObjectsV2Request listReq = ListObjectsV2Request.builder()
                .bucket(BUCKET)
                .prefix(SOURCE_PREFIX)
                .build();
        // 分页拉取所有对象
        s3Async.listObjectsV2Paginator(listReq).stream()
                .flatMap(resp -> resp.contents().stream())
                .forEach(s3Object -> {
                    String sourceKey = s3Object.key();
                    Matcher matcher = PREFIX_PATTERN.matcher(sourceKey);
                    if (!matcher.matches()) return;
                    String destKey = matcher.group(1) + "/" + matcher.group(2);
                    // 异步提交复制请求
                    CompletableFuture<CopyObjectResponse> future = s3Async.copyObject(CopyObjectRequest.builder()
                            .sourceBucket(BUCKET)
                            .sourceKey(sourceKey)
                            .destinationBucket(BUCKET)
                            .destinationKey(destKey)
                            .build());
                    future.whenComplete((resp, err) -> {
                        if (err != null) {
                            // 记录错误对象后续统一重试即可
                            System.err.printf("复制失败: %s%n", sourceKey);
                        }
                    });
                });
    }
}

3. S3 Batch Operations + Lambda 托管方案 #

如果不想自己维护运行环境，可选择全托管方案：

• S3 Batch Operations本身支持最大10亿条记录的单CSV清单，无需拆分，直接提交全量清单即可
• 批量任务操作类型选择「调用Lambda函数」，在Lambda中实现重命名+复制逻辑，Lambda并发由AWS自动托管，仅需监控任务完成状态即可
• 整体成本极低，150万次Lambda调用成本不足1元，无额外运维开销

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