我正在尝试处理一个文件(每行都是一个json文档)。该文件的大小可以达到数百mb到gb级别。因此,我编写了一个生成器代码,逐行从文件中获取每个文档。
def jl_file_iterator(file):
with codecs.open(file, 'r', 'utf-8') as f:
for line in f:
document = json.loads(line)
yield document
我的系统有4个核心,因此我想并行处理文件中的4行。目前我有这段代码,它一次取4行,并调用并行处理的代码。
threads = 4
files, i = [], 1
for jl in jl_file_iterator(input_path):
files.append(jl)
if i % (threads) == 0:
# pool.map(processFile, files)
parallelProcess(files, o)
files = []
i += 1
if files:
parallelProcess(files, o)
files = []
这是我的代码,实际的处理发生在这里。
def parallelProcess(files, outfile):
processes = []
for i in range(len(files)):
p = Process(target=processFile, args=(files[i],))
processes.append(p)
p.start()
for i in range(len(files)):
processes[i].join()
def processFile(doc):
extractors = {}
... do some processing on doc
o.write(json.dumps(doc) + '\n')
你可以看到,在我发送下一批4个文件进行处理之前,我会等待所有4个行的处理完成。但我想要的是,只要一个进程处理完文件,我就想开始分配下一个可用进程来处理下一行。我该怎么做?
PS:问题在于它是一个生成器,我不能加载所有文件并使用类似于map的东西运行进程。
谢谢你的帮助