批量读取250万+HTML文件并写入PostgreSQL的提速方案咨询

一、数据库操作：从单条更新转批量处理 #

当前单条UPDATE的方式会产生大量数据库交互，是性能瓶颈核心，优化方向：

• 用COPY命令批量导入：先把提取的数据写入临时CSV文件，再通过PostgreSQL的COPY命令直接导入，这是数据库批量写入效率最高的方式，比单条SQL快几十倍。
• 批量UPDATE语句：一次处理1000-5000条数据，用FROM VALUES构造批量更新语句，减少连接交互次数。同时必须用参数化查询，避免SQL注入，还能让数据库缓存执行计划。示例代码：

from psycopg2.extras import execute_values

def batch_update(conn, batch_data):
    # batch_data格式：[(id, order_status, tracking_id), ...]
    query = """
        UPDATE order_additional_info
        SET order_status = data.order_status, tracking_status = data.tracking_id
        FROM (VALUES %s) AS data(id, order_status, tracking_id)
        WHERE order_additional_info.id = data.id;
    """
    with conn.cursor() as cur:
        execute_values(cur, query, batch_data)
    conn.commit()

• 用连接池复用连接：用psycopg2.pool.SimpleConnectionPool创建连接池，避免每次操作都新建/关闭连接，节省连接建立的开销。

二、文件处理：并行化提升效率 #

单线程遍历文件会浪费CPU资源，因为读文件是IO密集型操作，和数据提取的CPU操作可以并行：

• 用多进程/多线程并行提取：如果是正则提取（CPU密集），用concurrent.futures.ProcessPoolExecutor，进程数设为CPU核心数的2-4倍；如果是HTML解析（IO+CPU），用ThreadPoolExecutor，线程数控制在50-100之间，避免磁盘IO过载。
• 批量提交数据：每收集1000条提取结果后，再批量写入数据库，减少锁竞争和数据库交互次数。
• 预扫描存在的文件：先遍历html/目录，生成实际存在的id列表，跳过不存在的文件，避免每次判断文件存在性的开销。

三、数据提取：优化正则与文件读取 #

• 预编译正则表达式：把正则对象定义在函数外部，不要每次提取都重新编译，减少重复计算：

import re

# 预编译正则，放在函数外
ORDER_STATUS_REG = re.compile(r'匹配订单状态的正则')
TRACKING_ID_REG = re.compile(r'匹配追踪号的正则')

def extract_html_information(file_path):
    with open(file_path, 'r', buffering=8192) as f:
        content = f.read()
        order_status = ORDER_STATUS_REG.search(content).group(1)
        tracking_id = TRACKING_ID_REG.search(content).group(1)
    return {'order_status': order_status, 'tracking_id': tracking_id}

• 增大文件读取缓冲区：打开文件时设置buffering=8192（或更大），减少磁盘IO的系统调用次数。
• 改用HTML解析库：如果HTML结构规范，用BeautifulSoup配合lxml解析器，比正则更稳定，复杂页面下解析效率也更高。

四、数据库配置调优 #

适当调整PostgreSQL的配置参数，提升批量操作性能：

• 增大work_mem：给排序、哈希操作分配更多内存，避免磁盘临时文件。
• 增大maintenance_work_mem：提升批量数据导入时的性能。
• 适当调高max_connections：配合并行处理，确保有足够的连接数，但不要超过服务器承载上限。

五、进度记录与容错 #

• 每处理10000条数据就记录一次当前id到日志文件，程序崩溃后可以从断点继续处理，不用从头开始。
• 捕获文件读取、数据提取、数据库操作的异常，记录错误id和原因，后续单独处理异常文件。

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