Python脚本导入CSV至SQL Server时数据类型错误的解决咨询

问题概述 #

将CSV数据插入SQL Server现有表（保留原有数据）时触发TDS RPC协议错误，提示参数24不是有效的float类型。尝试过转换字段但未解决，需修复错误并实现每月自动化执行。

报错信息 #

Traceback (most recent call last):
  File "C:\Users\sampleUser\Desktop\Python Script\Food Services\Import to SSMS.py", line 55, in <module>
    cursor.execute(query, row['Visit Date'], row['Client ID'],
pyodbc.ProgrammingError: ('42000', '[42000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]The incoming tabular data stream (TDS) remote procedure call (RPC) protocol stream is incorrect. Parameter 24 (""): The supplied value is not a valid instance of data type float. Check the source data for invalid values. An example of an invalid value is data of numeric type with scale greater than precision. (8023) (SQLExecDirectW)')

现有代码 #

import pandas as pd
import pyodbc

#File path to CSV
file_path = r'C:\Users\sampleUser\Downloads\someFile.csv'

#Read the CSV file into a DataFrame
df = pd.read_csv(file_path)


#Establish connection parameters
server = 'sampleServer'
database = 'sampleDatabase'
username = 'sampleUser'
password = 'PW1234!@#$'

#Connection string
conn_str = (
    f"DRIVER={{ODBC Driver 17 for SQL Server}};"
    f"SERVER={server};"
    f"DATABASE={database};"
    f"UID={username};"
    f"PWD={password}"
    )
 
#Connect to SQL Server
conn = pyodbc.connect(conn_str)
cursor = conn.cursor()

#Define table name
table_name = 'SomeTable'

#Insert DataFrame records one by one
for index, row in df.iterrows():
  query = f"""
INSERT INTO {table_name} (Visit Date, Guest ID, Last Name, First Name,
Date of Birth, Guest Age, Guest Gender, Guest Ethnicity,
Guest Ethnicity 2, Street, Line 2, City, County, State, Zip Code,
Housing Type, Latitude, Longitude, Household ID, Household Size, Primary Language,
Diet Restrictions, Taken At (UTC), Taken By, Company Visited,
Company Visited (Short Name), Plan Name, Individuals Served)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""
  
  cursor.execute(query, row['Visit Date'], row['Guest ID'],
                   row['Last Name'], row['First Name'],
                   row['Date of Birth'], row['Guest Age'],
                   row['Guest Gender'], row['Guest Ethnicity'],
                   row['Guest Ethnicity 2'], row['Street'],
                   row['Line 2'], row['City'], row['County'],
                   row['State'], row['Zip Code'], row['Housing Type'],
                   row['Latitude'], row['Longitude'], row['Household ID'],
                   row['Household Size'], row['Primary Language'],
                   row['Diet Restrictions'], row['Taken At (UTC)'],
                   row['Taken By'], row['Company Visited'],
                   row['Company Visited(Short Name)'], row['Plan Name'],
                   row['Individuals Served'])
  conn.commit()

cursor.close()
conn.close()

已尝试的无效方案 #

# Convert specific column to float
columns_to_convert = ['Taken By']

for col in columns_to_convert:
    df[col] = pd.to_numeric(df[col], errors='coerce')

df['Recorded By'].fillna(0, inplace=True)

解决方案 #

1. 修复字段转换的拼写错误 #

原尝试方案中存在字段名不匹配：转换的是Taken By，但填充的是Recorded By，导致无效值未被处理。修正后代码：

# 将Taken By转换为数值类型，无效值转为NaN
df['Taken By'] = pd.to_numeric(df['Taken By'], errors='coerce')
# 填充NaN为表字段允许的默认值（如0）
df['Taken By'].fillna(0, inplace=True)

2. 匹配字段数据精度 #

数INSERT语句的参数顺序，第24个参数对应Taken By列。若SQL Server表中该字段为decimal(p,s)类型，需确保数据精度匹配：

# 示例：若表字段是decimal(10,2)，保留两位小数
df['Taken By'] = df['Taken By'].round(2)

3. 优化为批量插入（更适合自动化） #

逐行插入效率低且易出错，改用pandas.to_sql批量插入，无需手动匹配参数顺序：

import pandas as pd
from sqlalchemy import create_engine

file_path = r'C:\Users\sampleUser\Downloads\someFile.csv'
df = pd.read_csv(file_path)

# 预处理Taken By字段
df['Taken By'] = pd.to_numeric(df['Taken By'], errors='coerce').fillna(0)

# 构建SQLAlchemy连接字符串
conn_str = f"mssql+pyodbc://{username}:{password}@{server}/{database}?driver=ODBC+Driver+17+for+SQL+Server"
engine = create_engine(conn_str)

# 批量插入，if_exists='append'保留原有数据
df.to_sql(
    name='SomeTable',
    con=engine,
    if_exists='append',
    index=False,
    chunksize=1000  # 分块插入，适配大数据量
)

4. 预校验异常数据 #

插入前排查Taken By列的无效值，避免脏数据导致报错：

# 筛选无法转换为数值的行
invalid_rows = df[pd.to_numeric(df['Taken By'], errors='coerce').isna()]
print(f"发现{len(invalid_rows)}条无效数据：")
print(invalid_rows[['Taken By']])

可根据业务逻辑手动修正这些数据，或设置默认值处理。

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