用pymongo流式读取需游标迭代+批量处理，设batch_size、禁用超时；导出JSON要处理ObjectId/日期，批量dump更高效；CSV需展平嵌套字段，注意索引和日志优化。
用 pymongo 连接并流式读取集合，避免内存爆掉
直接 list(collection.find()) 会把全部文档加载进内存，几百万条数据很容易 OOM。必须用游标迭代 + 批量处理。
- 设置
batch_size参数（如collection.find().batch_size(1000)），控制每次从 MongoDB 拉取的文档数 - 禁用游标超时：
collection.find(no_cursor_timeout=True)，防止大导出中途断连 - 加
.limit(n)或.skip().limit()做分片导出时，注意索引缺失会导致全表扫描，速度骤降
导出 JSON：用 json.dump() 配合 default 处理 ObjectId 和日期
MongoDB 的 ObjectId、datetime 类型不能直接 JSON 序列化，会报 TypeError: Object of type ObjectId is not JSON serializable。
- 定义转换函数：
default=lambda o: str(o) if hasattr(o, '_id') or isinstance(o, (ObjectId, datetime)) else None - 不要用
json.dumps(doc)单条写入——IO 开销大；改用json.dump(docs_list, f, default=...)批量写入 - 若需每行一个 JSON（JSONL 格式），用
json.dumps(doc, default=str)逐行写，比全量 dump 更省内存
导出 CSV：用 csv.DictWriter，但得先 flatten 嵌套字段
MongoDB 文档常含嵌套字典或列表，csv.DictWriter 只接受 flat 字典。硬展平容易丢数据或列名冲突。
- 简单场景：用
pandas.json_normalize(doc)处理单个文档，再合并成 DataFrame 导出（适合万级以内） - 大数据量别转 pandas：手写递归 flatten 函数，用
.连接层级（如user.profile.name），遇到列表则跳过或只取首项（需业务确认） - 字段名含逗号、换行符、双引号？
csv.writer默认能处理，但务必设quoting=csv.QUOTE_MINIMAL，别用QUOTE_ALL增大文件体积
性能关键：关掉 PyMongo 日志、用 find({}) 而非 find({}) 以外的空条件、避开 $where
导出慢，90% 不是网络或磁盘问题，而是查询本身低效。
立即学习“Python免费学习笔记（深入）”；
-
find({})是最优解；写成find({"_id": {"$exists": True}})或find({}).sort("_id")（无索引时）会强制扫描全表 - 确保
_id字段有默认索引（MongoDB 自带），排序导出就快；否则加.hint([("_id", 1)])显式指定 - 关掉 PyMongo 冗余日志：
import logging; logging.getLogger("pymongo").setLevel(logging.WARNING)
collection.find_raw_batches() 或投影 {"field1": 1, "field2": 1, "_id": 0} 减少传输和解析开销。 
