Python读取3GB大JSON文件遇][垃圾值解析报错如何解决？

核心思路 #

不需要将整个文件加载到内存，采用流式分块读取的方式处理目标垃圾串']['，同时处理分块边界可能出现的串截断问题，保证匹配替换的完整性。

方案1：先预处理生成合法JSON文件，再解析 #

该方案会生成修复后的新JSON文件，适合后续需要多次读取该文件的场景，全程内存占用仅和你设置的块大小挂钩，3GB文件处理仅需几十MB内存。
处理步骤：

• 定义目标匹配串为']['，长度为3，因此每次读取新块时，需要拼接上一次读取块的最后2个字符，避免匹配串被拆分在两个块中无法识别
• 逐块替换匹配串为,，写入新文件
• 最后处理文件首尾的多余括号（如果需要合并为单个数组的话）

代码示例：

def fix_json_file(source_path, target_path, chunk_size=16*1024):
    # 要匹配的垃圾串
    garbage = "]["
    garbage_len = len(garbage)
    # 保留上一个块的尾部字符，长度为垃圾串长度-1
    tail = ""
    with open(source_path, "r", encoding="utf-8") as src, open(target_path, "w", encoding="utf-8") as tgt:
        while True:
            chunk = src.read(chunk_size)
            if not chunk:
                # 读取结束，写入剩余的尾部
                tgt.write(tail)
                break
            # 拼接上一次的尾部和当前块
            full_chunk = tail + chunk
            # 替换垃圾串
            processed = full_chunk.replace(garbage, ",")
            # 新的尾部是处理后内容的最后 garbage_len - 1 个字符
            tail = processed[- (garbage_len - 1):] if len(processed) >= garbage_len - 1 else processed
            # 写入除了尾部之外的内容
            tgt.write(processed[:- (garbage_len - 1)] if len(processed) >= garbage_len - 1 else "")
    # 可选：如果源文件是两个数组拼接成的 [xxx][yyy]，修复后会变成 [xxx,yyy]，不需要额外处理

处理完成后，你可以直接用ijson或者pandas读取修复后的target_path文件即可。

方案2：自定义流式读取类，直接喂给ijson解析，无需生成中间文件 #

如果不需要留存修复后的文件，可以自定义文件包装类，在read方法中实时处理替换，全程不落地中间文件，内存占用更低。

代码示例：

import ijson

class FixJSONReader:
    def __init__(self, file_path, chunk_size=16*1024):
        self.file = open(file_path, "r", encoding="utf-8")
        self.chunk_size = chunk_size
        self.garbage = "]["
        self.garbage_len = len(self.garbage)
        self.tail = ""
        self.eof = False
    
    def read(self, n=-1):
        if self.eof and not self.tail:
            return ""
        # 凑够需要返回的字节数
        while not self.eof and (len(self.tail) < n or n == -1):
            chunk = self.file.read(self.chunk_size)
            if not chunk:
                self.eof = True
                break
            full_chunk = self.tail + chunk
            processed = full_chunk.replace(self.garbage, ",")
            self.tail = processed[- (self.garbage_len - 1):] if len(processed) >= self.garbage_len -1 else processed
            # 把除了尾部的内容返回
            res = processed[:- (self.garbage_len -1)] if len(processed) >= self.garbage_len -1 else processed
            return res
        # 读取结束或者凑够了长度
        if n == -1:
            res = self.tail
            self.tail = ""
            return res
        else:
            res = self.tail[:n]
            self.tail = self.tail[n:]
            return res
    
    def close(self):
        self.file.close()

# 使用示例
with FixJSONReader("your_3gb_file.json") as f:
    # 直接用ijson流式解析
    for item in ijson.items(f, "item"):
        # 处理每一条数据
        print(item)

注意事项 #

• 可以根据你的设备性能调整chunk_size参数，16KB~64KB是比较通用的最优值，内存占用不会超过100MB
• 正式处理全量文件前，可以先读取前10MB数据做测试，确认替换规则正确、解析正常后再跑全量任务
• 如果除了']['之外还有其他类型的垃圾字符，只需要调整garbage变量的取值，同步修改预留尾部的长度即可

内容的提问来源于stack exchange，提问作者Sandeep G