如何用Python多进程加速从文本文件提取PMID对应数据？

你的单进程代码因逐个请求NCBI API，处理大量PMID时速度受限。利用多进程并行发起API请求可显著提升效率，以下是优化方案：

核心思路 #

将单个PMID的文献信息提取逻辑封装为独立函数，通过进程池并行处理所有PMID，最后统一收集结果并保存为CSV。这种方式能同时发起多个API请求，避开单进程串行等待的网络瓶颈。

优化后的代码 #

import pandas as pd
import os
from concurrent.futures import ProcessPoolExecutor, as_completed

# 设置NCBI API密钥（建议填写，提升请求限额）
os.environ['NCBI_API_KEY'] = ""
import metapub

def fetch_pmid_data(pmid):
    """单个PMID的文献信息提取函数"""
    fetch = metapub.PubMedFetcher()
    try:
        article = fetch.article_by_pmid(pmid)
        # 整理单条数据
        return {
            "PMID": pmid,
            "PMC": article.pmc,
            "DOI": article.doi,
            "Year": article.history['pubmed'].strftime('%d-%m-%Y') if 'pubmed' in article.history else None,
            "Title": article.title,
            "Authors": article.authors,
            "Journal": article.journal,
            "Abstract": article.abstract,
            "Keywords": article.keywords,
            "Chemicals": list(map(lambda v: v["substance_name"], article.chemicals.values())) if article.chemicals else [],
            "Mesh_Descriptors": list(map(lambda v: v["descriptor_name"], article.mesh.values())) if article.mesh else [],
            "Mesh_Qualifiers": list(map(lambda v: v["qualifier_name"], article.mesh.values())) if article.mesh else [],
            "URL": article.url
        }
    except Exception as e:
        # 可选：取消注释打印错误信息便于排查
        # print(f"Error processing PMID {pmid}: {str(e)}")
        return {"PMID": pmid}  # 返回仅含PMID的字典，标记未成功获取

if __name__ == "__main__":
    # 读取PMID列表
    with open("pmid.txt", "r") as file:
        pmid_list = file.read().splitlines()

    # 初始化进程池，建议根据CPU核心数设置max_workers，NCBI有请求限额，别设太大（比如4-8）
    results = []
    with ProcessPoolExecutor(max_workers=6) as executor:
        # 提交所有任务
        futures = {executor.submit(fetch_pmid_data, pmid): pmid for pmid in pmid_list}
        
        # 逐个获取完成的任务结果
        for idx, future in enumerate(as_completed(futures), 1):
            results.append(future.result())
            # 每处理1000条保存一次进度
            if idx % 1000 == 0:
                print(f"Processed {idx} records")
                pd.DataFrame(results).to_csv("myfile_progress.csv", index=False)

    # 合并所有结果并保存最终文件
    final_data = pd.DataFrame(results)
    final_data.to_csv("myfile_final.csv", index=False)
    print('Finish')

关键说明 #

• 进程池设置：max_workers不要设置过大，NCBI对API请求有频率限制（未填API密钥时限额更低），建议设为4-8，避免触发封禁或请求失败。
• 任务函数独立：每个进程会独立初始化PubMedFetcher，避免多进程间的资源冲突。
• 异常处理：保留异常捕获，确保单个PMID处理失败不影响整体流程，失败的记录仅保留PMID，便于后续排查。
• 进度保存：每处理1000条保存一次进度文件，防止程序中断丢失数据。
• API密钥：务必填写NCBI API密钥，能大幅提升请求限额，进一步加快提取速度。

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