Python：读取和写入巨大的 Json 文件

Question

我是 python 新手。所以如果我不是用pythonic方式问问题，请原谅。

我的要求如下：

1. 我需要编写python代码来实现这个需求。

2. 将读取 60 个 json 文件作为输入。每个文件大约 150 GB。

3. 所有 60 个 json 文件的示例结构如下所示。请注意，每个文件只有一个 json 对象。而每个文件的巨大大小是因为那个巨大的json对象中包含的“array_element”数组的数量和大小。

   { "string_1":"abc", "string_1":"abc", "string_1":"abc", "string_1":"abc", "string_1":"abc", "string_1":"abc", “数组元素”：[] }

4. 转换逻辑很简单。我需要合并所有 60 个文件中的所有 array_element 并将其写入一个巨大的 json 文件。这几乎是 150GB X 60 将是输出 json 文件的大小。

我请求您帮助解决的问题：

1. 阅读：计划使用“ijson”模块的 ijson.items(file_object, "array_element")。您能否告诉我 ijson.items 是否会从 json 文件中的“array_element”数组中一次“生成”（即不将整个文件加载到内存中）一个项目？我不认为 json.load 在这里是一个选项，因为我们无法在内存中保存如此庞大的字典。

2. 对于写作：我打算使用 ijson.item 读取每个项目，并执行 json.dumps 以“编码”，然后使用 file_object.write 将其写入文件而不使用 json.dump，因为我不能拥有这样的内存中的巨大字典使用 json.dump。您能否告诉我是否需要在下面显示的代码中应用 f.flush() ？据我了解，内部缓冲区在满时会自动刷新，内部缓冲区的大小是恒定的，不会动态增长到内存超载的程度？请告诉我

3. 对于增量读取和写入巨大的 json 文件，是否有比上述方法更好的方法？

代码 sn-p 显示上述读写逻辑：

for input_file in input_files:
    with open("input_file.json", "r") as f:
         objects = ijson.items(f, "array_element")
         for item in objects:
              str = json.dumps(item, indent=2)
              with open("output.json", "a") as f:
                   f.write(str)
                   f.write(",\n")
                   f.flush()
    with open("output.json", "a") as f:
        f.seek(0,2)
        f.truncate(f.tell() - 1)
        f.write("]\n}")

希望我已经清楚地提出了我的问题。提前致谢！！

Answer 1

以下程序假定输入文件具有足够可预测的格式以跳过 JSON 解析以提高性能。

根据您的描述，我的假设是：

• 所有文件都具有相同的编码。
• 所有文件在开头的某个位置都有一个可以找到"array_element":[ 的位置，之后文件的“有趣部分”开始
• 所有文件的末尾都有一个位置，]} 标志着“有趣部分”的结束
• 所有“有趣的部分”都可以用逗号连接，并且仍然是有效的 JSON

当所有这些点都成立时，连接预定义的标题片段、相应的文件范围和页脚片段将生成一个大的、有效的 JSON 文件。

import re
import mmap

head_pattern = re.compile(br'"array_element"\s*:\s*\[\s*', re.S)
tail_pattern = re.compile(br'\s*\]\s*\}\s*$', re.S)

input_files = ['sample1.json', 'sample2.json']

with open('result.json', "wb") as result:
    head_bytes = 500
    tail_bytes = 50
    chunk_bytes = 16 * 1024

    result.write(b'{"JSON": "fragment", "array_element": [\n')

    for input_file in input_files:
        print(input_file)

        with open(input_file, "r+b") as f:
            mm = mmap.mmap(f.fileno(), 0)
            
            start = head_pattern.search(mm[:head_bytes])
            end = tail_pattern.search(mm[-tail_bytes:])

            if not (start and end):
                print('unexpected file format')
                break

            start_pos = start.span()[1]
            end_pos = mm.size() - end.span()[1] + end.span()[0]

            if input_files.index(input_file) > 0:
                result.write(b',\n')

            pos = start_pos
            mm.seek(pos)
            while True:
                if pos + chunk_bytes >= end_pos:
                    result.write(mm.read(end_pos - pos))
                    break
                else:
                    result.write(mm.read(chunk_bytes))
                    pos += chunk_bytes

    result.write(b']\n}')

如果文件格式是 100% 可预测的，您可以丢弃正则表达式并使用 mm[:head_bytes].index(b'...') 等进行开始/结束位置算术。