Synapse dedicated SQL pool写入方法及增量数据加载方案咨询

一、向Synapse Dedicated SQL Pool写入数据的常用实现方法 #

• COPY语句：微软官方首推的批量写入方式，支持直接从ADLS、Blob存储等位置加载CSV、Parquet等格式数据，无需额外权限配置，支持并行加载，性能比传统INSERT高3-10倍。使用示例：COPY INTO dbo.target_table FROM 'abfss://container@storage.dfs.core.windows.net/path/*' WITH (FILE_TYPE = 'PARQUET');
• PolyBase：适用于跨数据源联合查询后写入的场景，先创建外部数据源、外部文件格式、外部表，再通过CREATE TABLE AS SELECT (CTAS)语句将外部表数据写入SQL Pool内部表，适合TB级以上数据批量加载。
• Synapse Spark 连接器：通过Spark作业直接读写Dedicated SQL Pool，支持Python/Scala/Spark SQL语法，适合大数据量、需要提前做数据清洗转换的场景。
• BCP工具/SSIS：适合小批量、离线结构化数据同步场景，性能低于前三种方案，多用于传统SQL Server迁移场景。
• 常规INSERT语句：仅适合测试场景、小批量（小于1000行）数据写入，批量场景下性能极差不推荐使用。

二、Spark Notebook 增量加载Parquet/Delta Lake数据到Synapse SQL Warehouse落地方案 #

前置依赖 #

提前配置好Spark集群与Synapse Dedicated SQL Pool的网络连通性，配置拥有目标表读写权限的AAD服务主体或SQL认证账号。

增量识别逻辑 #

• 针对Delta Lake：直接利用Delta的时间旅行、版本号特性筛选增量数据，支持按更新时间、版本区间筛选，不需要额外维护增量标记。
• 针对Parquet：推荐使用分区字段（如dt、load_time）筛选新增文件，或通过文件最后修改时间标记增量，也可配合水印机制实现流式增量读取。

完整实现步骤（PySpark示例） #

1. 配置Synapse连接参数

# 配置连接信息
jdbc_url = "jdbc:sqlserver://<synapse-workspace>.sql.azuresynapse.net:1433;database=<sql-pool-name>;encrypt=true;trustServerCertificate=false;hostNameInCertificate=*.sql.azuresynapse.net;loginTimeout=30;"
db_properties = {
    "user": "<username>",
    "password": "<password>",
    "driver": "com.microsoft.sqlserver.jdbc.SQLServerDriver",
    "spark.synapse.linkedService": "<linked-service-name>" # 用工作区托管链路服务可省略账号密码配置
}
target_table = "dbo.target_table"

2. 读取增量数据

# Delta Lake 增量读取示例（读取最近24小时更新的数据）
from pyspark.sql.functions import current_timestamp, date_sub, col
delta_path = "abfss://container@storage.dfs.core.windows.net/delta_table_path/"
incremental_df = spark.read.format("delta").option("timestampAsOf", date_sub(current_timestamp(), 1)).load(delta_path)

# Parquet 增量读取示例（读取指定日期分区的数据）
parquet_path = "abfss://container@storage.dfs.core.windows.net/parquet_table_path/"
incremental_df = spark.read.parquet(parquet_path).filter(col("dt") >= "2024-01-01")

3. 写入Synapse（支持幂等写入避免重复）

# 方案1：临时表+MERGE实现Upsert（存在则更新，不存在则插入，适合有主键的表）
# 先将增量数据写入临时表
incremental_df.write \
    .mode("overwrite") \
    .synapsesql("dbo.temp_incremental_data")

# 执行MERGE语句合并数据
spark.sql("""
MERGE INTO dbo.target_table AS t
USING dbo.temp_incremental_data AS s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")

# 方案2：纯追加模式（适合日志类无重复数据的表）
incremental_df.write \
    .mode("append") \
    .option("truncate", "false") \
    .jdbc(url=jdbc_url, table=target_table, properties=db_properties)

落地注意事项 #

• 写入前合并小文件，避免过多文件导致加载性能下降
• 大批量写入前建议先关闭目标表的索引，写入完成后再重建索引
• 每次增量加载完成后对比源端和目标端的数据量、主键去重后条数，确保数据一致性
• 可以将增量标记（如最后同步的版本号、最大分区日期）持久化到配置表，实现自动化调度无需手动修改筛选条件

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