MySQL：拆分对应错误码至A/B类型列（超5000万数据）

需求说明 #

原表结构及数据：

其中error_type的每个A/B与error_code的每个数值一一对应，需转换为以下结构：

约束条件：表中包含其他列，数据量超5000万条，MySQL查询超时时间30秒，避免使用游标/存储过程（性能不足）。

方案一：纯MySQL高效查询（推荐） #

利用字符串处理函数+数字辅助表，避免游标和复杂循环，适合大表快速处理。

步骤1：创建数字辅助表 #

用于拆分空格分隔的字符串，数字范围覆盖error_type的最大元素个数（示例创建1-100，可按需扩展）：

CREATE TABLE IF NOT EXISTS nums (n INT PRIMARY KEY);
-- 插入1-100的数字
INSERT INTO nums(n)
WITH RECURSIVE cte AS (
    SELECT 1 AS n
    UNION ALL
    SELECT n+1 FROM cte WHERE n < 100
)
SELECT n FROM cte;

步骤2：转换查询 #

通过拆分字符串、匹配对应关系后聚合，保留原表其他列：

SELECT 
    t.id,
    t.type,
    -- 拼接A对应的error_code，空值自动忽略
    GROUP_CONCAT(CASE WHEN et = 'A' THEN ec END SEPARATOR ' ') AS error_code_A,
    GROUP_CONCAT(CASE WHEN et = 'B' THEN ec END SEPARATOR ' ') AS error_code_B,
    -- 保留原表其他列，按需添加
    t.other_column1,
    t.other_column2
FROM (
    SELECT 
        main.id,
        main.type,
        main.other_column1,
        main.other_column2,
        -- 拆分error_type的单个元素
        SUBSTRING_INDEX(SUBSTRING_INDEX(main.error_type, ' ', nums.n), ' ', -1) AS et,
        -- 拆分对应位置的error_code元素
        SUBSTRING_INDEX(SUBSTRING_INDEX(main.error_code, ' ', nums.n), ' ', -1) AS ec
    FROM your_table main
    JOIN nums 
        ON nums.n <= LENGTH(main.error_type) - LENGTH(REPLACE(main.error_type, ' ', '')) + 1
) t
GROUP BY t.id, t.type, t.other_column1, t.other_column2;

性能优化说明

• 数字辅助表仅需创建一次，后续可复用
• 拆分逻辑基于字符串函数，比游标/存储过程更高效
• 聚合操作在MySQL引擎内完成，避免数据导出导入的开销

方案二：Python批量处理（备选） #

若MySQL查询仍超时，可通过Python分批次读取处理，避免单条查询压力：

代码示例 #

import mysql.connector
from mysql.connector import pooling

# 初始化数据库连接池（按需修改配置）
db_pool = mysql.connector.pooling.MySQLConnectionPool(
    pool_name="data_pool",
    pool_size=8,
    host="your_host",
    user="your_username",
    password="your_password",
    database="your_database",
    charset="utf8mb4"
)

def process_single_row(row):
    """处理单条数据，拆分对应关系"""
    id_val, type_val, error_type, error_code, *other_cols = row
    et_list = error_type.split()
    ec_list = error_code.split()
    
    a_codes = []
    b_codes = []
    for et, ec in zip(et_list, ec_list):
        if et == 'A':
            a_codes.append(ec)
        elif et == 'B':
            b_codes.append(ec)
    
    return (id_val, type_val, ' '.join(a_codes), ' '.join(b_codes), *other_cols)

def batch_process():
    conn = db_pool.get_connection()
    cursor = conn.cursor()
    
    # 分批次读取数据，每次10000条（可根据内存调整）
    cursor.execute("SELECT id, type, error_type, error_code, other_column1, other_column2 FROM your_table")
    batch_size = 10000
    
    while True:
        batch = cursor.fetchmany(batch_size)
        if not batch:
            break
        
        # 处理当前批次
        processed_data = [process_single_row(row) for row in batch]
        
        # 写入目标表（需提前创建目标表结构）
        insert_sql = """
            INSERT INTO target_table 
            (id, type, error_code_A, error_code_B, other_column1, other_column2)
            VALUES (%s, %s, %s, %s, %s, %s)
        """
        cursor.executemany(insert_sql, processed_data)
        conn.commit()
    
    cursor.close()
    conn.close()

if __name__ == "__main__":
    batch_process()

注意事项

• 目标表需提前创建，结构与需求一致
• 调整batch_size以平衡内存占用和处理速度
• 使用连接池避免频繁创建数据库连接的开销

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