Spark读取分区数据时不同分区schema存在差异如何补全缺失列

问题原因 #

Spark默认读取分区存储的文件（如Parquet、ORC格式）时，仅会采样少量文件推断全局schema。如果采样到的首批文件来自无e列的EUR分区，最终推断出的schema就不会包含e列，US分区的e列数据会被直接丢弃。

方案1：开启schema自动合并（最便捷） #

开启Spark的分区schema合并配置，读取时会拉取所有分区的schema做自动合并，缺失的列会默认填充null值。
PySpark调用示例：

# 读数据时指定mergeSchema参数
df = spark.read.option("mergeSchema", "true").parquet("你的分区根目录路径")

Spark SQL 调用示例：

-- 先开启合并参数
SET spark.sql.parquet.mergeSchema=true;
-- 再读取数据
SELECT * FROM parquet.`你的分区根目录路径`;

提示：该方案需要遍历所有分区的文件获取schema，分区量级非常大时会有额外的性能开销。

方案2：手动指定完整schema（性能最优） #

如果业务schema固定，提前定义包含e列的完整schema，读取时直接传入即可，Spark会自动为缺失列填充null，不需要额外的schema采样开销。
PySpark调用示例：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 按实际字段类型调整，注意e列的nullable属性要设为True
full_schema = StructType([
    StructField("a", StringType(), nullable=True),
    StructField("b", StringType(), nullable=True),
    StructField("c", IntegerType(), nullable=True),
    StructField("d", IntegerType(), nullable=True),
    StructField("e", StringType(), nullable=True)
])

df = spark.read.schema(full_schema).parquet("你的分区根目录路径")

方案3：读取后手动补列（适合临时处理场景） #

如果已经完成数据读取，也可以手动判断列是否存在，不存在时补对应类型的null值：

from pyspark.sql.functions import lit

df = spark.read.parquet("你的分区根目录路径")
if "e" not in df.columns:
    # 按e列的实际类型调整cast的目标类型
    df = df.withColumn("e", lit(None).cast(StringType()))

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