S3写入Parquet数据时，如何减少消费者的访问停机时间？

以下是几种常用的技术手段，可最小化用户访问停机时间：

1. 临时目录写入 + 原子重命名 #

直接使用overwrite模式会先删除目标路径下的旧文件，再写入新文件，这30分钟的窗口内消费者会遇到文件缺失。改为先写入临时目录，待全部文件写完后再将临时目录重命名到目标路径，消费者只会看到完整的旧数据或新数据，不会出现中间状态。

Scala代码示例：

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.SparkSession

val spark = SparkSession.active
val targetPath = "s3://bucket/path/to/folder"
val tempPath = s"$targetPath.temp"

// 先将数据写入临时目录
df.write
  .partitionBy("year", "month", "day")
  .mode("overwrite")
  .parquet(tempPath)

// 获取Hadoop文件系统实例，执行目录重命名
val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
val tempDir = new Path(tempPath)
val targetDir = new Path(targetPath)

// 先清理旧的目标目录（若存在），再将临时目录移动到目标路径
if (fs.exists(targetDir)) {
  fs.delete(targetDir, true)
}
fs.rename(tempDir, targetDir)

注：S3本身没有原生的原子目录重命名，但此方式是先完成全量写入再执行目录替换，消费者不会看到不完整的文件集合。

2. 使用湖仓格式实现事务性写入 #

Delta Lake、Iceberg、Hudi这类湖仓格式支持ACID事务，写入过程中旧数据保持可访问，提交完成后原子切换到新数据，从根本上避免文件找不到的问题。

以Delta Lake为例，Scala代码示例：

import io.delta.tables._

df.write
  .partitionBy("year", "month", "day")
  .mode("overwrite")
  .format("delta")
  .save("s3://bucket/path/to/folder")

这类格式会维护数据的版本日志，消费者读取时只会看到已提交的完整数据版本，写入过程中的临时文件对消费者不可见。

3. 增量覆盖分区而非全量替换 #

如果业务逻辑允许，不要每次覆盖整个目录，只针对需要更新的分区进行写入。比如按year/month/day分区时，仅覆盖当前时间对应的分区，其他分区的文件始终可用，将影响范围缩小到单个分区。

Scala代码示例：

val targetPath = "s3://bucket/path/to/folder"
// 假设当前数据对应的分区参数
val currentYear = "2024"
val currentMonth = "05"
val currentDay = "20"

// 仅覆盖目标分区
df.write
  .mode("overwrite")
  .parquet(s"$targetPath/year=$currentYear/month=$currentMonth/day=$currentDay")

4. 开启S3版本控制 #

开启S3桶的版本控制后，被删除的旧文件不会立即消失，而是保留历史版本。消费者可以配置读取旧版本数据，或者在写入完成后切换到新版本，避免写入期间的文件缺失错误。

Spark读取时指定版本的示例：

spark.read
  .option("fs.s3a.version.id", "具体的旧版本ID")
  .parquet("s3://bucket/path/to/folder")

注：此方法需要消费者配合调整读取逻辑，更多是作为容错补充手段。

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