如何通过Spark Scala提升S3桶的数据存储速度？

1. 用Spark原生分区写入替代循环过滤（核心优化） #

你当前的写法是对每个name单独触发一次Spark Job，每个Job都要经历调度、序列化、IO初始化等固定开销，哪怕只有10个name，累加的开销也会非常显著。

直接使用Spark的partitionBy功能，一次Job就能完成所有分区的写入，完全避免多次Job的额外开销：

modifiedFinalDf.write
  .mode(
    fileWriteMode match {
      case FILE_WRITE.APPEND => SaveMode.Append
      case FILE_WRITE.OVER_WRITE => SaveMode.Overwrite
      case FILE_WRITE.DEFAULT => SaveMode.Ignore
    }
  )
  .partitionBy("name") // Spark会自动创建name=<determined_name>的目录结构
  .parquet(s"$bucketname/$datePath") // 日期路径作为父目录，保持你的原有路径格式

Spark会自动按name字段拆分数据并写入对应目录，整个过程仅需一次Job调度，大幅缩短耗时。

2. 控制输出文件大小，避免小文件 #

S3对大量小文件的读写效率极低，Spark默认并行度可能生成过多小文件，拖慢写入速度。可以通过以下参数调整：

• spark.sql.shuffle.partitions：如果数据经过shuffle，调整该值（默认200，数据量小时可设为10-50）
• spark.sql.files.maxRecordsPerFile：设置每个文件的最大记录数，比如设为1000000，让文件大小维持在128MB-256MB（Parquet格式的最优大小区间）
• spark.sql.files.openCostInBytes：调整文件打开成本估算值，帮助Spark优化文件合并逻辑

也可以在写入前用repartition("name")或coalesce调整分区数，确保每个分区的数据量匹配最优文件大小。

3. 优化S3客户端配置 #

调整Spark的S3相关参数，提升上传效率：

// 启用并行多部分上传
spark.conf.set("spark.hadoop.fs.s3a.fast.upload", "true")
// 设置多部分上传块大小为128MB
spark.conf.set("spark.hadoop.fs.s3a.multipart.size", "134217728")
// 增加S3连接池最大连接数
spark.conf.set("spark.hadoop.fs.s3a.connection.maximum", "100")
// 提升上传线程数
spark.conf.set("spark.hadoop.fs.s3a.threads.max", "32")

这些参数可以减少S3连接的等待时间，提升并行上传能力。

4. 移除不必要的本地并行操作 #

你使用的finalEventList.par.foreach是Scala本地线程池的并行，并非Spark的分布式并行。本地并行会导致多个Spark Job同时提交，引发资源竞争，反而降低整体效率。如果一定要保留循环逻辑，直接串行执行即可，避免本地线程带来的额外开销。

5. 预过滤和数据精简 #

在写入前，确保modifiedFinalDf只包含需要的字段，减少不必要的IO量：

// 仅保留需要写入的字段，避免冗余数据传输
val trimmedDf = modifiedFinalDf.select("name", "subject", "your_other_fields")

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