合并Parquet文件失败：Schema中字符串与Double类型冲突求助

合并多个Parquet文件Schema时抛出错误：org.apache.spark.SparkException: Failed merging schema.Failed to merge incompatible data types string and double，尝试多种方法未解决，需要可行方案。

用户相关代码：

df = spark.read.format("parquet").load(result.db_path)
old_columns = df.columns
for col in old_columns:
    df = df.withColumnRenamed(col,col.lower())
df = df.withColumn("tenant", lit(payload.tenant))\
       .withColumn("filename", input_file_name())
write_format = 'delta'
save_path = f'dbfs:_________{endpoint.lower()}/'
db = f'--------'
name = f'{endpoint.lower()}_raas'
table_name = f'{db}.{name}'

if not spark._jsparkSession.catalog().tableExists(db,name):
    # Write the data to its target.
    df.write \
      .format(write_format) \
      .save(save_path)
    # Create the table.
    spark.sql("CREATE TABLE " + table_name + " USING DELTA LOCATION '" + save_path + "'")
else:
    df.write.format(write_format).mode("overwrite").save(save_path)

可行解决方案 #

1. 读取时指定统一Schema #

提前定义目标Schema，强制所有文件按该Schema读取，跳过自动推断带来的类型冲突：

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

# 根据业务需求定义统一Schema，将冲突列设为同一类型
target_schema = StructType([
    StructField("col1", StringType(), True),
    StructField("conflict_col", StringType(), True),  # 示例：统一为字符串类型，可根据实际调整
    # 其他列按实际情况补充定义
])

# 读取时指定schema参数
df = spark.read.format("parquet").schema(target_schema).load(result.db_path)

选择类型时优先考虑业务兼容性，比如如果double值可以转为字符串存储就选StringType，反之则确保字符串能安全转为double。

2. 手动遍历文件处理冲突列 #

关闭自动合并，逐个读取文件后统一冲突列类型，再合并数据：

from pyspark.sql.types import StringType

# 获取所有Parquet文件路径
file_paths = [path for path in spark.sparkContext.wholeTextFiles(result.db_path).keys().collect()]

processed_dfs = []
for path in file_paths:
    single_df = spark.read.format("parquet").load(path)
    # 处理冲突列，示例：将double转成string
    if "conflict_col" in single_df.columns:
        if single_df.schema["conflict_col"].dataType.typeName() == "double":
            single_df = single_df.withColumn("conflict_col", single_df["conflict_col"].cast(StringType()))
        # 若需将string转double，需处理转换失败场景：
        # single_df = single_df.withColumn("conflict_col", single_df["conflict_col"].cast(DoubleType()).otherwise(None))
    processed_dfs.append(single_df)

# 合并所有处理后的DataFrame
final_df = processed_dfs[0]
for df in processed_dfs[1:]:
    final_df = final_df.unionByName(df, allowMissingColumns=True)

# 后续执行列重命名、添加列等原有逻辑
old_columns = final_df.columns
for col in old_columns:
    final_df = final_df.withColumnRenamed(col, col.lower())
final_df = final_df.withColumn("tenant", lit(payload.tenant))\
                   .withColumn("filename", input_file_name())

3. 结合Delta Lake的Schema演进 #

针对你写入Delta表的场景，先统一DataFrame中冲突列的类型，再开启Schema演进写入：

# 先完成冲突列的类型统一处理（参考方法1或2）
# 写入时开启mergeSchema选项，允许兼容的Schema变更
df.write.format(write_format).mode("append").option("mergeSchema", "true").save(save_path)

如果是覆盖写入，需确保当前DataFrame的Schema与目标Delta表完全一致，或先通过ALTER TABLE调整表Schema后再写入。

4. 处理类型转换异常 #

若需将字符串转为数值型，需提前清洗数据并处理转换失败的情况：

from pyspark.sql.functions import when, col, regexp_replace

# 清洗非数字字符，转换失败则设为null
df = df.withColumn("conflict_col", 
    when(regexp_replace(col("conflict_col"), "[^0-9.]", "").cast(DoubleType()).isNotNull(),
         regexp_replace(col("conflict_col"), "[^0-9.]", "").cast(DoubleType()))
    .otherwise(None)
)

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