使用fast_executemany向SQL Server插入mm/dd/yyyy字符串报错求助

问题背景 #

从BigQuery查询数据批量插入SQL Server 2008，开启fast_executemany=True时触发报错，关闭该参数后插入逻辑正常，但200万条数据耗时长达8小时。

伪代码实现 #

# 从BigQuery获取结果
client = bigquery.Client()
result = client.query(sql).result()

# 构造参数化插入语句
sql = 'INSERT INTO [SANDBOX].[dbo].[table_stg] VALUES ((?),(?),(?),(?),(?),(?),(?),(?),(?),(?))'

# 示例数据（实际为BigQuery返回结果）
data = [(2016, 'STRING', 'STRING', 'STRING', 'STRING', 'STRING', 123, 'STRING', '09/28/2015', '09/25/2016'), 
        (2016, 'STRING', 'STRING', 'STRING', 'STRING', 'STRING', 456, 'STRING', '09/28/2015', '09/25/2016')]

# SQL Server目标表结构
# [varchar](50) NULL,
# [varchar](50) NULL,
# [varchar](50) NULL,
# [varchar](50) NULL,
# [varchar](50) NULL,
# [varchar](50) NOT NULL,
# [int] NULL,
# [varchar](50) NULL,
# [datetime] NULL,
# [datetime] NULL

# 开启fast_executemany执行插入
cursor.fast_executemany = True
cursor.executemany(sql, data)
conn.commit()
cursor.close()
conn.close()

报错信息 #

Invalid character value for cast specification.
String data, right truncation: length 44 buffer 20, HY000

报错原因与解决方法 #

1. 字符串截断问题（right truncation） #

报错length 44 buffer 20说明：开启fast_executemany时，pyodbc会根据数据集第一条记录的字符串长度自动推断参数缓冲区大小，若后续记录存在更长的字符串，就会触发截断错误（即使目标表字段定义为varchar(50)）。

解决方式：手动指定每个参数的SQL类型，覆盖自动推断逻辑：

from pyodbc import SQL_VARCHAR, SQL_INTEGER, SQL_DATETIME

# 对应目标表10个字段的SQL类型
param_types = [
    SQL_VARCHAR(50),
    SQL_VARCHAR(50),
    SQL_VARCHAR(50),
    SQL_VARCHAR(50),
    SQL_VARCHAR(50),
    SQL_VARCHAR(50),
    SQL_INTEGER,
    SQL_VARCHAR(50),
    SQL_DATETIME,
    SQL_DATETIME
]

# 设置参数类型
cursor.setinputsizes(param_types)
cursor.fast_executemany = True
cursor.executemany(sql, data)

2. 日期格式转换问题（Invalid character value for cast specification） #

SQL Server 2008的datetime类型要求标准格式（如YYYY-MM-DD HH:MI:SS），而BigQuery返回的日期字符串是MM/DD/YYYY格式，fast_executemany下自动类型转换失败。

解决方式：将日期字符串转为Pythondatetime对象后再传入：

from datetime import datetime

# 处理数据中的日期字段（第9、10位）
processed_data = []
for row in data:
    dt1 = datetime.strptime(row[8], '%m/%d/%Y')
    dt2 = datetime.strptime(row[9], '%m/%d/%Y')
    processed_row = row[:8] + (dt1, dt2)
    processed_data.append(processed_row)

# 使用处理后的数据集执行插入
cursor.executemany(sql, processed_data)

fast_executemany与常规执行的自动切换机制 #

通过数据特征校验，自动选择最优执行模式：

• 使用fast_executemany的条件：
  1. 所有字符串字段长度均不超过目标表定义的长度
  2. 日期/数值字段类型与目标表完全匹配
• 回退到常规执行的条件：
  上述任一条件不满足时，直接使用常规executemany，或先清洗数据再启用fast_executemany。

实现示例 #

from datetime import datetime
from pyodbc import SQL_VARCHAR, SQL_INTEGER, SQL_DATETIME

def insert_data(cursor, sql, data, target_schema):
    # 目标表字段定义，格式如["varchar(50)", "int", "datetime"...]
    str_fields = [(i, int(col.split('(')[1].split(')')[0])) 
                  for i, col in enumerate(target_schema) if col.startswith('varchar')]
    date_field_indices = [i for i, col in enumerate(target_schema) if col == 'datetime']
    
    # 抽样验证数据有效性（取前1000条或全量）
    sample_size = min(1000, len(data))
    valid_for_fast = True
    
    # 检查字符串长度
    for row in data[:sample_size]:
        for idx, max_len in str_fields:
            if len(str(row[idx])) > max_len:
                valid_for_fast = False
                break
        if not valid_for_fast:
            break
    
    # 检查日期格式
    if valid_for_fast:
        for row in data[:sample_size]:
            for idx in date_field_indices:
                try:
                    datetime.strptime(row[idx], '%m/%d/%Y')
                except ValueError:
                    valid_for_fast = False
                    break
            if not valid_for_fast:
                break
    
    if valid_for_fast:
        # 处理日期并设置参数类型
        processed_data = []
        for row in data:
            dt_row = list(row)
            for idx in date_field_indices:
                dt_row[idx] = datetime.strptime(dt_row[idx], '%m/%d/%Y')
            processed_data.append(tuple(dt_row))
        
        param_types = []
        for col in target_schema:
            if col.startswith('varchar'):
                param_types.append(SQL_VARCHAR(int(col.split('(')[1].split(')')[0])))
            elif col == 'int':
                param_types.append(SQL_INTEGER)
            elif col == 'datetime':
                param_types.append(SQL_DATETIME)
        
        cursor.setinputsizes(param_types)
        cursor.fast_executemany = True
        cursor.executemany(sql, processed_data)
    else:
        # 常规执行
        cursor.fast_executemany = False
        cursor.executemany(sql, data)

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