Pyspark读取多个parquet文件时列数据类型不一致该如何处理？

不需要手动对所有存在类型差异的列逐一转换，可通过以下两种方案解决Parquet文件合并时的类型冲突问题：

方案1：预定义统一Schema读取（推荐，效率最高） #

提前声明所有字段的目标数据类型，读取Parquet文件时直接指定该Schema，Spark会自动将不同文件的同名字段转换为你指定的类型，无需后续额外处理。
示例代码如下：

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

# 自定义目标Schema，按你的业务需求调整字段和类型，此处将geo列统一设为StringType
target_schema = StructType([
    StructField("user_id", IntegerType(), nullable=True),
    StructField("geo", StringType(), nullable=True),
    StructField("pay_amount", DoubleType(), nullable=True),
    # 其余字段按需补充
])

# 读取所有Parquet文件时直接指定Schema
df = spark.read.schema(target_schema).parquet("/path/to/your/parquet_dir/*.parquet")

方案2：批量读取后对齐类型再合并 #

适合仅1-2个字段存在类型差异的场景，逐个读取文件后统一转换差异字段的类型，再按字段名合并所有DataFrame。
示例代码如下：

from pyspark.sql.functions import col

# 替换为你的文件路径列表
file_paths = [
    "/path/file1.parquet",
    "/path/file2.parquet",
    "/path/file3.parquet"
]
df_list = []

for fp in file_paths:
    temp_df = spark.read.parquet(fp)
    # 仅对存在类型差异的geo列做统一转换，其余列保持不变
    temp_df = temp_df.withColumn("geo", col("geo").cast("string"))
    df_list.append(temp_df)

# 按字段名合并所有DataFrame
final_df = df_list[0]
for df in df_list[1:]:
    final_df = final_df.unionByName(df)

注意要点 #

• 不要直接使用spark.read.option("mergeSchema", "true")读取文件：该参数仅用于补全不同文件间缺失的字段，遇到同名字段类型冲突时仍然会抛出错误，无法解决你的问题
• 类型转换时注意避免数据损失：如果geo列的String类型值存在非数字内容，不要统一转为Double类型，否则会产生空值丢失数据，优先统一转为String类型更稳妥

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