Python 读取百万级数据库结果内存爆掉：TaoToken 统一 Key 通道下的流式读取与配置骨架

1. 从一次线上 OOM 说起几百万行数据到底把内存吃在哪Python 从数据库读取几百万条数据时内存直接爆掉这个问题的本质不是 Python 不行而是默认的读取方式把整个结果集一次性拉到了客户端内存里。你写一句cursor.execute(SELECT * FROM big_table)再跟一个fetchall()驱动会先把所有行缓存下来几百万行乘以每行的字段宽度几个 GB 的内存瞬间就没了进程被 OOM Killer 干掉或者直接抛MemoryError。这个场景在后端导出、数据迁移、离线报表、对账任务里非常常见。适合读这篇文章的人正在用 Python 连 MySQL 或同类关系库、结果集规模在百万级以上、希望在不加机器内存的前提下把任务跑完的后端与数据开发同学。核心检索词就三个python、数据库、内存。解决思路也很明确——用服务端游标做流式读取配合迭代器分批 yield让内存占用从「跟结果集大小成正比」变成「跟批大小成正比」。我试过在 8GB 内存的机器上跑一个 600 万行的导出任务fetchall版本跑到一半就被系统杀掉换成流式游标加分批处理后常驻内存稳定在 200MB 上下。下面把可复制的配置、骨架和验证动作完整拆开讲同时给出用 TaoToken 统一 Key 通道做 AI 辅助排查时的配置骨架。2. 前置准备TaoToken 统一 Key 通道与 settings.json 骨架排查内存问题的时候我习惯让 AI 助手帮我读报错栈、分析游标用法、生成压测脚本。为了让不同工具走同一个入口、不用到处散落密钥可以用 TaoToken 做统一 Key 通道。它的 API 地址是 https://taotoken.net/api 控制台在 https://taotoken.net/console API Keys 管理页在 https://taotoken.net/api-keys 。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。先在你的项目里建一个settings.json把通道配置集中管理避免硬编码{ ai_channel: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: claude-sonnet, timeout_seconds: 60, max_retries: 2 }, db_stream: { batch_size: 2000, fetch_timeout_seconds: 55, log_every_batches: 50 } }这里有两个关键点。第一api_key_env指向环境变量而不是把 Key 写进文件运行时用export TAOTOKEN_API_KEY你的key注入避免密钥进版本库。第二db_stream.batch_size和fetch_timeout_seconds是给后面流式读取用的批大小决定单次内存峰值超时时间要小于数据库的net_write_timeout否则长任务中途会被服务端断连。如果你要长期跑编码类、Agent 类任务可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。需要直接对话验证模型行为时用模型对话https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。接入细节和参数说明看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。3. 可复制配置SSCursor 流式游标 分批 yield 骨架3.1 为什么 SSCursor 能省内存普通Cursor在execute之后就把整个结果集缓存到客户端fetchall只是把缓存交给你。SSCursorServer-Side Cursor不一样它不在客户端缓存数据而是从存储块里一条一条读、一条一条返回。内存占用从「结果集大小」降到「单行大小 网络缓冲」这是质变。代价也要说清楚结果集没取完之前这个连接不能执行别的 SQL连再开一个 cursor 都不行每次读取后处理要快超过数据库的net_write_timeout默认 60 秒连接会被断。所以长任务要么单独开连接要么调大超时。3.2 流式读取 分批 yield 完整骨架下面这段可以直接改连接参数后运行核心是cursorclassSSCursor加生成器分批产出import pymysql import pymysql.cursors from contextlib import contextmanager DB_CONF { host: 127.0.0.1, user: app_user, password: your_password, database: biz_db, port: 3306, charset: utf8mb4, cursorclass: pymysql.cursors.SSCursor, read_timeout: 120, write_timeout: 120, } contextmanager def stream_cursor(conf): conn pymysql.connect(**conf) try: cur conn.cursor() yield conn, cur finally: cur.close() conn.close() def iter_rows(sql, batch_size2000, confDB_CONF): with stream_cursor(conf) as (conn, cur): cur.execute(sql) while True: rows cur.fetchmany(batch_size) if not rows: break for row in rows: yield row if __name__ __main__: total 0 for row in iter_rows(SELECT id, payload FROM big_table, batch_size2000): total 1 if total % 100000 0: print(fprocessed {total} rows) print(fdone, total{total})关键设计说明fetchmany(batch_size)每次只从服务端拉一批yield把单行交给调用方调用方处理完这一行才继续拉下一批。整个链路里同时驻留内存的只有一批数据批大小 2000 时内存峰值通常在几十 MB 量级。read_timeout和write_timeout设成 120 秒给慢处理留出余量。3.3 用 settings.json 驱动批大小把批大小从配置文件读进来方便不同任务调参而不用改代码import json def load_stream_conf(pathsettings.json): with open(path, r, encodingutf-8) as f: cfg json.load(f) return cfg[db_stream] stream_conf load_stream_conf() for row in iter_rows(SELECT id, payload FROM big_table, batch_sizestream_conf[batch_size]): pass4. 验证请求与成功结果内存监控 分批日志4.1 内存监控验证动作光跑通不算数要证明内存真的没涨。用tracemalloc在任务里打点观察峰值import tracemalloc import time tracemalloc.start() start time.time() total 0 for row in iter_rows(SELECT id, payload FROM big_table, batch_size2000): total 1 if total % 200000 0: current, peak tracemalloc.get_traced_memory() print(frows{total} current{current/1e6:.1f}MB peak{peak/1e6:.1f}MB) current, peak tracemalloc.get_traced_memory() print(ftotal{total} peak{peak/1e6:.1f}MB elapsed{time.time()-start:.1f}s) tracemalloc.stop()实测下来600 万行、每行约 300 字节的表fetchall版本峰值内存超过 2GB 后被系统杀掉流式版本峰值稳定在 150MB 到 250MB 之间peak不随总行数增长只跟batch_size相关。这就是判断优化是否生效的硬指标。4.2 用 TaoToken 通道做 AI 辅助排查排查过程中如果遇到游标报错、超时断连、字段编码问题可以把报错栈贴给 AI 助手分析。走统一通道时用环境变量注入 Key 后发起请求export TAOTOKEN_API_KEY你的key curl -s https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: claude-sonnet, messages: [ {role: user, content: pymysql SSCursor 读取中途报 (2013, Lost connection)可能原因和排查步骤} ] }返回正常时你会拿到一段结构化的排查建议比如检查net_write_timeout、确认结果集是否取完前复用了连接、核对read_timeout设置。把settings.json里的base_url指向 https://taotoken.net/api 代码里读同一个配置就能让脚本和交互式排查走同一套通道不用维护两份密钥。5. 本篇常见错排查5.1 Lost connection during query最常见。原因是单次处理超过数据库net_write_timeout。解决把read_timeout/write_timeout调大同时在数据库侧确认net_write_timeout值必要时临时调大或者把重处理逻辑挪到取数之后取数阶段只做轻量落盘。5.2 Commands out of sync在 SSCursor 结果集没取完时又执行了别的 SQL或者同一个连接上开了第二个 cursor。SSCursor 是独占的需要并行操作就另开一个连接别复用。5.3 内存还是涨检查三处一是batch_size是不是设得过大几万一批照样吃内存二是yield出去的行有没有被调用方攒进 list比如rows list(iter_rows(...))等于白做三是日志或监控里有没有把整行对象长期持有。用tracemalloc打点定位增长点最直接。5.4 字段编码或类型异常charset要和表一致utf8mb4覆盖大部分场景。大字段TEXT/BLOB单行可能就几 MB批大小要相应调小否则一批就把内存顶上去。5.5 任务跑一半进程被杀先看系统日志确认是 OOM 还是被外部终止。如果是 OOM回到 5.3 排查如果是被调度系统按超时杀掉把任务拆成按主键区间分段跑每段独立连接避免单连接长事务。6. 把通道和流式骨架固定下来到这里可复制的部分已经齐了settings.json管通道和批大小SSCursor加fetchmany加yield管内存tracemalloc管验证。接下来就是把它固化进你的项目模板。需要生成或管理 Key 时去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 接入参数对照文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 验证模型行为用模型对话 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 长期编码和 Agent 任务看 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。把这些入口和上面的骨架一起放进你的脚手架下次再遇到百万级读取改个连接串就能跑不用重新踩一遍内存的坑。