Synapse Spark Notebook写入专用SQL池时Schema不匹配报错求助

错误根源 #

从报错信息来看，Synapse Spark连接器的Schema匹配规则是大小写敏感的列名、数据类型、可空性三者完全匹配，当前不匹配的核心原因包括：

1. 源DataFrame缺少目标表中的部分列（比如Activity_Start_Time、Activity_End_Time、Activity_Date等），导致连接器判定Schema不完整；
2. 目标表的Activity_Date为date类型，但连接器读取的目标Schema显示为TimestampType，存在类型映射不一致；
3. 可能存在列名大小写隐性差异（比如DataFrame列名是小写，目标表是大写，表面一致但实际存储有区别）。

具体解决步骤 #

1. 补全DataFrame的所有目标列 #

确保df_activity_staging包含目标表的全部14列，缺失的列可以通过withColumn添加默认值：

from pyspark.sql.functions import current_timestamp, lit

df_activity_staging = df_activity_staging \
    .withColumn("Activity_Start_Time", lit(None).cast("timestamp")) \
    .withColumn("Activity_End_Time", lit(None).cast("timestamp")) \
    .withColumn("Activity_Duration", lit(None).cast("double")) \
    .withColumn("Equipment_Operator_ID", lit(None).cast("long")) \
    .withColumn("Equipment_Travel_Point_DateTime", lit(None).cast("timestamp")) \
    .withColumn("Created_Date", current_timestamp()) \
    .withColumn("Last_Modified_Date", current_timestamp()) \
    .withColumn("Activity_Date", lit(None).cast("date"))

2. 严格对齐数据类型与可空性 #

• 确保DataFrame列的可空性和目标表一致：比如Equipment_Cycle_Number是NOT NULL，需保证DataFrame中该列无空值，可通过df.na.drop(subset=["Equipment_Cycle_Number"])处理；
• 修正Activity_Date类型：目标表为date类型，DataFrame对应列需转换为DateType：

df_activity_staging = df_activity_staging.withColumn("Activity_Date", df_activity_staging["Activity_Date"].cast("date"))

3. 显式指定列顺序 #

写入前按目标表的列顺序重新排列DataFrame，避免列顺序不一致导致匹配失败：

target_columns = [
    "Equipment_Cycle_Number", "Activity_Start_Time", "Activity_End_Time",
    "Activity_Duration", "Equipment_ID", "Equipment_Operator_ID",
    "Equipment_Cycle_Activity_ID", "Equipment_Travel_Point_DateTime",
    "Equipment_Location_UTM_X", "Equipment_Location_UTM_Y", "Equipment_Location_UTM_Z",
    "Created_Date", "Last_Modified_Date", "Activity_Date"
]

df_activity_staging = df_activity_staging.select(target_columns)

4. 验证列名大小写一致性 #

检查DataFrame列名与目标表是否完全一致，可通过以下命令查看列名：

print(df_activity_staging.columns)

若存在大小写差异，重命名列：

df_activity_staging = df_activity_staging.withColumnRenamed("equipment_cycle_number", "Equipment_Cycle_Number")

5. 更新Synapse Spark连接器版本 #

旧版本连接器可能存在类型映射bug（如将date识别为timestamp），确保使用最新版本（>=1.2.0），可在Spark配置中指定：

spark.conf.set("spark.jars.packages", "com.microsoft.azure:spark-mssql-connector_2.12:1.2.0")

6. 替代高效写入方案（若以上无效） #

如果连接器问题仍无法解决，可使用Synapse的COPY INTO语句实现高效写入：

1. 将DataFrame写入临时ABFS路径：

temp_path = "abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/temp_write"
df_activity_staging.write.mode("overwrite").parquet(temp_path)

2. 调用Synapse的COPY INTO语句：

COPY INTO [schema].[Table]
FROM 'abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/temp_write'
WITH (
    FILE_TYPE = 'PARQUET',
    CREDENTIAL = (IDENTITY = 'Storage Account Key', SECRET = '<your_storage_key>'),
    OVERWRITE = ON
)

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