任务中Parallel processing失效问题排查（附代码）

问题背景 #

我用pqdm包做并行处理并展示进度，设置n_jobs=60后CPU利用率达90%，但实际并行处理并未生效；改用单核心版本代码时，运行缓慢但能稳定执行。

问题根源 #

1. SQLite锁冲突：每个进程频繁创建、关闭SQLite连接，而SQLite默认采用单文件锁机制，多进程并发读写会触发大量锁等待。看似CPU占用率高，实则大部分时间都在等待锁释放，并行效率极低。
2. 进程间数据传输开销过大：将DataFrame转为字典列表传入pqdm，大量数据在进程间传输会产生额外开销，直接抵消了并行处理的优势。
3. 进程数设置不合理：60个进程远超常规CPU核心数（一般服务器核心数在16-32之间），频繁的进程切换会大幅增加系统开销。

修复方案 #

1. 预加载城市数据到内存：程序启动时将SQLite中的城市数据一次性加载到全局变量，所有进程直接使用内存数据，避免重复建立数据库连接。
2. 优化进程数量：使用multiprocessing.cpu_count()获取当前CPU核心数，以此设置n_jobs，减少进程切换开销。
3. 精简进程间传输数据：仅传递必要的字段，避免整行数据的冗余传输。

修改后的完整代码 #

import pandas as pd
import sqlite3
from fuzzywuzzy import fuzz
import time
import os
import multiprocessing
from pqdm.processes import pqdm
import requests
import zipfile
import io

# 全局变量存储预加载的城市数据，供所有进程共享使用
EUROPEAN_CITIES = {}

def download_and_process_geonames():
    # 下载人口大于1000的GeoNames城市数据
    url = "http://download.geonames.org/export/dump/cities1000.zip"
    try:
        response = requests.get(url)
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        print(f"下载文件出错: {e}")
        return

    try:
        z = zipfile.ZipFile(io.BytesIO(response.content))
    except zipfile.BadZipFile:
        print("下载的文件不是有效的zip文件。")
        return

    # 解压文件
    try:
        z.extractall()
    except Exception as e:
        print(f"解压zip文件出错: {e}")
        return

    # 检查文件是否存在
    if not os.path.exists('cities1000.txt'):
        print("解压后的文件'cities1000.txt'不存在。")
        return

    # 读取解压后的文件
    try:
        df = pd.read_csv('cities1000.txt', sep='\t', header=None, 
                         names=['geonameid', 'name', 'asciiname', 'alternatenames', 'latitude', 'longitude', 
                                'feature_class', 'feature_code', 'country_code', 'cc2', 'admin1_code', 
                                'admin2_code', 'admin3_code', 'admin4_code', 'population', 'elevation', 
                                'dem', 'timezone', 'modification_date'])
    except Exception as e:
        print(f"读取CSV文件出错: {e}")
        return

    # 筛选欧洲国家（可根据需要更新列表）
    european_countries = ['AD', 'AL', 'AT', 'BA', 'BE', 'BG', 'BY', 'CH', 'CY', 'CZ', 'DE', 'DK', 'EE', 'ES', 'FI', 'FR', 'GB', 'GR', 'HR', 'HU', 'IE', 'IS', 'IT', 'LI', 'LT', 'LU', 'LV', 'MC', 'MD', 'ME', 'MK', 'MT', 'NL', 'NO', 'PL', 'PT', 'RO', 'RS', 'RU', 'SE', 'SI', 'SK', 'SM', 'UA', 'VA', 'XK']
    european_cities = df[df['country_code'].isin(european_countries)]

    # 创建SQLite数据库
    try:
        conn = sqlite3.connect('europe_cities.db')
        
        # 写入数据库
        european_cities[['name', 'country_code', 'latitude', 'longitude']].to_sql('cities', conn, if_exists='replace', index=False)
        
        conn.close()
    except Exception as e:
        print(f"创建或写入数据库出错: {e}")
        return

    # 清理临时文件
    try:
        os.remove('cities1000.txt')
    except Exception as e:
        print(f"删除临时文件出错: {e}")

    print("数据库创建成功！")

def preload_cities(db_path):
    """预加载城市数据到全局变量，避免多进程重复连接数据库"""
    global EUROPEAN_CITIES
    try:
        with sqlite3.connect(db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT name, country_code, latitude, longitude FROM cities")
            cities = cursor.fetchall()
        
        # 按国家代码分组，减少后续匹配时的遍历范围
        for name, cc, lat, lon in cities:
            if cc not in EUROPEAN_CITIES:
                EUROPEAN_CITIES[cc] = []
            EUROPEAN_CITIES[cc].append((name, lat, lon))
    except Exception as e:
        print(f"预加载城市数据出错: {str(e)}")

def get_coordinates(city_name, country_code):
    if pd.isna(city_name) or pd.isna(country_code) or city_name is None or country_code is None:
        return None, None, None, "缺少或无效的城市名称/国家代码"
    
    # 从全局变量获取对应国家的城市列表
    cities = EUROPEAN_CITIES.get(country_code)
    if not cities:
        return None, None, None, f"未找到国家代码为{country_code}的城市"
    
    best_match = max(cities, key=lambda x: fuzz.ratio(str(city_name).lower(), str(x[0]).lower()))
    match_score = fuzz.ratio(str(city_name).lower(), str(best_match[0]).lower())
    
    if match_score < 50:  # 可调整匹配阈值
        return None, None, None, f"未找到与{city_name}匹配度较高的城市"
    
    return best_match[1], best_match[2], best_match[0], None

def process_inventor(row):
    lat, lon, matched_city, reason = get_coordinates(row['inventor_city_name'], row['inventor_country_code'])
    return {**row, 'latitude': lat, 'longitude': lon, 
            'matched_city': matched_city, 'matched_country_code': row['inventor_country_code'], 
            'geocoding_reason': reason}

def process_inventors(df, checkpoint_file):
    # 加载检查点（如果存在）
    if os.path.exists(checkpoint_file):
        processed_df = pd.read_csv(checkpoint_file)
        processed_ids = set(processed_df['inventor_id'])
        df = df[~df['inventor_id'].isin(processed_ids)]
    else:
        processed_df = pd.DataFrame()

    # 使用CPU核心数设置进程数
    n_jobs = multiprocessing.cpu_count()
    print(f"使用{n_jobs}个进程进行并行处理")

    # 转换为字典列表（仅传递必要字段，减少数据传输）
    rows = df.to_dict('records')
    
    # 使用pqdm并行处理
    results = pqdm(rows, process_inventor, n_jobs=n_jobs, desc="处理发明人数据")

    # 合并结果
    new_df = pd.DataFrame(results)
    processed_df = pd.concat([processed_df, new_df], ignore_index=True)

    return processed_df

if __name__ == '__main__':
    start_time = time.time()
    
    # 初始化数据库
    download_and_process_geonames()
    
    # 预加载城市数据到内存
    db_path = 'europe_cities.db'
    preload_cities(db_path)
    
    checkpoint_file = 'geocoding_checkpoint.csv'
    
    # 加载你的发明人数据（取消注释并修改路径）
    # unique_inventors = pd.read_csv('your_input_file.csv')
    
    # 处理数据
    unique_inventors_processed = process_inventors(unique_inventors, checkpoint_file)
    
    # 保存最终结果
    unique_inventors_processed.to_csv('unique_inventors_processed_final.csv', index=False)
    
    end_time = time.time()
    processing_time = end_time - start_time
    
    print(f"处理完成，耗时{processing_time:.2f}秒。")
    print(f"总处理记录数: {len(unique_inventors_processed)}")
    print(f"成功获取坐标的记录数: {len(unique_inventors_processed[unique_inventors_processed['latitude'].notna()])}")
    print(f"未获取坐标的记录数: {len(unique_inventors_processed[unique_inventors_processed['latitude'].isna()])}")
    print("处理后的数据已保存到'unique_inventors_processed_final.csv'。")

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