如何在爬取JSON并导出CSV时结合multiprocessing Pool优化效率？

Got it, let's work through this together. Your original code pulls the full JSON dataset, but processing large numbers of entries can drag. We can use multiprocessing.Pool to split the data transformation work across multiple processes, cutting down on total time. Here's the integrated solution, broken down step by step:

完整代码实现 #

import requests
import json
import csv
from multiprocessing import Pool

# 定义单条数据处理函数（每个进程负责处理一个条目）
def process_crypto(item):
    # 按需筛选或转换字段，这里示例提取常用信息，你可以根据需求调整
    return {
        "id": item.get("id"),
        "name": item.get("name"),
        "symbol": item.get("symbol"),
        "slug": item.get("slug"),
        "rank": item.get("rank"),
        "is_active": item.get("is_active")
    }

def main():
    # 1. 抓取原始JSON数据
    url = "https://s2.coinmarketcap.com/generated/search/quick_search.json"
    r = requests.get(url)
    contents = json.loads(r.content)
    
    # 2. 多进程并行处理数据
    # 进程数建议设为CPU核心数（可通过multiprocessing.cpu_count()自动获取）
    with Pool(processes=4) as pool:
        processed_results = pool.map(process_crypto, contents)
    
    # 3. 将处理后的数据写入CSV
    if processed_results:
        fieldnames = processed_results[0].keys()
        with open("crypto_data.csv", "w", newline="", encoding="utf-8") as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(processed_results)
        print("CSV文件已成功生成！")

if __name__ == "__main__":
    main()

核心部分说明 #

• process_crypto函数：这是每个进程执行的独立任务，负责单条加密货币数据的清洗和转换。你可以在这里添加字段校验、格式修正等逻辑，比如处理空值、转换数据类型。
• Pool的使用：Pool(processes=4)指定用4个进程并行处理，你可以根据自己的CPU核心数调整（比如用multiprocessing.cpu_count()自动匹配硬件）。pool.map()会自动把contents中的条目分配给不同进程，最后汇总所有处理结果。
• CSV写入：用csv.DictWriter可以直接把字典列表写入CSV，自动匹配表头和数据列，避免手动拼接字符串的麻烦。

注意要点 #

1. 如果是网络请求本身慢，多进程帮不上忙——这时候要考虑异步请求（比如aiohttp）。但你的问题是处理数据耗时，所以多进程是合适的方案。
2. 不要设置过多进程数，超过CPU核心数会导致进程切换开销增大，反而降低效率。
3. 确保process_crypto函数是进程安全的，不要在里面使用未加锁的全局变量。

内容的提问来源于stack exchange，提问作者j.doe