从Databricks向Azure SQL数据库执行Upsert操作的问题排查

架构老林

2026-6-30

解决方案

错误原因分析

spark.write.jdbc的table参数仅支持表名或返回结果集的SELECT查询语句，直接传入MERGE这类DML语句会触发语法错误，这就是你看到Incorrect syntax near '('的核心原因。
Spark端创建的临时视图v_new_entries仅在Spark会话内有效，Azure SQL数据库无法访问这个视图，所以MERGE语句的USING子句会找不到数据源。

推荐方案：先写临时表再执行MERGE

最稳妥高效的方式是先将DataFrame写入Azure SQL的会话级临时表，再通过JDBC执行MERGE语句完成Upsert，具体步骤如下：

将待处理DataFrame写入Azure SQL的会话级临时表（以#开头，会话结束自动销毁）
执行MERGE语句，将临时表数据合并到目标表
（可选）手动清理临时表

修改后的代码实现

def process_dataframe_upsert(url, dbtable, dataframe):
    # 定义会话级临时表名
    temp_table = "#tmp_upsert_data"
    
    # 步骤1：将DataFrame写入临时表
    dataframe.write.jdbc(
        url=url,
        table=temp_table,
        mode="overwrite",
        properties=connection_properties
    )
    
    # 步骤2：构造正确的MERGE语句（修复原INSERT字段不匹配问题）
    upsert_sql = f"""
        MERGE INTO {dbtable} AS target
        USING {temp_table} AS source
        ON target.Col1 = source.Col1 AND target.Col2 = source.Col2 
        WHEN MATCHED THEN
            UPDATE SET target.Col3 = target.Col3 + 1  
        WHEN NOT MATCHED THEN
            INSERT (Col1, Col2, Date, WorkShift, ExitStatus, Col3) 
            VALUES (source.Col1, source.Col2, source.Date, source.WorkShift, source.ExitStatus, source.Col3)
    """
    
    # 步骤3：通过JDBC执行MERGE语句
    spark = SparkSession.getActiveSession()
    conn = spark._jvm.java.sql.DriverManager.getConnection(
        url,
        connection_properties["user"],
        connection_properties["password"]
    )
    conn.createStatement().execute(upsert_sql)
    conn.close()
    
    # 可选：手动删除临时表（会话结束会自动销毁，可不执行）
    drop_temp_sql = f"DROP TABLE IF EXISTS {temp_table}"
    conn = spark._jvm.java.sql.DriverManager.getConnection(
        url,
        connection_properties["user"],
        connection_properties["password"]
    )
    conn.createStatement().execute(drop_temp_sql)
    conn.close()

关键说明

使用#前缀的会话级临时表，避免跨会话冲突，且会话结束后自动清理，无需持久化存储。
直接在数据库端执行MERGE逻辑，数据处理效率远高于Spark端计算后再写入。
修复了你原MERGE语句中INSERT字段与VALUES不匹配的问题（原代码INSERT列有6个，但VALUES仅传入3个，会触发语法错误）。

大数据量场景优化方案：分区级批量处理

如果数据量较大，可通过foreachPartition在每个分区内建立独立JDBC连接，批量插入数据后执行MERGE，减少单连接的传输压力：

def upsert_partition(partition_data, url, dbtable, conn_props):
    import pyodbc
    # 解析JDBC URL获取服务器和数据库信息
    server = url.split('//')[1].split(':')[0]
    db_name = url.split('database=')[1]
    # 建立ODBC连接
    conn = pyodbc.connect(
        f"DRIVER={{{conn_props['driver']}}};SERVER={server};DATABASE={db_name};UID={conn_props['user']};PWD={conn_props['password']}"
    )
    cursor = conn.cursor()
    
    # 创建分区临时表
    cursor.execute("""
        CREATE TABLE #tmp_part (
            Col1 INT,
            Col2 INT,
            Date DATE,
            WorkShift VARCHAR(50),
            ExitStatus VARCHAR(50),
            Col3 INT
        )
    """)
    
    # 批量插入分区数据
    insert_sql = """
        INSERT INTO #tmp_part (Col1, Col2, Date, WorkShift, ExitStatus, Col3)
        VALUES (?, ?, ?, ?, ?, ?)
    """
    cursor.executemany(insert_sql, partition_data)
    
    # 执行MERGE
    merge_sql = f"""
        MERGE INTO {dbtable} AS target
        USING #tmp_part AS source
        ON target.Col1 = source.Col1 AND target.Col2 = source.Col2
        WHEN MATCHED THEN
            UPDATE SET target.Col3 = target.Col3 +1
        WHEN NOT MATCHED THEN
            INSERT (Col1, Col2, Date, WorkShift, ExitStatus, Col3)
            VALUES (source.Col1, source.Col2, source.Date, source.WorkShift, source.ExitStatus, source.Col3)
    """
    cursor.execute(merge_sql)
    
    conn.commit()
    cursor.close()
    conn.close()

# 调用方式
test_df.rdd.foreachPartition(lambda partition: upsert_partition(partition, jdbc_url, table_name, connection_properties))

注意事项

需要在Spark集群所有节点上安装pyodbc和SQL Server ODBC驱动。
分区级处理适合TB级数据场景，能有效降低单连接的数据传输压力。

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