拆分大型json文件，并将每个部分分配给一个进程

Question

我正在尝试提炼一个包含多个进程的大型JSON数据集，大约14 GiB (1，200万行)，这样我就可以更快了。我创建了两个multiprocessing.Queue实例：in_q和out_q。他们会读取data-2021-09-29.jsonl文件(这是数据集)中的数据，并输出包含我感兴趣的数据的行，并将其写入另一个文件stokes_DE_file.jsonl (这是我提炼数据集的部分，输出文件是数据集的提炼版本)。我的机器上有16个CPU，假设我想将文件拆分为16个部分，行数将是变量LINES_PER_PROCESS。如何为每个进程分配文件的一部分？以下是我到目前为止编写的代码。从python中的多处理模块开始。下面是我的代码：

import multiprocessing as mp
import threading
import json
import reverse_geocoder as rg

LINES_PER_PROCESS = 12137928/(mp.cpu_count()-1)

def geocode_worker(in_q, out_q):
    while True:
        strike = in_q.get()
        if strike is None:
            out_q.put(None)
            return
        strike_location = (strike['lat'], strike['lon'])
        if rg.search(strike_location)[0]['cc'] == 'DE':
            out_q.put('{}\n'.format(strike))

def file_write_worker(out_q, fileobj, worker_count):
    while worker_count:
        for msg in iter(out_q.get, None):
            
            if msg is None:
                worker_count -= 1
            fileobj.write(msg)

def get_germany_strokes(jsonl_file):
    
    worker_count = mp.cpu_count() - 1
    in_q, out_q = mp.Queue(), mp.Queue()
    processes = [mp.Process(target=geocode_worker, args=(in_q, out_q)) for _ in range(worker_count)]
    
    for p in processes:
        p.start()
        
    with open('strokes_DE_file.json', newline='') as strokes_DE_file:
        file_writer = threading.Thread(target=file_write_worker, args=(out_q, strokes_DE_file, worker_count))
        file_writer.start()
        
    with open(jsonl_file, newline='') as file:
        next(file)
        for line in file:
            strokes = json.loads(line)['strokes']
            for strike in strokes:
                in_q.put(strike)
                    
get_germany_strokes('data-2021-09-29.jsonl')

Answer 1

我找到的解决方案是用Linux中的split -l命令拆分我的数据集，并指定每个子集的行数。我将每个子集分配给一个进程，总共15个进程大约是要创建的最佳进程数(考虑到开销/负载平衡)。