Synapse Spark中Delta Lake对应Databricks的schema演化及写入优化配置查询

1. MERGE操作自动Schema合并的替代方案 #

spark.databricks.delta.schema.autoMerge.enabled是Databricks闭源专属配置，Synapse Spark无直接对应开关，可通过以下方式实现同等效果：

• 若使用SQL语句执行MERGE：操作前先手动对齐源表与目标表的字段差异，新增字段提前通过ALTER TABLE语句添加到目标Delta表中
• 若使用DataFrame API执行MERGE：在操作中显式添加mergeSchema参数开启schema合并，示例代码如下：

from delta.tables import DeltaTable

target_delta = DeltaTable.forPath(spark, "/delta/table/path")

target_delta.alias("t") \
  .merge(
    source_df.alias("s"),
    "t.id = s.id"
  ) \
  .whenMatchedUpdateAll() \
  .whenNotMatchedInsertAll() \
  .option("mergeSchema", "true") \
  .execute()

注：如果需要全局临时开启schema合并校验，可添加配置spark.conf.set("spark.sql.parquet.mergeSchema", "true") #

2. Merge操作分区文件控制与写入性能优化方案 #

Databricks的autoOptimize.optimizeWrite和autoCompact同样是闭源专属功能，Synapse Spark可通过以下开源Delta原生能力实现同等优化效果：

• 开启Spark自适应查询(AQE)自动合并小分区，Synapse Spark 3.1+默认已开启，未开启可手动配置：
  spark.conf.set("spark.sql.adaptive.enabled", "true")
  同时根据集群算力调整shuffle分区数，避免生成过多小文件：
  spark.conf.set("spark.sql.shuffle.partitions", 200) 数值可根据集群核心数调整，一般为总核心数的2~3倍
• 预设置Delta表目标文件大小，写入时自动控制单文件尺寸，避免小文件生成：
  执行SQL：ALTER TABLE <你的Delta表名> SET TBLPROPERTIES ('delta.targetFileSize' = '134217728') 示例为设置单文件目标大小128MB
• Merge操作完成后手动执行小文件合并，可搭配ZORDER提升查询性能：
  SQL方式：OPTIMIZE <你的Delta表名> ZORDER BY (<常用关联/过滤字段>)
  Python API方式：

  delta_table = DeltaTable.forName(spark, "<你的Delta表名>")
  delta_table.optimize().zorderBy("<常用关联/过滤字段>").executeCompaction()
  

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