如何使用Python将大型XML文件数据插入Postgresql表并展示

1. 环境依赖安装 #

先安装需要的第三方库：
pip install psycopg2-binary lxml

2. 建表语句 #

首先在PostgreSQL中创建对应字段的表，可根据业务需求调整字段类型：

CREATE TABLE clearing_operations (
    id SERIAL PRIMARY KEY,
    file_id VARCHAR(32) NOT NULL,
    file_type VARCHAR(16) NOT NULL,
    start_date DATE NOT NULL,
    end_date DATE NOT NULL,
    inst_id VARCHAR(16) NOT NULL,
    file_date DATE NOT NULL,
    oper_id VARCHAR(32) NOT NULL UNIQUE,
    oper_type VARCHAR(16) NOT NULL,
    msg_type VARCHAR(16) NOT NULL,
    sttl_type VARCHAR(16) NOT NULL,
    oper_date TIMESTAMP NOT NULL,
    host_date TIMESTAMP NOT NULL,
    oper_amount NUMERIC(18,2) NOT NULL,
    oper_currency VARCHAR(3) NOT NULL,
    request_amount NUMERIC(18,2) NOT NULL,
    request_currency VARCHAR(3) NOT NULL
);

3. 完整实现代码 #

核心逻辑说明： #

• 采用lxml.iterparse增量解析大型XML，避免全量加载占用过高内存
• 自动适配XML自带命名空间，无需修改原始XML结构
• 批量提交数据到PostgreSQL，大幅提升入库效率

import psycopg2
from lxml import etree

# 配置参数
XML_PATH = "替换为你的XML文件实际路径"
DB_CONFIG = {
    "host": "替换为数据库地址",
    "port": 5432,
    "user": "替换为数据库用户名",
    "password": "替换为数据库密码",
    "dbname": "替换为数据库名称"
}
NS = {"sv": "http://bpc.ru/sv/SVXP/clearing"} # 对应示例XML的命名空间
BATCH_SIZE = 1000 # 每1000条提交一次，可根据服务器性能调整

def main():
    # 初始化数据库连接
    conn = psycopg2.connect(**DB_CONFIG)
    cur = conn.cursor()
    insert_sql = """
        INSERT INTO clearing_operations (
            file_id, file_type, start_date, end_date, inst_id, file_date,
            oper_id, oper_type, msg_type, sttl_type, oper_date, host_date,
            oper_amount, oper_currency, request_amount, request_currency
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    """
    batch_data = []
    file_level_fields = {}
    # 增量解析XML
    for event, elem in etree.iterparse(XML_PATH, events=("end",), tag=["{"+NS["sv"]+"}clearing", "{"+NS["sv"]+"}operation"]):
        tag_name = elem.tag.split("}")[-1]
        if tag_name == "clearing":
            # 提取文件级公共字段
            file_level_fields = {
                "file_id": elem.xpath("sv:file_id/text()", namespaces=NS)[0],
                "file_type": elem.xpath("sv:file_type/text()", namespaces=NS)[0],
                "start_date": elem.xpath("sv:start_date/text()", namespaces=NS)[0],
                "end_date": elem.xpath("sv:end_date/text()", namespaces=NS)[0],
                "inst_id": elem.xpath("sv:inst_id/text()", namespaces=NS)[0],
                "file_date": elem.xpath("sv:file_date/text()", namespaces=NS)[0]
            }
        elif tag_name == "operation":
            # 提取单条操作明细字段
            oper_data = {
                "oper_id": elem.xpath("sv:oper_id/text()", namespaces=NS)[0],
                "oper_type": elem.xpath("sv:oper_type/text()", namespaces=NS)[0],
                "msg_type": elem.xpath("sv:msg_type/text()", namespaces=NS)[0],
                "sttl_type": elem.xpath("sv:sttl_type/text()", namespaces=NS)[0],
                "oper_date": elem.xpath("sv:oper_date/text()", namespaces=NS)[0],
                "host_date": elem.xpath("sv:host_date/text()", namespaces=NS)[0],
                "oper_amount": elem.xpath("sv:oper_amount/sv:amount_value/text()", namespaces=NS)[0],
                "oper_currency": elem.xpath("sv:oper_amount/sv:currency/text()", namespaces=NS)[0],
                "request_amount": elem.xpath("sv:oper_request_amount/sv:amount_value/text()", namespaces=NS)[0],
                "request_currency": elem.xpath("sv:oper_request_amount/sv:currency/text()", namespaces=NS)[0]
            }
            # 合并字段生成入库行
            row = (
                file_level_fields["file_id"], file_level_fields["file_type"], file_level_fields["start_date"],
                file_level_fields["end_date"], file_level_fields["inst_id"], file_level_fields["file_date"],
                oper_data["oper_id"], oper_data["oper_type"], oper_data["msg_type"], oper_data["sttl_type"],
                oper_data["oper_date"], oper_data["host_date"], oper_data["oper_amount"], oper_data["oper_currency"],
                oper_data["request_amount"], oper_data["request_currency"]
            )
            batch_data.append(row)
            # 达到批量阈值提交数据
            if len(batch_data) >= BATCH_SIZE:
                cur.executemany(insert_sql, batch_data)
                conn.commit()
                batch_data = []
            # 清理已解析节点释放内存
            elem.clear()
            while elem.getprevious() is not None:
                del elem.getparent()[0]
    # 提交剩余未入库数据
    if batch_data:
        cur.executemany(insert_sql, batch_data)
        conn.commit()
    cur.close()
    conn.close()

if __name__ == "__main__":
    main()

4. 数据查询展示 #

入库完成后可以直接在PostgreSQL中查询数据，示例查询语句：

-- 查询前10条操作记录
SELECT oper_id, oper_date, oper_amount, oper_currency FROM clearing_operations LIMIT 10;

如果需要用Python展示查询结果，可以新增如下代码：

def show_query_result():
    conn = psycopg2.connect(**DB_CONFIG)
    cur = conn.cursor()
    cur.execute("SELECT oper_id, oper_date, oper_amount, oper_currency FROM clearing_operations LIMIT 10")
    results = cur.fetchall()
    print("操作ID|操作时间|金额|币种")
    print("---|---|---|---")
    for row in results:
        print(f"{row[0]}|{row[1]}|{row[2]}|{row[3]}")
    cur.close()
    conn.close()

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