从2.8GB XML文件提取数据并转存为Pandas DataFrame/CSV的技术求助

问题背景 #

存在一个大体积USPTO商标转让XML压缩包，无法直接全量导入Python程序，常规XML解析代码执行失败，且无法直接打开文件确认内部结构，需要将数据提取为Pandas DataFrame/CSV格式，或存入数据库，以便Python检索使用。

解决步骤 #

1. 先确认XML文件结构（避免全量加载） #

通过逐行读取压缩包内的XML文件头部内容，快速了解文件的层级结构和核心标签：

import zipfile

# 读取压缩包内XML的前100行
with zipfile.ZipFile("asb19550103-20211231-01.zip", 'r') as zf:
    xml_filename = zf.namelist()[0]
    with zf.open(xml_filename) as f:
        for _ in range(100):
            line = f.readline().decode('utf-8')
            print(line)

2. 流式XML解析（适配大文件） #

使用ElementTree的迭代解析功能，逐元素处理数据，无需一次性加载整个文件到内存：

import xml.etree.ElementTree as ET
import pandas as pd
import zipfile

def parse_large_xml(zip_path):
    data_rows = []
    with zipfile.ZipFile(zip_path, 'r') as zf:
        xml_filename = zf.namelist()[0]
        with zf.open(xml_filename) as f:
            # 初始化迭代解析器
            context = ET.iterparse(f, events=('start', 'end'))
            context = iter(context)
            event, root = next(context)
            
            current_record = {}
            # 替换为第一步查到的实际记录标签（比如<assignment>）
            record_tag = 'assignment'
            
            for event, elem in context:
                if event == 'end' and elem.tag == record_tag:
                    # 提取当前记录的所有字段
                    for child in elem:
                        current_record[child.tag] = child.text
                    data_rows.append(current_record.copy())
                    current_record.clear()
                    # 清理元素释放内存
                    root.clear()
    
    # 转换为DataFrame并返回
    return pd.DataFrame(data_rows)

# 执行解析并保存为CSV
df = parse_large_xml("asb19550103-20211231-01.zip")
df.to_csv("trademark_assignments.csv", index=False)

3. 数据库存储方案（超大规模数据适配） #

如果DataFrame仍占用过多内存，可直接将流式解析的数据存入SQLite，后续用Python直接查询：

import xml.etree.ElementTree as ET
import zipfile
import sqlite3

def xml_to_sqlite(zip_path, db_path):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # 先解析首个记录生成表结构
    with zipfile.ZipFile(zip_path, 'r') as zf:
        xml_filename = zf.namelist()[0]
        with zf.open(xml_filename) as f:
            context = ET.iterparse(f, events=('start', 'end'))
            context = iter(context)
            event, root = next(context)
            
            record_tag = 'assignment'
            for event, elem in context:
                if event == 'end' and elem.tag == record_tag:
                    columns = [child.tag for child in elem]
                    # 创建数据表
                    create_sql = f"CREATE TABLE IF NOT EXISTS assignments ({', '.join([f'{col} TEXT' for col in columns])})"
                    cursor.execute(create_sql)
                    root.clear()
                    break
    
    # 迭代插入所有数据
    with zipfile.ZipFile(zip_path, 'r') as zf:
        xml_filename = zf.namelist()[0]
        with zf.open(xml_filename) as f:
            context = ET.iterparse(f, events=('start', 'end'))
            context = iter(context)
            event, root = next(context)
            
            current_record = {}
            for event, elem in context:
                if event == 'end' and elem.tag == record_tag:
                    for child in elem:
                        current_record[child.tag] = child.text
                    # 插入单条记录
                    placeholders = ', '.join(['?' for _ in columns])
                    insert_sql = f"INSERT INTO assignments VALUES ({placeholders})"
                    cursor.execute(insert_sql, tuple(current_record[col] for col in columns))
                    current_record.clear()
                    root.clear()
    
    conn.commit()
    conn.close()

# 生成SQLite数据库
xml_to_sqlite("asb19550103-20211231-01.zip", "trademark_assignments.db")

关键注意事项 #

• 必须通过第一步的逐行读取，替换代码中record_tag为实际的记录级标签
• 流式解析过程中务必调用root.clear()，避免内存溢出
• 如果XML包含命名空间，elem.tag会带有命名空间前缀（如{http://uspto.gov}assignment），需提取纯标签名处理

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