从MySQL的Blob型XML数据提取指定格式数据（5000万行）

MySQL端优化方案（规避查询超时） #

1. 分批次查询+临时表存储 #

针对单条大查询超时问题，通过分批次提取数据并存储到临时表，最后统一聚合统计：

• 创建临时表存储中间结果：

CREATE TEMPORARY TABLE xml_extract_result (
    source_type ENUM('Source A', 'Source B'),
    error_code VARCHAR(50),
    count INT DEFAULT 0
) ENGINE=InnoDB;

• 分批次循环提取数据（按id分段，每次处理10万行）：

SET @last_id = '';
WHILE EXISTS (SELECT 1 FROM your_table WHERE id > @last_id LIMIT 1) DO
    INSERT INTO xml_extract_result (source_type, error_code)
    SELECT
        CASE
            WHEN EXTRACTVALUE(req_data, '//source') = 'A' THEN 'Source A'
            WHEN EXTRACTVALUE(req_data, '//source') = 'B' THEN 'Source B'
            ELSE NULL
        END AS source_type,
        EXTRACTVALUE(req_data, '//error_code') AS error_code
    FROM your_table
    WHERE id > @last_id
    ORDER BY id
    LIMIT 100000;

    SET @last_id = (SELECT MAX(id) FROM your_table WHERE id > @last_id ORDER BY id LIMIT 100000);
END WHILE;

• 聚合临时表得到最终结果：

SELECT source_type, error_code, COUNT(*) AS total_count
FROM xml_extract_result
WHERE source_type IS NOT NULL AND error_code IS NOT NULL
GROUP BY source_type, error_code;

2. 调整会话超时参数+索引优化 #

• 延长单次查询的超时时间（会话级生效）：

SET SESSION MAX_EXECUTION_TIME = 300000; -- 设置为5分钟，单位毫秒

• 确保id列已设为主键或唯一索引，避免全表扫描时的性能损耗。

Python批量处理方案（精准解析+本地聚合） #

1. 依赖安装 #

pip install mysql-connector-python lxml

2. 核心代码实现 #

通过分批次拉取数据，用lxml精准解析XML（规避MySQL内置函数的解析缺陷），本地完成聚合统计：

import mysql.connector
from lxml import etree
from collections import defaultdict

# 数据库配置
db_config = {
    'host': '你的数据库地址',
    'user': '你的用户名',
    'password': '你的密码',
    'database': '你的数据库名',
    'charset': 'utf8mb4'
}

# 初始化聚合统计字典
stats = defaultdict(lambda: defaultdict(int))

def parse_xml(xml_data):
    """解析XML，提取source和error_code"""
    try:
        root = etree.fromstring(xml_data)
        source = root.findtext('.//source')
        error_code = root.findtext('.//error_code')
        source_type = 'Source A' if source == 'A' else 'Source B' if source == 'B' else None
        return source_type, error_code
    except Exception:
        # 跳过解析失败的无效XML数据
        return None, None

def batch_process(batch_size=100000):
    conn = mysql.connector.connect(**db_config)
    cursor = conn.cursor(dictionary=True)
    
    last_id = ''
    while True:
        # 拉取当前批次数据
        query = """
            SELECT id, req_data
            FROM your_table
            WHERE id > %s
            ORDER BY id
            LIMIT %s
        """
        cursor.execute(query, (last_id, batch_size))
        rows = cursor.fetchall()
        if not rows:
            break
        
        # 处理每一行数据
        for row in rows:
            source_type, error_code = parse_xml(row['req_data'])
            if source_type and error_code:
                stats[source_type][error_code] += 1
        
        last_id = rows[-1]['id']
        print(f"已处理 {len(rows)} 条数据，当前进度id: {last_id}")
    
    cursor.close()
    conn.close()

def print_result_table():
    """生成指定格式的结果表格"""
    print("| 数据源 | 错误码 | 数量 |")
    print("|--------|--------|------|")
    for source in ['Source A', 'Source B']:
        for error_code, count in stats[source].items():
            print(f"| {source} | {error_code} | {count} |")

if __name__ == '__main__':
    batch_process()
    print_result_table()

方案优势 #

• lxml解析XML比MySQL的EXTRACTVALUE更精准，支持复杂XML结构，避免解析偏差
• 本地聚合统计，大幅降低数据库端计算压力
• 分批次拉取数据，既规避单次查询超时，也不会占用过多内存

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