单方法包裹的百万级数据处理程序如何实现多进程优化

兄弟，我看你把所有业务逻辑全塞进MSP()里了，嵌套循环跑超百万条数据还加大量字符串匹配，慢是肯定的——这种CPU密集型任务天生适合用多进程拆开来跑。之前试Stack Overflow的例子没成功，大概率是没把核心并行逻辑从单方法里抽出来，全堆一块根本没法让多进程发挥作用。我给你一步步改：

第一步：把核心匹配逻辑拆成独立函数 #

首先得把嵌套循环里的核心工作——用单个bat元素去匹配所有dg行的逻辑，抽成一个单独的函数。这样每个进程可以单独处理一个bat元素，互相不干扰：

def process_bat_element(d, dg_data):
    import re
    # 解包传入的dg数据：pla、nam、cat对应原代码里的三个列表
    pla, nam, cat = dg_data
    matched_rows = []
    for j in range(len(nam)):
        e = str(nam[j])
        substring = str(d)
        # 用原始字符串写正则，避免转义坑；加re.escape处理特殊字符
        pattern = re.compile(r'\b' + re.escape(substring) + r'\b', flags=re.IGNORECASE)
        if pattern.search(e):
            # 把匹配到的行数据存起来，后续统一写入
            matched_rows.append([pla[j], nam[j], cat[j]])
    return matched_rows

第二步：改造MSP()，引入多进程池 #

原来的嵌套循环要换成multiprocessing.Pool来并行处理bat列表里的每个元素。注意要提前把dg的数据打包好传给子进程，还要统一处理结果和文件写入：

def MSP():
    import multiprocessing
    import pandas as pd
    from pandas.io.excel import ExcelFile
    import csv
    from pathlib import Path

    # 确保输出目录存在
    Path("Output").mkdir(exist_ok=True)

    # 读取第一个Excel文件，提取bat列表
    xlsx = ExcelFile('XX.xlsx')
    df = xlsx.parse(xlsx.sheet_names[0])
    df2 = pd.DataFrame(df, columns=['XX'])
    bat = df2['XX'].values.tolist()

    # 读取第二个CSV文件，打包需要的数据传给子进程
    Ofilename = input('Enter a filename: ')
    dg = pd.read_csv(Ofilename, usecols=['XX','XXX','XXXX'], encoding="ISO-8859-1")
    dg_data = (
        dg['XX'].values.tolist(),
        dg['XXX'].values.tolist(),
        dg['XXXX'].values.tolist()
    )

    print("\nJob started, processing data...")

    # 创建进程池，用CPU核心数-1，避免占满系统资源
    with multiprocessing.Pool(processes=multiprocessing.cpu_count()-1) as pool:
        # 用starmap传递多个参数：每个bat元素 + 打包好的dg数据
        results = pool.starmap(process_bat_element, [(d, dg_data) for d in bat])

    # 收集所有进程返回的匹配结果
    all_matched_rows = []
    for result in results:
        all_matched_rows.extend(result)

    # 统一写入CSV，避免多进程写文件冲突
    filename = "Output/XX words.csv"
    with open(filename, 'w', encoding="ISO-8859-1", newline='') as csvfile:
        csvwriter = csv.writer(csvfile)
        # 写入表头
        csvwriter.writerow(['XX','XXX','XXXX'])
        # 批量写入所有匹配行
        csvwriter.writerows(all_matched_rows)

    z = len(all_matched_rows)
    print(f"The total words XX is {z}")

几个关键注意事项 #

• 数据传递效率：别让每个子进程都重新读一遍CSV文件，提前把需要的列表打包好传给子进程，减少IO开销。
• 文件写入安全：绝对不要让多个进程同时写同一个文件，很容易出现数据丢失或乱序。统一收集结果后由主进程写入，既安全又高效。
• 正则优化：用re.compile预编译正则表达式，再加re.escape处理substring里的特殊字符（比如.、*），避免正则匹配出错。
• 进程池管理：用with语句管理Pool，会自动帮你完成进程的关闭和等待，不用手动写close()和join()，代码更简洁可靠。

最后，主函数里的调用保持简单：

if __name__ == '__main__':
    MSP()

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