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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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 Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。需要直接对话验证模型行为时用模型对话：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 。接入细节和参数说明看文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
3. 可复制配置：SSCursor 流式游标 + 分批 yield 骨架
3.1 为什么 SSCursor 能省内存
普通Cursor在execute之后就把整个结果集缓存到客户端，fetchall只是把缓存交给你。SSCursor（Server-Side Cursor）不一样，它不在客户端缓存数据，而是从存储块里一条一条读、一条一条返回。内存占用从「结果集大小」降到「单行大小 + 网络缓冲」，这是质变。
代价也要说清楚：结果集没取完之前，这个连接不能执行别的 SQL，连再开一个 cursor 都不行；每次读取后处理要快，超过数据库的net_write_timeout（默认 60 秒）连接会被断。所以长任务要么单独开连接，要么调大超时。
3.2 流式读取 + 分批 yield 完整骨架
下面这段可以直接改连接参数后运行，核心是cursorclass=SSCursor加生成器分批产出：
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_size=2000, conf=DB_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_size=2000): total += 1 if total % 100000 == 0: print(f"processed {total} rows") print(f"done, total={total}")关键设计说明：fetchmany(batch_size)每次只从服务端拉一批，yield把单行交给调用方，调用方处理完这一行才继续拉下一批。整个链路里同时驻留内存的只有一批数据，批大小 2000 时内存峰值通常在几十 MB 量级。read_timeout和write_timeout设成 120 秒，给慢处理留出余量。
3.3 用 settings.json 驱动批大小
把批大小从配置文件读进来，方便不同任务调参而不用改代码：
import json def load_stream_conf(path="settings.json"): with open(path, "r", encoding="utf-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_size=stream_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_size=2000): total += 1 if total % 200000 == 0: current, peak = tracemalloc.get_traced_memory() print(f"rows={total} current={current/1e6:.1f}MB peak={peak/1e6:.1f}MB") current, peak = tracemalloc.get_traced_memory() print(f"total={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_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite ，接入参数对照文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，验证模型行为用模型对话 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite ，长期编码和 Agent 任务看 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。把这些入口和上面的骨架一起放进你的脚手架，下次再遇到百万级读取，改个连接串就能跑，不用重新踩一遍内存的坑。
