如何向Azure Dedicated SQL池的指定分区写入数据

Databricks 内置的 Synapse 连接器（com.databricks.spark.sqldw）不支持 Spark 原生的 partitionBy 写入参数，该参数仅适用于文件类数据源（如Delta、Parquet等）的分区目录生成，对Synapse这类数仓数据源无效。原生overwrite模式默认是整表删除重建，不符合增量刷新指定分区的需求，可通过以下方案实现：

方案：预删除目标分区+追加写入（最通用方案） #

实现逻辑 #

• 先从Delta源表过滤出需要刷新的指定countryid分区数据
• 写入前先删除Synapse目标表中对应countryid的存量数据
• 再将过滤好的分区数据追加到Synapse目标表，实现分区级别的刷新

代码示例 #

# 1. 定义需要刷新的分区值，可按需求调整
target_country_ids = [1, 5, 10]
# 2. 从Delta表读取指定分区的数据
delta_df = spark.read.format("delta") \
  .load("<你的Delta表路径>") \
  .filter(f"countryid in ({','.join([str(id) for id in target_country_ids])})")
# 3. 构造预执行的删数SQL
delete_sql = f"DELETE FROM <你的Synapse目标表名> WHERE countryid in ({','.join([str(id) for id in target_country_ids])})"
# 4. 写入Synapse，使用preActions参数执行预删数，mode改为append
delta_df.write \
  .format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://<你的连接串>") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("dbTable", "<你的Synapse目标表名>") \
  .option("tempDir", "wasbs://<容器名>@<存储账号名>.blob.core.windows.net/<临时目录>") \
  .option("maxStrLength", 4000) \
  .option("preActions", delete_sql) \
  .mode("append") \
  .save()

注意事项 #

• 需保证Delta侧筛选的分区范围和preActions里的删数范围完全一致，避免出现多删、漏删数据的问题
• 对数据一致性要求高的场景，可以先将过滤后的分区数据写入Synapse的临时表，校验数据无误后再执行目标表的分区删除和数据插入操作，避免中间写入失败导致数据丢失
• 若需要同时刷新多个分区，可直接扩展target_country_ids的取值列表即可，无需修改其他逻辑

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