Spark读取无_SUCCESS文件的文件夹时如何原生触发错误异常

Spark 没有提供读取数据源时自动校验_SUCCESS文件存在性的开箱即用配置，但可以直接调用 Spark 内置的 Hadoop 文件系统原生 API 实现需求，无需引入第三方依赖或从零实现文件校验逻辑：

• 核心实现逻辑：调用 Spark 上下文内置的 Hadoop FileSystem 接口，直接校验目标路径下的_SUCCESS文件是否存在，校验不通过直接抛出异常终止流程即可，S3 的 S3A 协议完全兼容该接口。

Scala 代码示例 #

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

// 替换为你的目标S3路径
val targetPath = new Path("s3a://你的存储桶/数据存储路径/")
val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
val successFilePath = new Path(targetPath, "_SUCCESS")

// 校验逻辑
if (!fs.exists(successFilePath)) {
  throw new RuntimeException(s"数据校验失败：目标路径 ${targetPath.toString} 不存在_SUCCESS文件，上游写入未完成或失败")
}

// 校验通过后再执行读取操作
val dataDF = spark.read.parquet(targetPath.toString)

PySpark 代码示例 #

from py4j.java_gateway import java_import

# 导入Hadoop文件系统相关类
java_import(spark._jvm, "org.apache.hadoop.fs.Path")
java_import(spark._jvm, "org.apache.hadoop.fs.FileSystem")

# 替换为你的目标S3路径
target_path = spark._jvm.Path("s3a://你的存储桶/数据存储路径/")
fs = spark._jvm.FileSystem.get(spark._jsc.hadoopConfiguration())
success_file_path = spark._jvm.Path(target_path, "_SUCCESS")

# 校验逻辑
if not fs.exists(success_file_path):
    raise RuntimeError(f"数据校验失败：目标路径 {target_path.toString()} 不存在_SUCCESS文件，上游写入未完成或失败")

# 校验通过后再执行读取操作
data_df = spark.read.parquet(target_path.toString())

额外说明：
Spark 默认没有集成_SUCCESS校验逻辑，是因为该文件仅为 Spark 写入任务成功的标记，非 Spark 生成的数据集通常不会携带该文件，因此 Spark 官方将该校验能力以通用文件系统 API 的形式开放，由用户根据自身场景按需调用。

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