Spark加载S3文件时如何跳过损坏文件并记录日志？

针对你遇到的S3中个别损坏GZIP文件导致整个Spark任务失败的问题，有两种可行方案：

方案一：逐个读取文件并捕获异常（可记录损坏文件） #

这种方法可以精准定位并记录每个损坏的文件，同时跳过它们继续处理正常文件。

修改后的代码示例：

from pyspark.sql import SparkSession, DataFrame
import pyspark.sql.functions as F
from py4j.protocol import Py4JJavaError

def read_data(spark: SparkSession) -> DataFrame:
    # 获取S3目标路径下的所有文件
    fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())
    path = spark._jvm.org.apache.hadoop.fs.Path("s3://my_bucket/some_folder/")
    file_statuses = fs.listStatus(path)
    all_files = [status.getPath().toString() for status in file_statuses if not status.isDirectory()]
    
    valid_dfs = []
    corrupted_files = []
    
    for file_path in all_files:
        try:
            # 读取单个文件
            df = spark.read.format("json").load(file_path)
            # 可选：添加源文件名字段，方便后续数据溯源
            df = df.withColumn("source_file", F.lit(file_path))
            valid_dfs.append(df)
        except Py4JJavaError as e:
            error_msg = str(e.java_exception)
            if "incorrect header check" in error_msg:
                corrupted_files.append(file_path)
                print(f"损坏文件已记录: {file_path}")
            else:
                # 非GZIP头错误的异常，重新抛出避免遗漏问题
                raise e
    
    # 输出损坏文件统计，也可写入日志文件或数据库
    if corrupted_files:
        print(f"共发现 {len(corrupted_files)} 个损坏文件: {corrupted_files}")
        # 可选：将损坏文件列表写入S3指定路径留存
        spark.createDataFrame([(f,) for f in corrupted_files], ["corrupted_file"])\
             .write.mode("overwrite").text("s3://my_bucket/corrupted_files_log/")
    
    # 合并所有正常读取的DataFrame
    if valid_dfs:
        return valid_dfs[0].unionAll(valid_dfs[1:])
    else:
        # 无有效文件时返回空DataFrame，可根据业务需求调整
        return spark.createDataFrame([], schema="")

方案二：启用Spark全局容错配置（简单高效） #

Spark 2.1及以上版本支持通过配置自动跳过损坏文件，无需修改读取逻辑，但无法主动记录损坏文件列表（需通过Spark日志排查）。

修改后的代码示例：

from pyspark.sql import SparkSession, DataFrame

def read_data(spark: SparkSession) -> DataFrame:
    # 开启忽略损坏文件的全局配置
    spark.conf.set("spark.sql.files.ignoreCorruptFiles", "true")
    # 可选：同时开启忽略缺失文件
    spark.conf.set("spark.sql.files.ignoreMissingFiles", "true")
    
    spark_reader = spark.read.format("json")
    return spark_reader.load("s3://my_bucket/some_folder/")

两种方案对比 #

• 方案一：优势是能精准记录损坏文件路径，适合需要追踪问题文件的场景；缺点是逐个读取文件，性能略低于批量读取。
• 方案二：优势是配置简单、性能和原批量读取一致；缺点是无法主动获取损坏文件列表，需依赖Spark运行日志定位问题。

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