将PySpark DataFrame写入Synapse专用池时遇ODBC驱动错误求助

错误根源 #

你遇到的('IM002', '[IM002] [unixODBC][Driver Manager]Data source name not found and no default driver specified (0) (SQLDriverConnect)')错误，核心问题是**pyodbc.connect()复用了Spark JDBC格式的URL，但pyodbc需要ODBC标准的连接字符串，两种格式不兼容**。另外你的CREATE TABLE语法存在错误（用了AS(而非标准的列定义括号）。

修正方案 #

方案1：用Spark JDBC统一完成表创建（推荐） #

放弃pyodbc，直接借助Spark JDBC的能力创建表，避免连接格式冲突：

import pyodbc

# 检查表是否存在的逻辑保持不变
def table_exists(table_name):
    query = f"SELECT TABLE_NAME AS table_name FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = '{table_name}'"
    try:
        result = spark.read.jdbc(url=jdbc_url, table=f"({query}) as temp", properties=connection_properties).collect()
        return len(result) > 0
    except Exception as e:
        print(f"Error checking if table exists: {e}")
        return False

# 修改表创建逻辑，用Spark JDBC生成表结构
def create_table(table_name):
    # 取DataFrame的空行来初始化表结构，自动匹配列类型
    empty_df = tblBooking_DIM_archive.limit(0)
    empty_df.write.jdbc(
        url=jdbc_url,
        table=f"dbo.{table_name}",
        mode="overwrite",
        properties={
            **connection_properties,
            "createTableOptions": "WITH (DISTRIBUTION = ROUND_ROBIN)"
        }
    )

# 表名定义
table_name = "tblBooking_DIM_archive"

# 写入逻辑
if table_exists(table_name):
    tblBooking_DIM_archive.write.jdbc(url=jdbc_url, table=table_name, mode="append", properties=connection_properties)
else:
    create_table(table_name)
    tblBooking_DIM_archive.write.jdbc(url=jdbc_url, table=table_name, mode="append", properties=connection_properties)

方案2：为pyodbc配置正确的ODBC连接字符串 #

如果必须保留pyodbc创建表的逻辑，需单独定义ODBC格式的连接字符串，不要复用Spark的JDBC URL：

import pyodbc

# 检查表是否存在的逻辑保持不变
def table_exists(table_name):
    query = f"SELECT TABLE_NAME AS table_name FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = '{table_name}'"
    try:
        result = spark.read.jdbc(url=jdbc_url, table=f"({query}) as temp", properties=connection_properties).collect()
        return len(result) > 0
    except Exception as e:
        print(f"Error checking if table exists: {e}")
        return False

# 定义ODBC专用连接字符串，替换为你的实际参数
odbc_conn_str = (
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=你的Synapse服务器名.sql.azuresynapse.net;"
    "DATABASE=你的数据库名;"
    "UID=你的用户名;"
    "PWD=你的密码;"
    "Encrypt=yes;"
    "TrustServerCertificate=no;"
    "Connection Timeout=30;"
)

# 修正CREATE TABLE语法错误
def create_table(table_name):
    create_table_sql = f"""
    CREATE TABLE dbo.{table_name} WITH (
    DISTRIBUTION = ROUND_ROBIN
    ) (
        SurrogateKey BIGINT,
        TenantID NVARCHAR(100),
        pkBookingID INT,
        fkUserID INT,
        fkLocationID INT,
        dateAdded DATETIME,
        trackDate DATETIME,
        fkHostID INT,
        noShow BIT,
        BookingSource INT,
        deleteBooking INT,
        RealStartDateTimeUTC DATETIME,
        RealEndDateTimeUTC DATETIME,
        bookingRealStartTime NVARCHAR(100),
        bookingRealEndTime NVARCHAR(100),
        cleanDownPeriod INT,
        cleanDownUnit INT,
        PipelineExecutionDate NVARCHAR(100)
    )
    """
    try:
        with pyodbc.connect(odbc_conn_str) as conn:
            with conn.cursor() as cursor:
                cursor.execute(create_table_sql)
                conn.commit()
    except Exception as e:
        print(f"Error creating table: {e}")

# 表名定义
table_name = "tblBooking_DIM_archive"

# 写入逻辑
if table_exists(table_name):
    tblBooking_DIM_archive.write.jdbc(url=jdbc_url, table=table_name, mode="append", properties=connection_properties)
else:
    create_table(table_name)
    tblBooking_DIM_archive.write.jdbc(url=jdbc_url, table=table_name, mode="overwrite", properties=connection_properties)

额外检查项 #

• 运行环境需安装对应版本的ODBC驱动（Linux下可安装msodbcsql18包）
• Spark JDBC的connection_properties必须包含driver=com.microsoft.sqlserver.jdbc.SQLServerDriver

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