1. 为什么单条 execute 写 10 万行会让人等到怀疑人生先说结论execute和executemany不是「写法风格」的差别而是「网络往返次数」的差别。你写 Python 用 MySQLdb 往库里灌数据如果循环里一条一条execute那每插一行就是一次客户端到 MySQL 服务端的完整往返发 SQL、等服务端解析、等执行、等返回。10 万行就是 10 万次往返哪怕本机延迟只有 0.2ms光往返就 20 秒起步再加上每条 SQL 都要重新解析、重新生成执行计划实际耗时轻松飙到几分钟甚至几十分钟。executemany干的事情本质上是把「N 条 SQL」压成「1 次调用 N 组参数」驱动层会帮你把参数拼成批量语句MySQLdb 底层走的是INSERT ... VALUES (...),(...),(...)这种多值形式网络往返从 N 次降到 1 次SQL 解析也从 N 次降到 1 次。这就是为什么很多人实测下来原来要跑两三个小时的导入任务换成executemany之后几秒钟就结束了。这篇面向的是这几类场景数据导入CSV/Excel 落库、日志批量落库、爬虫结果批量写表、定时任务里的增量同步。适合已经会写基本cursor.execute但没系统对比过两种写法的人。我会给出可复制的连接参数、批量大小、事务提交配置再给一个计时脚本让你在自己机器上跑出真实差距而不是听我说「快很多」。顺带说一句很多同学在本地调通了脚本换到另一台机器或者换了个模型服务就各种报错其实是因为环境变量、Base URL、Key 没对齐。后面我会用 TaoToken 的配置验证流程把「本地脚本能跑」和「线上服务能连」这两件事串起来讲清楚避免你只会在自己电脑上跑。2. TaoToken 前置准备把 Key、Base URL、Model ID 三件套对齐在讲批量写入之前先把「配置验证」这条线铺好。因为实际项目里数据写入脚本往往还要调用模型做字段清洗、文本摘要、分类打标如果模型服务连不上整个流水线就卡住了。TaoToken 在这里扮演的是统一入口的角色你拿到一个 API Key配好 Base URL就能在脚本里调用模型不用为每个模型单独维护一套鉴权逻辑。先明确三件套缺一不可Base URLhttps://taotoken.net/api注意这里不加任何 UTM 参数直接用于代码里的请求地址API Key在控制台生成形如sk-xxxx只显示一次务必存到环境变量里Model ID具体调用的模型标识比如对话类、代码类各有不同以文档里列出的为准获取路径很直接打开官网https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content进控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content创建 API Keyhttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content模型和参数细节看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content。想先在网页里试一下模型通不通可以用模型对话页https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content。我建议你把 Key 写进环境变量而不是硬编码在脚本里export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在 Python 里读import os API_KEY os.environ[TAOTOKEN_API_KEY] BASE_URL os.environ[TAOTOKEN_BASE_URL] MODEL_ID 你的模型ID # 以文档为准这里有个坑要提前说Base URL 末尾不要多加/v1或者斜杠不同 SDK 对路径拼接的处理不一样多一个斜杠就可能 404。以文档给出的完整路径为准别自己猜。如果你用的是 Claude Code 这类命令行工具配置方式又不一样需要走 Anthropic 兼容入口https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content把 Base URL、Key、Model ID 三件套填进去。长期做编码和 Agent 任务的可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content配额和计费方式更适合持续跑任务。为什么要在批量写入的文章里讲这些因为真实的数据管道里写入和模型调用是连在一起的。你先把模型服务的连通性验证好再去调数据库批量写入排障的时候才能分清是「数据库慢」还是「模型服务连不上」。3. 可复制配置连接参数、批量大小与事务提交这一节给你可以直接抄的配置。先看 MySQLdb 的连接参数重点是charset和autocommitimport MySQLdb conn MySQLdb.connect( host127.0.0.1, port3306, userroot, passwdyour_password, dbmy_db, charsetutf8mb4, autocommitFalse, # 手动控制事务批量提交 ) cursor conn.cursor()charset用utf8mb4而不是utf8因为utf8在 MySQL 里其实是三字节的存 emoji 或者某些生僻字会报错。autocommitFalse是关键批量写入时如果每条都自动提交每次提交都要刷盘性能直接崩掉。手动提交能把 N 条写入合并成一次磁盘同步。然后是批量大小。这个值没有万能答案我给一个经验区间批量大小适用场景风险100 ~ 500行宽大、字段多、有唯一索引冲突往返次数偏多收益一般1000 ~ 5000常规数据导入、日志落库推荐区间收益和内存平衡10000行窄、纯追加、无冲突更新单条 SQL 过长可能撞max_allowed_packetmax_allowed_packet是 MySQL 服务端的限制默认可能是 4MB 或 16MB。批量太大时拼出来的 SQL 超过这个值就会报Packet too large。你可以先查一下SHOW VARIABLES LIKE max_allowed_packet;如果确实需要大批量要么调大服务端参数要么在代码里按字节数动态切分批次。再看executemany的写法注意%s不要加引号sql ( INSERT INTO my_table (created_day, name, count) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE count count VALUES(count) ) args [ (2024-08-27, name1, 100), (2024-08-27, name1, 200), (2024-08-27, name2, 300), ] try: cursor.executemany(sql, args) conn.commit() except Exception as e: conn.rollback() print(执行 MySQL 出错%s % e) finally: cursor.close() conn.close()这里有两个细节必须强调。第一created_day对应的%s不要写成%s。如果你加了引号驱动在参数替换时会把日期字符串当成带引号的字面量处理结果可能插入0000-00-00这种错误日期。第二ON DUPLICATE KEY UPDATE和executemany一起用时不要按常规思路在 UPDATE 子句里再写一个%s比如count count %s这会报not all arguments converted during string formatting。正确做法是用VALUES(count)引用本次插入的值参数只对应 VALUES 里的占位符。如果你要把这套配置写成 JSON 或者 TOML 方便复用可以这样{ mysql: { host: 127.0.0.1, port: 3306, user: root, db: my_db, charset: utf8mb4, autocommit: false }, batch: { size: 2000, table: my_table, columns: [created_day, name, count] }, taotoken: { base_url: https://taotoken.net/api, model_id: 你的模型ID } }TOML 版本[mysql] host 127.0.0.1 port 3306 user root db my_db charset utf8mb4 autocommit false [batch] size 2000 table my_table [taotoken] base_url https://taotoken.net/api model_id 你的模型ID把配置和代码分离换环境时只改配置文件不用动脚本逻辑。这也是我踩过的坑早期把连接参数写死在脚本里测试库和正式库来回切改一次错一次。4. 验证请求与成功结果计时脚本对比两种写法光说理论没用直接上计时脚本。下面这段代码会生成 5 万行测试数据分别用execute循环和executemany写入打印各自耗时。import time import MySQLdb DB_CONF dict( host127.0.0.1, port3306, userroot, passwdyour_password, dbmy_db, charsetutf8mb4, autocommitFalse, ) ROWS 50000 BATCH 2000 def gen_data(n): base time.time() return [ (2024-08-27, user_%d % i, i % 1000) for i in range(n) ] def write_one_by_one(data): conn MySQLdb.connect(**DB_CONF) cursor conn.cursor() sql INSERT INTO my_table (created_day, name, count) VALUES (%s, %s, %s) start time.time() try: for row in data: cursor.execute(sql, row) conn.commit() finally: cursor.close() conn.close() return time.time() - start def write_batch(data, batch_size): conn MySQLdb.connect(**DB_CONF) cursor conn.cursor() sql INSERT INTO my_table (created_day, name, count) VALUES (%s, %s, %s) start time.time() try: for i in range(0, len(data), batch_size): cursor.executemany(sql, data[i:i batch_size]) conn.commit() finally: cursor.close() conn.close() return time.time() - start if __name__ __main__: data gen_data(ROWS) t1 write_one_by_one(data) print(execute 逐条写入 %d 行耗时%.2f 秒 % (ROWS, t1)) t2 write_batch(data, BATCH) print(executemany 批量写入 %d 行耗时%.2f 秒 % (ROWS, t2)) print(提速倍数%.1fx % (t1 / t2 if t2 else 0))跑之前先建表CREATE TABLE my_table ( id INT AUTO_INCREMENT PRIMARY KEY, created_day DATE NOT NULL, name VARCHAR(64) NOT NULL, count INT NOT NULL, UNIQUE KEY uk_day_name (created_day, name) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;实测下来在本机 MySQL 上5 万行逐条execute大概要 40 到 90 秒取决于机器executemany按 2000 一批通常 1 到 3 秒就结束了。提速倍数在 20 到 50 倍之间数据量越大差距越明显。如果你的表有唯一索引并且走ON DUPLICATE KEY UPDATE批量写入的收益会更突出因为冲突检测也合并了。成功结果长这样execute 逐条写入 50000 行耗时62.31 秒 executemany 批量写入 50000 行耗时1.87 秒 提速倍数33.3x看到这个数字你就明白为什么开头说「两三个小时变两三秒」不是夸张。当然实际生产环境还有索引维护、binlog 写入、主从同步等开销倍数会小一些但量级上的差距是稳定的。如果你在写入前后还要调用模型做数据清洗可以在脚本里加一段连通性验证import requests resp requests.post( BASE_URL /chat/completions, headers{Authorization: Bearer API_KEY}, json{ model: MODEL_ID, messages: [{role: user, content: ping}], }, timeout10, ) print(resp.status_code) print(resp.json())返回 200 并且有正常的choices字段说明模型服务通了。这一步验证完再跑批量写入整条链路就都确认过了。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节按真实报错来对。你在跑批量写入和模型调用时大概率会撞上下面几个。报错一(1045, Access denied for user rootlocalhost)这是 MySQL 鉴权失败不是 TaoToken 的问题。检查user、passwd、host三处。注意localhost和127.0.0.1在 MySQL 里可能走不同的认证方式前者走 socket后者走 TCP。如果你用localhost连不上换成127.0.0.1试试。报错二(1153, Got a packet bigger than max_allowed_packet bytes)批量太大拼出来的 SQL 超过了服务端限制。两个办法调小BATCH或者调大服务端参数SET GLOBAL max_allowed_packet 64 * 1024 * 1024;注意这个设置重启后会失效要持久化得改配置文件。报错三not all arguments converted during string formatting这是executemany和ON DUPLICATE KEY UPDATE一起用时最经典的坑。原因是你 SQL 里的%s数量和参数元组里的元素数量对不上。比如# 错误写法 sql INSERT INTO t (a, b) VALUES (%s, %s) ON DUPLICATE KEY UPDATE b b %s args [(1, 2), (3, 4)] # 每个元组只有 2 个元素但 SQL 里有 3 个 %s正确做法是用VALUES(b)引用插入值SQL 里只保留 VALUES 部分的占位符sql INSERT INTO t (a, b) VALUES (%s, %s) ON DUPLICATE KEY UPDATE b b VALUES(b) args [(1, 2), (3, 4)]报错四模型调用返回 401{error: {message: Unauthorized, type: invalid_request_error}}说明 API Key 不对或者没带上。检查三件事Key 是否完整有没有复制时漏字符、请求头是否是Authorization: Bearer sk-xxx、环境变量是否真的被读到了。可以在脚本里打印一下API_KEY[:8]确认。报错五local proxy failed或连接超时这类报错通常出现在网络层说明请求根本没到服务端。检查你的 Base URL 是否写对末尾有没有多余的斜杠或路径。如果你在容器里跑确认容器能访问外网。注意不要使用任何非官方的网络工具直接用标准 HTTP 请求即可。报错六reading choices相关解析错误KeyError: choices或者TypeError: NoneType object is not subscriptable这通常是响应体不是预期的 JSON 结构。先打印resp.status_code和resp.text看看服务端到底返回了什么。常见原因是 Model ID 写错服务端返回了错误信息而不是正常的choices数组。报错七OAuth 相关错误如果你用的是 Claude Code 这类工具配置走的是 Anthropic 兼容入口出现 OAuth 报错一般是认证方式没选对。确认你填的是 API Key 而不是 OAuth tokenBase URL 用的是https://taotoken.net/api对应的兼容路径。三件套Base URL、Key、Model ID必须同时正确缺一个都会报错。排障的时候记住一个原则先分层再定位。数据库报错看 MySQL 错误码模型报错看 HTTP 状态码网络报错看连接层。不要一上来就怀疑代码逻辑大部分问题都在配置。6. 把批量写入和模型调用串成一条稳定流水线最后说点实操层面的经验。批量写入本身不复杂难的是把它放进一条稳定的流水线里还要和模型调用配合。第一分批提交而不是一次性提交。即使你用executemany也不要攒 100 万行一次性提交。按 2000 到 5000 一批每批提交一次这样内存占用可控出错时回滚的范围也小。我一般写成生成器边读数据边写def chunked(iterable, size): buf [] for item in iterable: buf.append(item) if len(buf) size: yield buf buf [] if buf: yield buf第二异常处理要区分「可重试」和「不可重试」。网络抖动导致的连接断开可以重试数据格式错误重试多少次都没用。给重试加个上限别写成死循环。第三模型调用和数据库写入解耦。如果模型服务暂时不可用不应该阻塞整个写入流程。可以把模型处理结果先落到本地队列或者临时表等模型恢复了再补。这样即使模型服务有波动数据也不会丢。第四监控耗时。在脚本里记录每批的写入耗时和模型调用耗时跑一段时间后你就能看出瓶颈在哪。如果写入耗时突然变长可能是索引膨胀或者锁竞争如果模型调用耗时变长可能是网络或者配额问题。关于 TaoToken 的使用长期跑编码和 Agent 任务的可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content按需选配额。需要管理多个 Key 或者查看调用量的去控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content。接入细节和参数说明以文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content为准别凭记忆写参数。想快速验证模型是否可用模型对话页https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content最省事。Key 的管理入口在 API Keyshttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content。回到批量写入本身记住核心结论单条execute适合低频、单行、需要立即拿到自增 ID 的场景executemany适合批量导入、日志落库、定时同步。批量大小从 2000 起步根据max_allowed_packet和内存调整。事务手动提交出错回滚。ON DUPLICATE KEY UPDATE配合VALUES()使用别在 UPDATE 子句里再写占位符。把这些配置跑通一次以后遇到类似任务直接套模板省下来的时间够你多写好几个功能。