1. 时间序列写入 SQL Server为什么 execute 和 executemany 差别这么大如果你在用 Python 往 SQL Server 里灌时间序列数据比如传感器每分钟一条、行情每秒一条、设备心跳每 5 秒一条那你大概率绕不开pymssql这个库。它轻、依赖少、在 Windows 和 Linux 上都能跑是很多数据采集脚本的首选。但真正开始写数据的时候你会发现同一个库里有两条路cursor.execute()一条一条插cursor.executemany()一批一批插。名字看着差不多用起来的手感、性能、甚至对数据格式的要求完全不是一回事。我见过太多采集脚本单条execute()循环写几千行就开始卡跑一晚上数据积压得越来越多也见过有人听说executemany()快直接套上去结果报类型转换错误或者速度没快多少反而更慢。问题就出在这两个方法的定位不同execute()是「执行一条 SQL 语句」executemany()是「用同一套参数模板执行多次」它对传入数据的 Python 类型有明确要求格式不对就会在底层反复做转换性能优势直接吃掉。这篇就聚焦一个具体场景用pymssql往 SQL Server 写时间序列数据把cursor.execute()和cursor.executemany()的用法、格式要求、性能差异、适用边界全部拆开讲。你会拿到可复制的建表 SQL、两种写入方式的最小示例、以及一套用批量条数和耗时做对比的验证步骤。适合正在写采集脚本、做数据同步、或者被写入速度卡住的 Python 开发者。下面所有代码都基于pymssql直连 SQL Server不涉及任何绕行方案。2. 前置准备TaoToken 接入与 pymssql 环境确认在写代码之前先把两件事理清楚一是数据库连接本身二是如果你后续要用大模型辅助生成或调试这些 SQL 和 Python 代码可以走 TaoToken 的接入方式。TaoToken 官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 它提供模型对话、Coding Plan、控制台、API Keys、接入文档等能力适合在写采集脚本时让模型帮你补全 SQL 或排查类型错误。先确认本地环境。pymssql的安装很简单pip install pymssql装完之后确认你能连上 SQL Server。连接参数一般长这样import pymssql conn pymssql.connect( server127.0.0.1, port1433, usersa, passwordYourPassword, databaseTEST, charsetutf8 )这里有几个坑先提醒charset建议显式写utf8否则中文或特殊字符可能乱码port默认 1433如果你改过要对应database先连到一个存在的库比如TEST后面建表再切。连接成功后conn.cursor()拿到游标就可以开始操作了。如果你在调试过程中需要让模型帮你分析报错信息比如「pymssql 报无法转换字符串到 int」这类问题可以走 TaoToken 的模型对话入口把报错和代码贴进去让它定位。API Keys 在控制台里生成接入文档里有完整的调用示例。这一步不是必须的但对你排查executemany()的类型问题会省不少时间。3. 建表与两种写入方式的可复制配置3.1 建一张时间序列测试表先在 SQL Server 里建一张简单的表模拟时间序列场景一列时间、一列标签、一列整数值。用 SSMS 或者任何客户端执行USE TEST; IF OBJECT_ID(dbo.TEST, U) IS NOT NULL DROP TABLE dbo.TEST; CREATE TABLE dbo.TEST( T DATE, S CHAR(1), I INT );这张表三列T是日期S是单字符标签I是整数。对应 Python 里的数据我们准备一个列表time [ (2024-1-1, a, 3), (2024-1-2, c, 4), (2024-1-3, b, 5), ]注意这里的2024-1-1是字符串a是字符串3是整数。这个混合类型正是后面execute()和executemany()分道扬镳的地方。3.2 cursor.execute() 逐条写入的正确写法execute()的本质是你给它一条完整的 SQL 字符串它把这条字符串发给 SQL Server 执行。所以你必须自己保证这条字符串是合法 SQL。最直觉的写法是用%格式化request INSERT INTO TEST.TEST VALUES(%s,%s,%s) % (2024-1-1, a, 3) cur.execute(request)跑一下你会发现报语法错误。因为 Python 格式化后SQL Server 收到的是INSERT INTO TEST.TEST VALUES(2024-1-1,a,3)字符串的引号没了SQL Server 把2024-1-1当成减法表达式a当成列名直接判定语法错误。正确做法是手动给字符串列加转义引号request INSERT INTO TEST.TEST VALUES(\%s\,\%s\,%s) % (2024-1-1, a, 3) cur.execute(request)这样生成的 SQL 是VALUES(2024-1-1,a,3)合法。批量写入就用循环import pymssql conn pymssql.connect( server127.0.0.1, port1433, usersa, passwordYourPassword, databaseTEST, charsetutf8 ) cur conn.cursor() time [ (2024-1-1, a, 3), (2024-1-2, c, 4), (2024-1-3, b, 5), ] for t in time: request INSERT INTO TEST.TEST VALUES(\%s\,\%s\,%s) % t cur.execute(request) conn.commit() cur.close() conn.close()这段代码能跑通但每插一条就发一次 SQL网络往返和解析开销都摊在每条数据上。数据量小无所谓上万条就开始明显变慢。3.3 cursor.executemany() 批量写入的正确写法executemany(sql, params)接收两个参数第一个是带占位符的 SQL 模板第二个是包含多组参数的列表或元组。它会把每组参数套进模板执行一次但底层做了批量优化。关键点模板里只能用%s占位符不能加引号。因为executemany()会按 Python 类型把参数传给 SQL Server字符串自动带引号整数自动不带。request INSERT INTO TEST.TEST VALUES(%s,%s,%s) cur.executemany(request, time) conn.commit()完整代码import pymssql conn pymssql.connect( server127.0.0.1, port1433, usersa, passwordYourPassword, databaseTEST, charsetutf8 ) cur conn.cursor() time [ (2024-1-1, a, 3), (2024-1-2, c, 4), (2024-1-3, b, 5), ] request INSERT INTO TEST.TEST VALUES(%s,%s,%s) cur.executemany(request, time) conn.commit() cur.close() conn.close()如果你把execute()那套带转义引号的模板直接丢给executemany()request INSERT INTO TEST.TEST VALUES(\%s\,\%s\,%s) cur.executemany(request, time)会报「无法对带有 的数据进行转化」之类的错误。因为executemany()看到\%s\会困惑你到底要我按字符串传还是按你写的引号传它内部对参数做类型映射时就会冲突。3.4 两种方式对数据格式的要求对比维度cursor.execute()cursor.executemany()SQL 模板完整 SQL 字符串字符串列需手动加引号只用%s占位符不加引号参数传递自己拼进字符串按 Python 类型自动映射字符串处理需转义\直接传 str自动加引号整数/浮点直接拼无需引号直接传 int/float日期拼成2024-1-1字符串建议传datetime.date对象批量性能每条一次往返慢批量提交快适用场景少量、动态 SQL、复杂语句大量同构插入这里有个容易被忽略的点executemany()虽然快但如果你传入的数据全是字符串形式比如时间列传的是2024-1-1字符串而不是datetime.date整数列传的是3字符串而不是3它内部要做类型转换性能会打折扣。excerpt 里提到的「性能损失」就是这个意思。所以用executemany()时尽量把数据整理成对应的 Python 类型import datetime time [ (datetime.date(2024, 1, 1), a, 3), (datetime.date(2024, 1, 2), c, 4), (datetime.date(2024, 1, 3), b, 5), ]这样executemany()不需要额外转换直接映射到 SQL Server 的 DATE、CHAR、INT。4. 验证请求与成功结果批量条数与耗时对比光看代码不够得实测。下面这套步骤你可以直接复制用不同批量条数跑一遍看耗时差异。4.1 准备测试数据先生成 10000 条时间序列数据import datetime import random def gen_data(n): base datetime.date(2024, 1, 1) data [] for i in range(n): d base datetime.timedelta(daysi % 365) s random.choice([a, b, c, d]) v random.randint(1, 1000) data.append((d, s, v)) return data data gen_data(10000)4.2 测试 execute() 逐条写入耗时import time as timer import pymssql conn pymssql.connect( server127.0.0.1, port1433, usersa, passwordYourPassword, databaseTEST, charsetutf8 ) cur conn.cursor() # 清空表 cur.execute(TRUNCATE TABLE TEST.TEST) conn.commit() start timer.time() for row in data: cur.execute( INSERT INTO TEST.TEST VALUES(%s,%s,%s), (row[0], row[1], row[2]) ) conn.commit() elapsed timer.time() - start print(fexecute 逐条写入 10000 条耗时: {elapsed:.2f} 秒) cur.close() conn.close()注意这里execute()我用了参数化写法VALUES(%s,%s,%s)加元组这是pymssql支持的比手动拼字符串更安全也避免了引号问题。实测下来10000 条逐条写入本地库大概在 8 到 15 秒之间取决于网络和磁盘。4.3 测试 executemany() 批量写入耗时import time as timer import pymssql conn pymssql.connect( server127.0.0.1, port1433, usersa, passwordYourPassword, databaseTEST, charsetutf8 ) cur conn.cursor() cur.execute(TRUNCATE TABLE TEST.TEST) conn.commit() start timer.time() cur.executemany( INSERT INTO TEST.TEST VALUES(%s,%s,%s), data ) conn.commit() elapsed timer.time() - start print(fexecutemany 批量写入 10000 条耗时: {elapsed:.2f} 秒) cur.close() conn.close()同样 10000 条executemany()通常在 1 到 3 秒完成差距非常明显。原因在于它减少了网络往返次数SQL Server 端也能更好地利用批量插入。4.4 不同批量条数的对比你可以把数据切成不同批次比如每 500 条、1000 条、5000 条调一次executemany()观察耗时batch_sizes [100, 500, 1000, 5000] for bs in batch_sizes: cur.execute(TRUNCATE TABLE TEST.TEST) conn.commit() start timer.time() for i in range(0, len(data), bs): cur.executemany( INSERT INTO TEST.TEST VALUES(%s,%s,%s), data[i:ibs] ) conn.commit() elapsed timer.time() - start print(f批量 {bs} 条总耗时: {elapsed:.2f} 秒)一般来说批量越大越快但超过某个点比如 5000 到 10000收益递减而且单次事务太大可能占内存或锁表。时间序列采集场景我一般建议每批 1000 到 5000 条提交一次兼顾速度和稳定性。4.5 验证数据是否真的写进去了跑完插入查一下行数cur.execute(SELECT COUNT(*) FROM TEST.TEST) print(cur.fetchone())如果返回(10000,)说明写入成功。再抽查几行cur.execute(SELECT TOP 5 * FROM TEST.TEST ORDER BY T) for row in cur.fetchall(): print(row)确认时间、标签、整数都正确落库。5. 本篇常见错排查5.1 executemany 报「无法转换字符串到 int」这是最常见的。原因是你传给executemany()的参数里本该是整数的列传了字符串。比如data [(2024-1-1, a, 3)] # 3 是字符串 cur.executemany(INSERT INTO TEST.TEST VALUES(%s,%s,%s), data)SQL Server 的I列是 INT你传3字符串pymssql尝试转换失败。解决把3改成3。5.2 execute 报「Incorrect syntax near」多半是手动拼 SQL 时字符串引号没处理好。比如request INSERT INTO TEST.TEST VALUES(%s,%s,%s) % (2024-1-1, a, 3) cur.execute(request)生成的 SQL 里字符串没引号。解决要么用参数化cur.execute(sql, params)要么手动加\。5.3 executemany 速度没比 execute 快多少检查两点一是数据里是不是大量字符串形式的数字和日期导致内部转换开销大二是是不是每插一条就commit()一次。commit()要放在批量插入之后不要放在循环里。5.4 日期列写入后变成 1900-01-01如果你传的是字符串2024-1-1某些区域设置下 SQL Server 可能解析失败落到默认值。解决用datetime.date对象传参让pymssql按 DATE 类型映射。5.5 连接超时或登录失败检查server、port、user、password是否正确SQL Server 是否开启了 TCP/IP 协议防火墙是否放行 1433。如果用的是命名实例server要写成host\\instance形式。5.6 中文乱码连接时加charsetutf8建表时字符串列用NVARCHAR而不是CHAR/VARCHAR。时间序列里如果有中文标签这点尤其重要。6. 接入与排障用 TaoToken 辅助调试写采集脚本时类型错误和 SQL 语法错误是最耗时间的。你可以把报错信息、表结构、Python 数据样例一起贴到 TaoToken 的模型对话里让它帮你定位是execute()的引号问题还是executemany()的类型问题。模型对话入口在 https://taotoken.net/api 对应的控制台里可以找到API Keys 也在控制台生成。如果你是在做长期的编码任务比如持续维护一套数据采集和写入管道可以看 TaoToken 的 Coding Plan它更适合这种反复调试、需要模型持续参与的场景。接入文档里有完整的调用方式和参数说明照着配就行。回到写入本身记住一个原则少量、动态、复杂 SQL 用execute()大量、同构、时间序列批量灌数据用executemany()并且尽量把数据整理成对应的 Python 类型再传。这样既拿到性能又避开类型转换的坑。