如何使用Scala/Java实现同S3存储桶内跨目录文件移动？

你遇到的报错核心是两个问题：

1. 方法选型错误：fs.moveFromLocalFile()/fs.copyFromLocalFile()是Hadoop FileSystem专门为本地文件系统（file://协议）到分布式存储的拷贝场景设计的方法，强制要求源路径必须是本地路径，完全不支持源和目标都是S3路径的操作，这就是抛出Wrong FS: s3a:// expected file:///的直接原因。
2. 路径笔误：你定义的目标路径s3a:/path-to-destination-directory/格式错误，S3A协议标准路径格式为s3a://<桶名>/<前缀路径>，协议头后必须跟双斜杠。

S3本身没有原生目录概念，所谓目录移动本质是将源前缀下的所有对象复制到目标前缀，再删除源对象。Hadoop 2.8及以上版本自带的S3A客户端已经封装了同桶移动的优化逻辑，底层直接调用S3服务端COPY接口，不需要把文件下载到本地重传，直接用通用的rename()方法即可实现需求。

import org.apache.hadoop.fs.Path
import org.apache.spark.sql.SparkSession

def copyFromInputFilesToArchive(spark: SparkSession) : Unit = {
  // 修正路径格式，替换为实际的桶名和路径前缀
  val sourcePath = new Path("s3a://your-target-bucket/source-directory/")
  val destPath = new Path("s3a://your-target-bucket/destination-directory/")
  val hadoopConf = spark.sparkContext.hadoopConfiguration

  // 可选配置：调整同桶移动的并行线程数，根据集群资源调整
  hadoopConf.setInt("fs.s3a.rename.threadpool.size", 10)
  val fs = sourcePath.getFileSystem(hadoopConf)

  // 前置校验：源路径必须存在
  if (!fs.exists(sourcePath)) {
    throw new RuntimeException(s"Source S3 path $sourcePath not exist")
  }

  // 调用通用rename方法，S3A同桶场景自动走服务端COPY+源文件删除逻辑
  val moveSuccess = fs.rename(sourcePath, destPath)
  if (!moveSuccess) {
    throw new RuntimeException(s"Failed to move files from $sourcePath to $destPath")
  }
}

• 如果你的Hadoop版本低于2.8，S3A客户端未内置rename优化，需要先递归列出源路径下所有文件，逐个执行拷贝后删除源文件；不建议直接升级客户端的场景下可以用分布式方式并行处理，避免Driver单点性能瓶颈。
• 移动操作前确认目标路径下无重名文件，可通过fs.s3a.overwrite.allowed配置项控制是否允许覆盖文件，避免误删数据。
• 单目录下文件量超过10万时，不建议在Driver端单线程执行移动操作，建议通过Spark分布式遍历所有对象键，并行执行COPY和DELETE操作，避免请求超时。

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