如何从Databricks向SQL Server存储过程传递DataFrame列参数？

我在SQL Server中有一个用于执行UPSERT操作的存储过程，该存储过程的参数对应DataFrame的列。我的upsert_df DataFrame包含ColA、ColB、ColC三列，需将这些列作为参数传递给该存储过程。我使用pyodbc从Azure Databricks执行存储过程，编写了如下代码：

import pyodbc
connect_string = "DRIVER={ODBC Driver 17 for SQL Server};"
connect_string += f"SERVER={jdbchostname},"
connect_string += f"{jdbcport};"
connect_string += f"DATABASE={db};"
connect_string += f"UID={username};"
connect_string += f"PWD={pwd}"

conn = pyodbc.connect(connect_string)
cursor = conn.cursor()  


ColA = upsert_df['ColA']
ColB = upsert_df['ColB']
ColC = upsert_df['ColC']
    

proc_call = f'EXEC Proc_Name @ColA=?, @ColB=?, @ColC=?'
    
try:
   cursor.execute(proc_call, (ColA, ColB, ColC))
   conn.commit()
except Exception as e:
   print('Error=', e)   

执行代码时出现错误：Error= <class 'pyodbc.Error'> returned a result with an error set。请问我传递参数的方式是否正确，或是代码存在其他问题？

你的参数传递方式不正确，核心问题在于直接把DataFrame的整列（Series/Column对象）传递给存储过程参数，但存储过程需要的是单条记录的具体值，而非整列数据。另外代码还有其他细节问题，具体修正如下：

1. 参数传递错误修正 #

upsert_df['ColA']返回的是整列数据对象，无法直接作为存储过程的单值参数，需要遍历DataFrame的每一行，传递每行的字段值：

• 若使用Pandas DataFrame：

import pyodbc

# 修正连接字符串的SERVER格式，用冒号分隔主机和端口
connect_string = (
    f"DRIVER={{ODBC Driver 17 for SQL Server}};"
    f"SERVER={jdbchostname}:{jdbcport};"
    f"DATABASE={db};"
    f"UID={username};"
    f"PWD={pwd}"
)

conn = pyodbc.connect(connect_string)
cursor = conn.cursor()  

proc_call = 'EXEC Proc_Name @ColA=?, @ColB=?, @ColC=?'

try:
    # 遍历每行，传递单条记录的参数值
    for _, row in upsert_df.iterrows():
        cursor.execute(proc_call, (row['ColA'], row['ColB'], row['ColC']))
    conn.commit()
except pyodbc.Error as e:
    print('Error=', e)
    conn.rollback()  # 出错时回滚事务，避免数据不一致
finally:
    cursor.close()
    conn.close()  # 释放数据库连接资源

• 若使用Spark DataFrame（Azure Databricks常用），建议用foreach处理每行：

import pyodbc

def upsert_single_row(row):
    # 构建连接字符串
    connect_string = (
        f"DRIVER={{ODBC Driver 17 for SQL Server}};"
        f"SERVER={jdbchostname}:{jdbcport};"
        f"DATABASE={db};"
        f"UID={username};"
        f"PWD={pwd}"
    )
    conn = None
    cursor = None
    try:
        conn = pyodbc.connect(connect_string)
        cursor = conn.cursor()
        proc_call = 'EXEC Proc_Name @ColA=?, @ColB=?, @ColC=?'
        # 从Spark Row对象中直接取值
        cursor.execute(proc_call, (row.ColA, row.ColB, row.ColC))
        conn.commit()
    except pyodbc.Error as e:
        print('Error=', e)
        if conn:
            conn.rollback()
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

# 对DataFrame的每一行执行UPSERT
upsert_df.foreach(upsert_single_row)

2. 连接字符串的关键修正 #

原代码中SERVER部分用逗号分隔主机和端口，SQL Server ODBC驱动要求必须用冒号分隔（如SERVER=xxx:1433），这是可能导致连接失败的重要原因。

3. 其他必要优化 #

• 捕获具体的pyodbc.Error而非泛型Exception，便于定位数据库相关问题
• 添加事务回滚逻辑，避免部分数据提交引发的业务不一致
• 务必在最后关闭游标和连接，防止数据库资源泄漏

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