用 iterparse 增量解析大 XML

功能说明

ElementTree 的 parse 会把整棵文档树读入内存，几百兆的导出文件很容易耗尽内存。iterparse 以事件流方式解析，只在遇到开始或结束标签时回调，允许边读边处理。本例监听结束事件，只关心名为 record 的元素：每当一个记录结束，就把它的直接子元素转换成字典产出，然后调用 clear 释放该元素及其子节点。

调用方得到的是生成器，可以逐条写入数据库、转换成 JSON 行或做聚合，内存占用与文件大小基本无关。演示先写一个两记录的小文件，再读取并列出结果，最后删除临时文件。

关键代码

ET.iterparse 接收路径和事件元组，返回迭代器；循环中的 element 是当前结束的元素。标签比较直接使用字符串；若文档带命名空间，标签会变成花括号形式，需要先统一。字典推导遍历直接子元素，把文本去空白后作为值，缺失文本用空字符串。yield 之后才 clear，确保调用方已拿到需要的数据。主程序用 try/finally 保证临时文件被删除。

使用注意

clear 只释放当前元素，父节点仍保留对已清空子节点的引用，长时间解析时还应对父节点做清理，或改用 XMLPullParser 控制。命名空间、属性和重复子标签需要额外处理，本例只保留直接子元素的文本。XML 实体扩展可能造成 billion laughs 攻击，处理不可信文档应使用 defusedxml 或禁用实体解析。流式解析无法回退，若业务需要随机访问应改用数据库或索引。

验证方式

运行脚本，输出包含两条记录，id 分别为 1 和 2，name 为 Ada 与 Grace，说明逐条解析正确。可把 fixture 改成上千条记录并在循环中打印内存占用，观察内存不会随记录数线性增长。若去掉 clear，内存曲线会明显上升。

代码示例

import xml.etree.ElementTree as ET
from collections.abc import Iterator
from pathlib import Path


def iter_records(path: Path, tag: str = "record") -> Iterator[dict[str, str]]:
    context = ET.iterparse(path, events=("end",))
    for _, element in context:
        if element.tag != tag:
            continue
        record = {
            child.tag: (child.text or "").strip() for child in element
        }
        yield record
        element.clear()


def write_fixture(path: Path) -> None:
    path.write_text(
        "<catalog>"
        "<record><id>1</id><name>Ada</name></record>"
        "<record><id>2</id><name>Grace</name></record>"
        "</catalog>",
        encoding="utf-8",
    )


if __name__ == "__main__":
    fixture = Path("catalog.xml")
    write_fixture(fixture)
    try:
        print(list(iter_records(fixture)))
    finally:
        fixture.unlink()