在Databricks中使用PYODBC连接Azure SQL执行更新操作遇驱动问题

一、解决驱动找不到的错误 #

错误Can't open lib 'SQL Server' : file not found的核心原因是Databricks集群（Linux环境）未安装适用于Linux的Microsoft SQL Server ODBC驱动，且连接字符串误用了Windows平台的驱动名称。

1. 安装ODBC驱动 #

通过集群初始化脚本在集群启动时自动安装驱动（确保每次集群重启后驱动都存在）：

• 打开Databricks集群配置，找到"初始化脚本"选项，添加以下bash脚本（适用于Ubuntu/Debian系的Databricks Runtime）：

#!/bin/bash
# 更新包列表
sudo apt-get update -y
# 安装依赖工具
sudo apt-get install -y curl apt-transport-https
# 添加微软官方源
curl https://packages.microsoft.com/keys/microsoft.asc | sudo apt-key add -
curl https://packages.microsoft.com/config/ubuntu/20.04/prod.list | sudo tee /etc/apt/sources.list.d/mssql-release.list
# 安装ODBC驱动18版本（也可替换为msodbcsql17）
sudo apt-get update -y
sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18

• 保存集群配置并重启集群，确保驱动安装完成。

2. 修正连接字符串 #

Linux平台的驱动名称与Windows不同，且需适配Azure SQL Server的加密要求，正确的连接字符串写法：

import pyodbc

# 配置参数
driver = "ODBC Driver 18 for SQL Server"  # 与安装的驱动版本对应，17版本则写"ODBC Driver 17 for SQL Server"
jdbcHostname_dev = "your-server.database.windows.net"
jdbcDatabase_dev = "your-db-name"
sql_user = "your-username"
sql_password = "your-password"

# 建立连接
cnx = pyodbc.connect(
    f"DRIVER={{{driver}}};SERVER={jdbcHostname_dev};DATABASE={jdbcDatabase_dev};UID={sql_user};PWD={sql_password};Encrypt=yes;TrustServerCertificate=yes"
)

注意事项：

• 移除Trusted_Connection=yes：该参数用于Windows集成认证，Linux环境下无法使用
• 用分号分隔参数（原代码中user={},password={}的逗号是错误的）
• 添加Encrypt=yes和TrustServerCertificate=yes：Azure SQL Server强制要求加密连接

二、执行更新/插入操作 #

连接成功后，即可通过pyodbc执行SQL语句，注意必须手动提交事务：

1. 单条更新操作 #

cursor = cnx.cursor()

# 编写更新SQL（使用参数化查询避免SQL注入）
update_sql = """
UPDATE your_table
SET target_column = ?
WHERE id = ?
"""
# 执行更新
cursor.execute(update_sql, ("updated_value", 123))
# 提交事务
cnx.commit()

# 关闭游标和连接
cursor.close()
cnx.close()

2. 批量插入/更新操作 #

处理大量数据时，使用executemany提升效率：

import pandas as pd

# 假设已有Spark DataFrame需要写入
spark_df = spark.table("your_spark_table")
pandas_df = spark_df.toPandas()

cnx = pyodbc.connect(...)
cursor = cnx.cursor()

# 批量插入示例
insert_sql = """
INSERT INTO your_table (col1, col2, col3)
VALUES (?, ?, ?)
"""
# 将Pandas DataFrame转为元组列表
data_tuples = [tuple(row) for row in pandas_df.values]
# 批量执行
cursor.executemany(insert_sql, data_tuples)
cnx.commit()

cursor.close()
cnx.close()

三、高效处理大数据的替代方案 #

如果需要处理超大数据集，单节点的pyodbc效率较低，可结合Spark的foreachBatch实现分布式批量处理：

def upsert_batch(batch_df, batch_id):
    # 将当前批次的Spark DataFrame转为Pandas DataFrame
    pandas_batch = batch_df.toPandas()
    # 建立连接并执行批量操作
    cnx = pyodbc.connect(...)
    cursor = cnx.cursor()
    
    upsert_sql = """
    MERGE INTO your_table AS target
    USING (VALUES (?, ?, ?)) AS source(id, col1, col2)
    ON target.id = source.id
    WHEN MATCHED THEN UPDATE SET target.col1 = source.col1, target.col2 = source.col2
    WHEN NOT MATCHED THEN INSERT (id, col1, col2) VALUES (source.id, source.col1, source.col2)
    """
    
    data_tuples = [tuple(row) for row in pandas_batch.values]
    cursor.executemany(upsert_sql, data_tuples)
    cnx.commit()
    
    cursor.close()
    cnx.close()

# 对Spark DataFrame执行批量Upsert
spark_df.foreachBatch(upsert_batch)

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