PySpark中如何获取Schema不匹配的HDFS文件名并排除对应数据

要定位Schema不匹配的HDFS文件并排除这些文件的处理，核心是将单个文件的路径与它的Schema绑定校验，而不是一次性读取所有文件合并成DataFrame。以下是两种可行方案：

方案一：逐个读取文件并校验Schema（推荐） #

这种方法直接针对每个文件推断Schema，与目标Schema对比，能精准定位不匹配的文件，同时只保留符合要求的文件数据。

步骤与代码实现 #

1. 定义Schema校验函数，判断单个文件的Schema是否与目标Schema匹配
2. 遍历所有文件路径，批量校验并分类
3. 合并合法文件的DataFrame并写入表，记录不匹配的文件

# 目标Schema（你的配置中的my_schema）
schema = my_schema

file_paths = [
    "hdfs/2022-12-12/file1.json",
    "hdfs/2022-10-10/file2.json",
    "hdfs/2024-02-12/file1.json",
    "hdfs/2020-11-01/file20.json"
]

def validate_file_schema(file_path, target_schema):
    # 读取单个文件并推断其Schema
    df_single = spark.read.json(file_path)
    file_schema = df_single.schema
    
    # 严格校验：匹配字段名、数据类型、nullable属性及字段顺序
    if file_schema == target_schema:
        return (True, df_single)
    # 可选：宽松校验（忽略字段顺序和nullable）
    # target_fields = {f.name: f.dataType for f in target_schema.fields}
    # file_fields = {f.name: f.dataType for f in file_schema.fields}
    # if file_fields == target_fields:
    #     return (True, df_single)
    else:
        return (False, file_path)

valid_dfs = []
rejected_files = []

# 遍历所有文件进行校验
for path in file_paths:
    is_valid, result = validate_file_schema(path, schema)
    if is_valid:
        valid_dfs.append(result)
    else:
        rejected_files.append(result)

# 合并合法数据并写入表
if valid_dfs:
    # 初始化空DataFrame（与目标Schema一致）
    final_df = spark.createDataFrame(spark.sparkContext.emptyRDD(), schema)
    for df in valid_dfs:
        final_df = final_df.union(df)
    final_df.write.saveAsTable("your_target_table")

# 输出或保存不匹配的文件列表
print("Schema不匹配的文件：", rejected_files)
# 可选：将不匹配记录写入日志表
# spark.createDataFrame([(f,) for f in rejected_files], ["rejected_file_path"]).write.saveAsTable("rejected_files_log")

补充优化 #

如果需要处理空文件或不存在的文件，可以在校验函数中增加文件状态检查：

def validate_file_schema(file_path, target_schema):
    try:
        # 检查HDFS文件是否存在且非空
        fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())
        file_path_jvm = spark._jvm.org.apache.hadoop.fs.Path(file_path)
        if not fs.exists(file_path_jvm):
            return (False, f"{file_path} 文件不存在")
        if fs.getFileStatus(file_path_jvm).getLen() == 0:
            return (False, f"{file_path} 文件为空")
    except Exception as e:
        return (False, f"{file_path} 校验失败：{str(e)}")
    
    df_single = spark.read.json(file_path)
    file_schema = df_single.schema
    return (file_schema == target_schema, df_single if file_schema == target_schema else file_path)

方案二：保留文件名读取后按文件统计（适合大数量文件） #

如果文件数量极多，逐个读取效率较低，可以一次性读取所有文件并保留文件名，再通过_corrupt_record字段或Schema匹配度统计不匹配的文件。但这种方法更适合记录级不匹配的场景，若需判断整个文件Schema不匹配，需额外统计每个文件的异常记录占比。

代码示例 #

# 读取所有文件，保留文件名并启用PERMISSIVE模式（默认）
df_with_file = spark.read.option("mode", "PERMISSIVE") \
    .json(file_paths) \
    .withColumn("source_file", input_file_name())

# 检查每条记录是否存在损坏字段（_corrupt_record）
corrupt_records = df_with_file.filter(col("_corrupt_record").isNotNull())

# 获取所有包含损坏记录的文件
rejected_files = corrupt_records.select("source_file").distinct().rdd.flatMap(lambda x: x).collect()

# 过滤出合法记录并写入表
valid_df = df_with_file.filter(col("_corrupt_record").isNull()).drop("_corrupt_record")
valid_df.write.saveAsTable("your_target_table")

print("包含Schema不匹配记录的文件：", rejected_files)

这种方法的缺点是无法区分“部分记录Schema不匹配”和“整个文件Schema不匹配”，如果你的需求是排除任何包含不匹配记录的文件，可以使用此方法；若仅排除整个Schema完全不匹配的文件，方案一更精准。

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