如何修改Spark临时目录并添加时间戳避免Parquet文件互删

我完全懂你这个痛点——当多个Spark作业共用同一个输出目录，还都是用追加模式写Parquet文件时，临时文件互相覆盖、误删的问题简直防不胜防。给临时目录加上唯一时间戳的思路非常靠谱，既能彻底隔离不同作业的临时文件，写完再统一移到最终目录，完美规避冲突。下面是适配Spark 2.4.4的具体实现方案，分步骤给你讲清楚：

1. 生成带唯一时间戳的临时目录路径 #

首先得给每个作业生成一个独有的临时目录名，用毫秒级时间戳就足够保证唯一性（极端高并发场景还能加个随机串或者作业ID强化）。这里给Scala和Python两种常用语言的示例：

Scala 实现 #

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

// 生成精确到毫秒的时间戳字符串，确保每个作业的临时目录唯一
val timeStamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS"))
val tempOutputPath = s"hdfs:/outputFile/0/tmp_$timeStamp"

Python 实现 #

from datetime import datetime

# 截取到毫秒的时间戳字符串，避免重复
time_stamp = datetime.now().strftime("%Y%m%d%H%M%S%f")[:-3]
temp_output_path = f"hdfs:/outputFile/0/tmp_{time_stamp}"

2. 将DataFrame写入专属临时目录 #

接下来就把你的DataFrame写入这个带时间戳的临时目录，用追加模式即可——因为每个作业的临时目录都是独有的，完全不用担心和其他作业的文件冲突：

Scala 实现 #

df.write
  .mode("append")
  .parquet(tempOutputPath)

Python 实现 #

df.write \
  .mode("append") \
  .parquet(temp_output_path)

3. 移动临时文件到最终输出目录并清理临时目录 #

写入完成后，需要把临时目录里的Parquet文件（跳过_SUCCESS这类标记文件，按需选择）移动到最终的hdfs:/outputFile/目录，最后删除临时目录释放空间。这里直接用Hadoop的FileSystem API操作，Spark本身已经依赖Hadoop客户端，不用额外引入依赖：

Scala 实现 #

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

val sc = SparkContext.getOrCreate()
val fs = FileSystem.get(sc.hadoopConfiguration)
val finalOutputPath = new Path("hdfs:/outputFile/")
val tempPath = new Path(tempOutputPath)

// 遍历临时目录下的所有文件
val fileStatuses = fs.listStatus(tempPath)
fileStatuses.foreach { status =>
  val srcPath = status.getPath
  // 只移动.parquet文件，可根据需求调整过滤规则
  if (srcPath.getName.endsWith(".parquet")) {
    val destPath = new Path(finalOutputPath, srcPath.getName)
    // 如果目标文件已存在，可选择覆盖或跳过，这里示例为覆盖
    if (fs.exists(destPath)) {
      fs.delete(destPath, false)
    }
    // HDFS的rename操作是原子性的，不用担心中间状态
    fs.rename(srcPath, destPath)
  }
}

// 最后删除整个临时目录
fs.delete(tempPath, true)

Python 实现 #

from pyspark import SparkContext
from org.apache.hadoop.fs import Path, FileSystem

sc = SparkContext.getOrCreate()
fs = FileSystem.get(sc._jsc.hadoopConfiguration())
final_output_path = Path("hdfs:/outputFile/")
temp_path = Path(temp_output_path)

# 遍历临时目录下的文件
file_statuses = fs.listStatus(temp_path)
for status in file_statuses:
    src_path = status.getPath()
    # 只处理.parquet文件，按需调整过滤条件
    if src_path.getName().endswith(".parquet"):
        dest_path = Path(final_output_path, src_path.getName())
        # 目标文件存在则先删除再移动
        if fs.exists(dest_path):
            fs.delete(dest_path, False)
        fs.rename(src_path, dest_path)

# 删除临时目录，释放HDFS空间
fs.delete(temp_path, True)

额外注意事项 #

• 唯一性强化：如果你的集群作业并发量极高，毫秒级时间戳仍有极小概率重复，可以在时间戳后拼接java.util.UUID.randomUUID().toString()或者作业的ID，进一步保证临时目录唯一。
• 异常处理：建议在代码中加入try-catch（Scala）或try-except（Python）块，比如写入临时目录失败时，自动清理已创建的临时目录，避免HDFS上残留垃圾文件。
• 原子性保障：HDFS的rename操作是原子性的，所以移动文件的过程不会出现文件损坏或部分移动的问题，放心使用。

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