Spark读取存在数据类型不一致的Parquet文件解决方案问询

问题背景 #

存在多个列数相同的Parquet文件，其中部分列的数据类型不一致（例如cost列在file1中是int，file2中是int64），使用Spark直接读取或尝试mergeSchema均报错，直接转换写入也因读取阶段失败无法执行，需要找到无需全量重写源文件的类型兼容方法。

报错回顾 #

• 直接读取报错：类型不匹配，期望int但实际为INT64
• mergeSchema报错：无法合并int和bigint类型
• 转换写入报错：读取阶段已因类型不匹配失败

解决方案 #

1. 手动指定统一Schema读取（无需重写源文件） #

这是最直接有效的方法，提前定义包含兼容类型的Schema，强制Spark用该Schema解析所有文件。将类型不一致的列指定为范围更大的类型（如将int和int64统一为LongType），Spark会自动完成类型向上转换，不会丢失数据。

示例代码：

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

# 根据实际表结构定义统一Schema，将cost列设为LongType（兼容int和int64）
unified_schema = StructType([
    StructField("cost", LongType(), nullable=True),
    StructField("order_id", StringType(), nullable=True),
    StructField("user_id", IntegerType(), nullable=True),
    # 其他列按实际情况补充，保持列名、数量与源文件一致
])

# 使用自定义Schema读取所有Parquet文件
df = spark.read.schema(unified_schema).parquet("s3://mybucket/costs")

# 验证读取结果
df.printSchema()
df.show()

2. 配置Spark允许数值类型自动兼容（Spark 3.0+） #

对于Spark 3.0及以上版本，可以通过修改配置让Spark在读取时自动兼容数值类型的转换，避免手动定义Schema的繁琐。在SparkSession初始化时添加以下配置：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ParquetTypeCompatibility") \
    .config("spark.sql.parquet.mergeSchema", "true") \
    .config("spark.sql.legacy.parquet.numericTypeToStringConversion", "false") \
    .config("spark.sql.parquet.readLegacyFormat", "true") \
    .getOrCreate()

# 再尝试读取
df = spark.read.parquet("s3://mybucket/costs")

注意：该方法依赖Spark版本和具体配置，兼容性可能不如手动指定Schema稳定，建议优先使用方案1。

为什么之前的尝试失败？ #

• mergeSchema失败：Spark默认不允许合并int和bigint这类“不兼容”的数值类型，因为两者的存储范围不同，合并逻辑会直接抛出错误。
• 转换写入失败：读取阶段已经因类型不匹配报错，代码无法执行到转换和写入步骤，必须先解决读取问题。

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