如何使用 scala 和 aws-java-sdk 从 S3 存储桶中获取所有 S3ObjectSummary？答案

【问题标题】：How to get all S3ObjectSummary from an S3 bucket using scala and the aws-java-sdk?如何使用 scala 和 aws-java-sdk 从 S3 存储桶中获取所有 S3ObjectSummary？
【发布时间】：2022-05-06 19:08:02
【问题描述】：

我有一个 scala 项目，我尝试实施一项需要访问 Amazon S3 存储桶的服务。

我想得到一个桶中所有对象的列表，但是s3Client.listObjects的结果集被分页到了1000个项目。

必须获取多个 objectListings 才能获得所有结果。

我找到了一个example Java implementation，但它依赖于可变性（在 while 循环中覆盖 objectListing）：

AmazonS3 s3Client = AmazonS3Provider.getS3Client();
ListObjectsRequest req = new ListObjectsRequest().withBucketName(realBucket).withPrefix(!preprefix.equals("") ? preprefix + "/" + prefix : prefix);
ObjectListing objectListing = s3Client.listObjects(req);
List<S3ObjectSummary> summaries = objectListing.getObjectSummaries();

while (objectListing.isTruncated()) {
    objectListing = s3Client.listNextBatchOfObjects(objectListing);
    summaries.addAll(objectListing.getObjectSummaries());
}

虽然我可以将其翻译成 scala 很好，但我想使用更惯用的 scala 方式。

如何使用 scala 获取存储桶的所有页面？

【问题讨论】：

你可以试试Benji S3（由我工作的公司），它更符合 Scala 习惯（DSL，基于 akka 流）

标签： scala amazon-s3 aws-sdk

【解决方案1】：

我现在使用递归方法并在每次迭代期间填充结果对象。一旦到达最后一页，它将返回最终集合。

相关部分发生在getAllSummaries 方法中，我保留了其他实现细节，以便它可以帮助其他人更轻松地使其工作。（我的 AmazonS3Config 是一个包含我的 S3 凭证的基本案例类。）

import com.amazonaws.auth.{AWSStaticCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.regions.Regions
import com.amazonaws.services.s3.model.{ObjectListing, S3ObjectSummary}
import com.amazonaws.services.s3.{AmazonS3, AmazonS3ClientBuilder}

import scala.collection.JavaConverters._

object Starter extends App with Configurable {

  private lazy val client: AmazonS3 = createAmazonClient(this.config.s3)

  val objects = getAllObjects()

  def getAllObjects(): Seq[S3ObjectSummary] = {
    val bucket = "YOUR_BUCKET_NAME"
    val prefix = ""

    val objectListing: ObjectListing = client.listObjects(bucket, prefix)

    getAllSummaries(objectListing)
  }

  private def getAllSummaries(list: ObjectListing,
                              res: Seq[S3ObjectSummary] = Seq.empty[S3ObjectSummary]): Seq[S3ObjectSummary] =
    list.isTruncated match {
      case false => {
        res ++ list.getObjectSummaries.asScala
      }
      case true =>
        val newList = this.client.listNextBatchOfObjects(list)
        getAllSummaries(newList, res ++ newList.getObjectSummaries.asScala)

    }

  private def createAmazonClient(config: AmazonS3Config): AmazonS3 = {
    val region = Regions.valueOf(config.region)
    val awsCredentials = new BasicAWSCredentials(config.accessKey, config.secretKey)

    AmazonS3ClientBuilder
      .standard()
      .withCredentials(new AWSStaticCredentialsProvider(awsCredentials))
      .withRegion(region)
      .build()
  }
}

【讨论】：

【解决方案2】：

下面有错误！第一批数据已丢失。您必须更改代码：getAllSummaries(newList, res ++ newList.getObjectSummaries.asScala) 到 getAllSummaries(newList, res ++ list.getObjectSummaries.asScala) getAllSummaries() 函数的正确代码是：

 private def getAllSummaries(list: ObjectListing,
                              res: Seq[S3ObjectSummary] = Seq.empty[S3ObjectSummary]): Seq[S3ObjectSummary] =
    list.isTruncated match {
      case false => {
        res ++ list.getObjectSummaries.asScala
      }
      case true =>
        val newList = this.client.listNextBatchOfObjects(list)
        getAllSummaries(newList, res ++ list.getObjectSummaries.asScala)

    }

但 Scala 更好的风格是：

@tailrec
def getAllSummaries(list: ObjectListing,
                    acc: Seq[S3ObjectSummary]): Seq[S3ObjectSummary] =
  if (list.isTruncated) {
    val newList = client.listNextBatchOfObjects(list)
    getAllSummaries(newList, acc ++ list.getObjectSummaries.asScala)
  } else {
    acc ++ list.getObjectSummaries.asScala
  }

并用作：

val objectListing: ObjectListing = client.listObjects(bucket, prefix)
getAllSummaries(objectListing, Seq.empty[S3ObjectSummary])

【讨论】：