使用Spark SQL写入Delta表时遭遇Schema不匹配异常的解决咨询

嗨，我之前也碰到过一模一样的问题！核心原因是你创建目标Delta表的时候没有指定明确的Schema，系统默认生成了一个空Schema的表，和临时表的Schema完全不匹配，插入自然就失败了。下面给你几个靠谱的解决办法，按优先级排序：

方法1：从Spark DataFrame提取Schema，创建表时明确指定 #

既然你已经把Pandas DataFrame转成了Spark DataFrame（df_spark），可以直接从它身上提取Schema的DDL格式字符串，然后在CREATE TABLE语句里指定，这样目标表的Schema就和临时表完全一致了：

from pyspark.sql import SparkSession

DB = database_name
TMP_TBL = temporary_table
TBL = table_name

sesh = SparkSession.builder.getOrCreate()
df_spark = sesh.createDataFrame(df)
df_spark.createOrReplaceTempView(TMP_TBL)

# 关键步骤：从Spark DataFrame提取DDL格式的Schema
schema_ddl = df_spark.schema.simpleString()

create_db_query = f"""
CREATE DATABASE IF NOT EXISTS {DB}
COMMENT "This is a database"
LOCATION "/tmp/{DB}"
"""

# 修改CREATE TABLE语句，新增SCHEMA指定
create_table_query = f"""
CREATE TABLE IF NOT EXISTS {DB}.{TBL}
USING DELTA
SCHEMA {schema_ddl}
TBLPROPERTIES (delta.autoOptimize.optimizeWrite = true, delta.autoOptimize.autoCompact = true)
COMMENT "This is a table"
LOCATION "/tmp/{DB}/{TBL}";
"""

insert_query = f"""
INSERT INTO TABLE {DB}.{TBL} select * from {TMP_TBL}
"""

sesh.sql(create_db_query)
sesh.sql(create_table_query)
sesh.sql(insert_query)

方法2：直接用Spark DataFrame的Write API（更简洁高效） #

其实完全不需要绕临时表和Spark SQL的弯路，直接用Spark DataFrame的write接口就能一步完成表创建+数据写入，还能自动保证Schema一致：

from pyspark.sql import SparkSession

DB = database_name
TBL = table_name

sesh = SparkSession.builder.getOrCreate()
df_spark = sesh.createDataFrame(df)

# 先创建数据库（如果需要）
sesh.sql(f"""
CREATE DATABASE IF NOT EXISTS {DB}
COMMENT "This is a database"
LOCATION "/tmp/{DB}"
""")

# 直接写入Delta表，自动创建表并匹配Schema
df_spark.write \
    .format("delta") \
    .mode("append")  # 首次创建可以用"overwrite"，后续追加数据用"append"
    .option("delta.autoOptimize.optimizeWrite", "true") \
    .option("delta.autoOptimize.autoCompact", "true") \
    .saveAsTable(f"{DB}.{TBL}", path=f"/tmp/{DB}/{TBL}")

这种方式不仅代码更短，还能避免手动写SQL可能出现的拼写错误或者Schema遗漏问题。

方法3：允许Delta自动合并Schema（适合表已存在的场景） #

如果你的目标表已经创建好了，只是Schema和新数据有差异，可以在INSERT语句里加上MERGE SCHEMA参数，让Delta自动兼容新增字段或兼容的数据类型变更：

INSERT INTO {DB}.{TBL} MERGE SCHEMA select * from {TMP_TBL}

不过要注意：这个功能只能兼容新增字段、字段顺序变化或者数据类型向上兼容（比如int转long），如果是字段删除或者数据类型向下兼容（比如long转int）还是会报错，使用前要确认数据类型的兼容性。

额外排查小技巧 #

如果还是不确定哪里Schema不匹配，可以在代码里加两行打印：

# 打印临时表的Schema
print("临时表Schema:")
df_spark.printSchema()

# 打印目标表的Schema（创建后执行）
print("目标表Schema:")
sesh.sql(f"DESCRIBE {DB}.{TBL}").show(truncate=False)

对比两者的字段名、数据类型、是否允许为空等信息，就能快速定位差异点了。

备注：内容来源于stack exchange，提问作者andKaae