Python解析2GB大XML写入Oracle报MemoryError问题求助

API 探路者

2026-8-29

问题现象

使用Python将XML文件写入Oracle数据库时，2MB大小的XML文件可正常处理，处理2GB大小的XML文件时程序运行约2分钟触发报错，设备运行内存为8GB，报错信息如下：

self._root = parser._parse_whole(source)
MemoryError

问题根因

代码中使用的xml.etree.ElementTree的ET.parse()是全量加载模式，会把整个XML文件完整解析成DOM树存入内存。XML格式本身冗余度极高，2GB的XML文件解析为内存对象通常会占用原始文件大小10~20倍的内存，8GB运行内存无法承载。同时代码将所有解析结果全量存入rows列表后再一次性转DataFrame，会额外占用大量内存，最终触发内存溢出错误。

附原始参考信息

XML文件结构示例

<wmX
    xmlns:xs="http://www.w3.org/2001/XMLSchema"
    xmlns="http://www.wtest" majorRelease="97" minorRelease="1" currDate="2022-04-04+00:00" producer="wtest.core" name="id:19015 name:GTXML01" currSeqNo="304" >
    <finObj chgFlag="i" idName="Instrument" idVal="223434343">
        <ids>
            <id scheme="WKN">434535434</id>
            <id scheme="ISIN">TREE123123</id>
        </ids>
        <section chgFlag="i" idName="E" idVal="12345235564">
            <fld idName="ED001">
                <datVl chgFlag="i">4</datVl>
            </fld>
            <fld idName="ED005">
                <datVl chgFlag="i">14</datVl>
            </fld>
            <fld idName="ED006">
                <datVl chgFlag="i">91</datVl>
            </fld>
            <table chgFlag="i" idName="GV325">
                <row chgFlag="i" idVal="1">
                    <fld idName="GV325A">
                        <datVl chgFlag="i">1</datVl>
                    </fld>
                    <fld idName="GV325B">
                        <datVl chgFlag="i">2001-12-01</datVl>
                    </fld>
                    <fld idName="GV325C">
                        <datVl chgFlag="i">2006-04-30</datVl>
                    </fld>
                    <fld idName="GV325D">
                        <datVl chgFlag="i">01</datVl>
                    </fld>
                </row>
            </table>
        </section>
    </finObj>
</wmX>

问题原始代码

import pandas as pd
import xml.etree.ElementTree as ET
from collections import defaultdict
import xmltodict
import sqlalchemy
from sqlalchemy import create_engine
import cx_Oracle


username = "XXX"
password = "XXX-"
host = "XXX"
port = XXX
database = "XXXXX"
engine = create_engine(f'oracle://{username}:{password}@{host}:{port}/?service_name={database}', echo=False)

tree = ET.parse(r'P:\Kopie.xml')

table_name = "H_TEST"

root = tree.getroot()

# we need the namespace to call the objects
namespace = root.tag.split('}')[0].strip('{')

# the data is in two section
sections = list(list(root)[0])[1:]

# the default row is the data that is the same in each example
default_row = {
    "TR_NAME": root.get("name"),
    "TRE_CURRDATE": root.get("currDate"),
}


def add_new_row():
    """Nested function to create new row"""
    # we copy to create a depy copy of the default row
    new_row = default_row.copy()
    # new values
    new_row.update({
        "SECTION_IDNAME": section.get("idName"),
    })
    # the field inside is either datVl or txtVl
    for f in list(field):
        if f.tag[-5:] == "datVl":
            new_row.update({
                "DATVL_SECTION_DATA": f.text,
                "DATVL_SECTION_CHGFLAG": f.get("chgFlag")
            })
        if f.tag[-5:] == "txtVl":
            new_row.update({
                "TXTVL_SECTION_DATA": f.text,
            })
    return new_row


rows = []

# iterate through sections
for section in sections:
    # iterate through individual fields        
            for field in row.findall("xmlns:fld", namespaces={"xmlns": namespace}):
                new_row = add_new_row()
                new_row.update({
                    "TABLE_IDNAME": table.get("idName"),
                    "TABLE_CHGFLAG": table.get("chgFlag"),
                })
                rows.append(new_row)

# make a DataFrame
df = pd.DataFrame(rows)
# insert in database

解决方法

核心思路是流式逐块解析，避免全量加载，分批次入库不攒全量数据：

把ET.parse()换成ET.iterparse()迭代解析模式，仅在遇到指定标签时加载对应内容到内存，解析完成后立即释放无用节点，内存占用可稳定在几十MB级别。
不要把所有解析结果全量存在rows列表里最后一次性转DataFrame入库，攒够1000~5000条就批量写入Oracle，写完立刻清空缓存列表。
每解析完一个节点就手动调用clear()方法释放节点内存，避免无用节点残留占用空间。

修改后的核心逻辑参考：

import pandas as pd
import xml.etree.ElementTree as ET
from sqlalchemy import create_engine

# 数据库连接配置
username = "XXX"
password = "XXX-"
host = "XXX"
port = XXX
database = "XXXXX"
engine = create_engine(f'oracle://{username}:{password}@{host}:{port}/?service_name={database}', echo=False)
table_name = "H_TEST"
BATCH_SIZE = 2000  # 每2000条批量入库一次

ns = ""
default_row = {}
rows_cache = []

# 迭代解析，监听节点结束事件逐块处理
for event, elem in ET.iterparse(r'P:\Kopie.xml', events=("end",)):
    # 首次读取根节点wmX，提取公共属性和命名空间
    if elem.tag.endswith("wmX"):
        ns = elem.tag.split('}')[0].strip('{')
        default_row = {
            "TR_NAME": elem.get("name"),
            "TRE_CURRDATE": elem.get("currDate"),
        }
        elem.clear()
        continue
    
    # 处理section下的普通fld节点
    if elem.tag == f"{{{ns}}}fld" and elem.getparent().tag == f"{{{ns}}}section":
        new_row = default_row.copy()
        new_row["SECTION_IDNAME"] = elem.getparent().get("idName")
        new_row["TABLE_IDNAME"] = None
        new_row["TABLE_CHGFLAG"] = None
        # 提取datVl/txtVl字段值
        for child in elem:
            if child.tag.endswith("datVl"):
                new_row["DATVL_SECTION_DATA"] = child.text
                new_row["DATVL_SECTION_CHGFLAG"] = child.get("chgFlag")
                new_row["TXTVL_SECTION_DATA"] = None
            if child.tag.endswith("txtVl"):
                new_row["TXTVL_SECTION_DATA"] = child.text
                new_row["DATVL_SECTION_DATA"] = None
                new_row["DATVL_SECTION_CHGFLAG"] = None
        rows_cache.append(new_row)
        elem.clear()
    
    # 处理table下row内的fld节点
    if elem.tag == f"{{{ns}}}fld" and elem.getparent().tag == f"{{{ns}}}row":
        row_node = elem.getparent()
        table_node = row_node.getparent()
        section_node = table_node.getparent()
        new_row = default_row.copy()
        new_row["SECTION_IDNAME"] = section_node.get("idName")
        new_row["TABLE_IDNAME"] = table_node.get("idName")
        new_row["TABLE_CHGFLAG"] = table_node.get("chgFlag")
        for child in elem:
            if child.tag.endswith("datVl"):
                new_row["DATVL_SECTION_DATA"] = child.text
                new_row["DATVL_SECTION_CHGFLAG"] = child.get("chgFlag")
                new_row["TXTVL_SECTION_DATA"] = None
            if child.tag.endswith("txtVl"):
                new_row["TXTVL_SECTION_DATA"] = child.text
                new_row["DATVL_SECTION_DATA"] = None
                new_row["DATVL_SECTION_CHGFLAG"] = None
        rows_cache.append(new_row)
        elem.clear()

    # 缓存达到批量阈值就执行入库，清空缓存
    if len(rows_cache) >= BATCH_SIZE:
        pd.DataFrame(rows_cache).to_sql(table_name, engine, if_exists='append', index=False)
        rows_cache.clear()

# 处理最后剩余的不足批量数的缓存数据
if rows_cache:
    pd.DataFrame(rows_cache).to_sql(table_name, engine, if_exists='append', index=False)
    rows_cache.clear()

注意：Python 3.9以下版本的标准库xml.etree.ElementTree元素不支持getparent()方法，可手动维护节点父级关系字典，或直接换用lxml库的iterparse功能，自带父节点支持且解析速度更快。

内容的提问来源于stack exchange，提问作者Marcus

Python解析2GB大XML写入Oracle报MemoryError问题求助

XML文件结构示例 #

问题原始代码 #

XML文件结构示例

问题原始代码