Azure Blob中Spark生成的Part文件转独立CSV的实现问题

问题根源在于你使用的java.io.File类仅支持本地文件系统，无法识别Azure Blob的WASB分布式路径。改用Hadoop原生的FileSystem API即可解决该问题，以下是完整实现方案：

1. 生成并写入DataFrame（保留原核心逻辑） #

val dfOutput = spark.sql("""SELECT * FROM Query""")
// coalesce(1)将数据合并为单个分区，确保仅生成一个part文件（数据量较大时需谨慎使用，可替换为repartition）
dfOutput.coalesce(1)
        .write
        .option("header", "true")
        .mode("overwrite")
        .format("csv")
        .save(OutputFile) // OutputFile需为WASB路径，例如"wasb://mycontainer@myexamplestorage.blob.core.windows.net/file.csv"

2. 复制part文件并清理原文件夹 #

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}

// 直接复用Spark自带的Hadoop配置，已包含Azure Blob的访问配置
val hadoopConfig = spark.sparkContext.hadoopConfiguration
val fs = FileSystem.get(hadoopConfig)

val srcDirPath = new Path(OutputFile)
val destFilePath = new Path("wasb://mycontainer@myexamplestorage.blob.core.windows.net/resultfile.csv")

// 遍历原目录，筛选出目标part文件（排除校验文件和_SUCCESS标志）
val partFiles = fs.listStatus(srcDirPath)
  .filter(status => !status.isDirectory 
                    && status.getPath.getName.startsWith("part-") 
                    && !status.getPath.getName.endsWith(".crc"))

if (partFiles.nonEmpty) {
  val srcPartPath = partFiles(0).getPath
  // 先删除已存在的目标文件（避免冲突）
  if (fs.exists(destFilePath)) fs.delete(destFilePath, true)
  // 用rename移动文件（比copy更高效，无需重复传输数据）
  fs.rename(srcPartPath, destFilePath)
  // 删除原输出目录及其中的冗余文件
  fs.delete(srcDirPath, true)
} else {
  throw new RuntimeException("未找到目标part数据文件")
}

关键细节说明 #

• 复用Spark的Hadoop配置，无需手动添加Azure存储账户密钥等信息，避免配置遗漏。
• 使用FileSystem.listStatus枚举WASB路径下的文件，替代不支持分布式文件系统的FileUtil.listFiles。
• 优先使用rename操作而非copy，减少Blob存储中的数据传输开销。
• 过滤逻辑排除了.crc校验文件和_SUCCESS标志文件，确保只处理有效CSV数据文件。

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