如何优化使用 json 转储将字典列表写入文件？

Question

目前我有以下实现将list 的dictionaries 写入文件。下面的file_limit_counter 将被初始化为0，而file_limit 现在将被初始化为50。因此，每当计数器等于文件限制时，它就会开始将输出写入新文件。计数器的方法是将输出分成多个文件：

self.file = self.generate_new_micro_file_name()
for response_dict in all_response_dict_list:
    if file_limit_counter == file_limit:
        self.propagate_log_msg('wrote {} records '.format(file_limit))
        self.file = self.generate_new_micro_file_name()
        file_limit_counter = 0
    with open(self.file, 'a', encoding="utf8") as open_out_file:
        json.dump(response_dict, open_out_file)
        open_out_file.write('\n')
    file_limit_counter += 1

列表all_response_dict_list 将包含如下内容：

somelist = [{"Name":"a1","Age":"24"},{"Name":"a2","Age":"26"}]

我的意图是在我的输出文件中包含这样的内容：

{"Name":"a1","Age":"24"}
{"Name":"a2","Age":"26"}
...

上述方法效果很好。但是，当涉及到大量词典（例如 5000）时，它往往会放慢一点（大约需要 10 分钟）。因此，如果有人已经遇到过这种情况，那将会很有帮助。我认为，如果我们可以并行执行上述相同的操作，即同时将多个字典写入同一个文件而不是一个一个地写入，我认为会花费更少的时间。

Answer 1

这通过保持文件打开而不是打开/关闭 50 次，在 1.1 秒内为 1,000 个文件处理 50,000 个名称。

import json
import time

# Generate.

part1 = time.time()

all_response_dict_list = []
for i in range(50000):
    all_response_dict_list.append( {'name':f'a{i}', 'age':str(i) } )

part2 = time.time()
print( part2 - part1 )

# Write.

file_limit = 50
file_limit_counter = 0
fnum = 0

def generate_new_micro_file_name():
    global fnum
    fnum += 1
    return f'base{fnum}.json'

open_out_file = open(generate_new_micro_file_name(), 'a', encoding='utf8')
for response_dict in all_response_dict_list:
    if file_limit_counter == file_limit:
        open_out_file = open(generate_new_micro_file_name(), 'a', encoding='utf8')
        file_limit_counter = 0
    json.dump(response_dict, open_out_file)
    open_out_file.write('\n')
    file_limit_counter += 1

print( time.time() - part2 )

Answer 2

您可以将列表拆分为 file_limit 大小的块，然后立即写入文件。

试试：

chunks = [all_response_dict_list[i: i+file_limit] for i in range(0, len(all_response_dict_list), file_limit)]

for chunk in chunks:
    with open(self.generate_new_micro_file_name()) as outfile:
        json.dump(chunk, outfile)

Answer 3

在我的机器上运行

import json
import time


all_response_dict_list = [{'name': f'a{i}', 'age': i} for i in range(50_000)]
file_limit = 50


def fgen():
    fnum = 1
    while True:
        yield f'/Users/andy/j/base{fnum}.json'
        fnum += 1


def main():
    G = fgen()
    for offset in range(0, len(all_response_dict_list), file_limit):
        with open(next(G), 'a') as outfile:
            for rd in all_response_dict_list[offset:offset+file_limit]:
                json.dump(rd, outfile)
                outfile.write('\n')


if __name__ == '__main__':
    start = time.perf_counter()
    main()
    end = time.perf_counter()
    print(f'Duration={end-start:.4f}s')