如何阻止Spark在Schema不匹配时向Parquet文件追加数据？

这个问题确实挺常见的——Spark默认对Parquet的Schema演进比较宽容，哪怕待追加数据多列或少列，都会默默允许写入，最后读的时候再合并Schema。但如果我们要严格管控，避免这种“意外的Schema变更”，有两种靠谱的方式可以实现：

一、手动校验Schema后再执行追加 #

既然Parquet文件本身包含Schema信息，我们可以先读取目标路径的现有Schema，和待写入DataFrame的Schema做严格对比，不一致就直接抛出异常阻止写入。

实现代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.types import *
import random
import string

spark = SparkSession.builder.appName('learn').master('yarn').enableHiveSupport().getOrCreate()

# 定义Schema校验函数
def check_schema_consistency(spark, target_path, input_df):
    # 读取目标路径的现有Schema
    existing_schema = spark.read.parquet(target_path).schema
    # 严格对比两个Schema（包括字段名、类型、nullable属性、顺序）
    if existing_schema != input_df.schema:
        raise ValueError(
            f"Schema不匹配，终止追加操作！\n现有Schema: {existing_schema}\n待写入Schema: {input_df.schema}"
        )

# ---------------------- 首次写入数据 ----------------------
schema1 = StructType([
    StructField('id_inside', LongType(), nullable=False),
    StructField('name_inside', StringType(), nullable=False),
])
data1 = [[random.randint(0, 5), ''.join(random.choice(string.ascii_lowercase) for _ in range(10))] for _ in range(10)]
df1 = spark.createDataFrame(data1, schema=schema1)
df1.write.format('parquet').mode('overwrite').save('/tmp/df1_2')

# ---------------------- 尝试追加不同Schema的数据 ----------------------
schema2 = StructType([
    StructField('id_inside', LongType(), nullable=False),
    StructField('name_inside', StringType(), nullable=False),
    StructField('name_inside2', StringType(), nullable=False),
])
data2 = [[random.randint(0, 5), *[''.join(random.choice(string.ascii_lowercase) for _ in range(10))]*2 ] for _ in range(10)]
df2 = spark.createDataFrame(data2, schema=schema2)

# 先校验Schema，再执行追加
try:
    check_schema_consistency(spark, '/tmp/df1_2', df2)
    df2.write.format('parquet').mode('append').save('/tmp/df1_2')
    print("Schema一致，追加成功")
except ValueError as e:
    print(e)

执行这段代码时，因为df2多了一列，会直接抛出错误，阻止写入操作。

二、使用Hive表（或Spark SQL表）来约束Schema #

如果把数据写入Spark SQL/Hive表，而不是直接写入文件路径，Spark会自动校验待写入数据的Schema是否和表Schema一致——不一致的话会直接报错，拒绝追加。

实现代码示例 #

# 首次写入为Hive表（自动创建表并绑定Schema）
df1.write.format('parquet').mode('overwrite').saveAsTable('default.df1_table')

# 尝试用不同Schema追加数据
try:
    df2.write.format('parquet').mode('append').saveAsTable('default.df1_table')
    print("追加成功")
except Exception as e:
    print(f"Schema不匹配，追加被拦截: {str(e)}")

这种方式更省心，因为表的Schema是固定的，Spark会自动帮我们做校验，不需要手动写对比逻辑。

补充说明 #

Spark默认允许Parquet Schema演进的原因是：Parquet本身支持列式存储的Schema变更，且Spark默认开启了spark.sql.parquet.mergeSchema参数（默认值为true），读取时会自动合并所有分区文件的Schema。但如果我们只是想禁用读取时的Schema合并，并不会阻止写入时的Schema不匹配——所以必须通过上述两种方式来拦截写入操作。

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