使用Python的ijson库转换100GB JSON到CSV时遇解析错误求助

问题背景 #

使用Python的ijson库处理100GB级大型JSON文件转换为CSV，代码如下：

import ijson
import csv

input_file_path = 'path_to_json_file' #json file is in dump
output_file_path = 'path_to_the_csv_file.csv'

# List the fieldnames you want to include in the CSV file
desired_fieldnames = [
"col_1","col_2","col_3",...
]

# Buffer to store rows before writing to CSV
buffer_size = 1000000
rows_buffer = []

def write_buffer(writer, buffer):
    for row in buffer:
        writer.writerow(row)

with open(input_file_path, 'rb') as input_file, open(output_file_path, 'w', newline='', encoding='utf-8') as output_file:
    objects = ijson.items(input_file, 'rows.item.doc')
    writer = csv.DictWriter(output_file, fieldnames=desired_fieldnames)
    writer.writeheader()

    for item in objects:
        # Create a new dictionary with only the desired fields
        filtered_item = {field: item.get(field, '') for field in desired_fieldnames}
        rows_buffer.append(filtered_item)

        if len(rows_buffer) >= buffer_size:
            write_buffer(writer, rows_buffer)
            rows_buffer = []

    # Write any remaining rows in the buffer
    if rows_buffer:
        write_buffer(writer, rows_buffer)

运行后抛出解析错误：

Traceback (most recent call last):
  File "path_to_python_program", line 64, in <module>
    for item in objects:
ijson.common.IncompleteJSONError: parse error: unallowed token at this point in JSON text
          ~version\":\"sahdhdhash==\"}},  ]}
                     (right here) ------^

解决方案 #

1. 定位并修复JSON语法错误 #

从报错片段看，问题出在~version这个键名——JSON规范要求键名必须用双引号包裹，此处大概率缺失了开头的双引号（正确格式应为"~version":"sahdhdhash=="）。针对超大文件，可通过以下方式定位错误：

• 使用jq流式检查：执行jq --stream '.' large_file.json，工具会精准定位到语法错误的位置。
• 编写轻量Python脚本流式排查：

import json

def find_json_error(file_path):
    decoder = json.JSONDecoder()
    with open(file_path, 'r', encoding='utf-8') as f:
        buffer = ''
        line_num = 0
        for line in f:
            line_num +=1
            buffer += line
            try:
                while buffer:
                    obj, idx = decoder.raw_decode(buffer)
                    buffer = buffer[idx:]
            except json.JSONDecodeError as e:
                print(f"错误位置：第{line_num}行，偏移量{e.pos}")
                print(f"错误片段：{buffer[e.pos-20:e.pos+20]}")
                return

find_json_error('path_to_json_file')

定位到错误后，可通过sed、awk或Python文件操作修正缺失的双引号或其他语法问题。

2. 使用容错型JSON解析库替代ijson #

如果源文件无法修改，可换用对不规范JSON有容错能力的库，比如dirtyjson：

import dirtyjson
import csv

buffer_size = 1000000
rows_buffer = []

def write_buffer(writer, buffer):
    for row in buffer:
        writer.writerow(row)

with open(input_file_path, 'r', encoding='utf-8') as input_file, open(output_file_path, 'w', newline='', encoding='utf-8') as output_file:
    writer = csv.DictWriter(output_file, fieldnames=desired_fieldnames)
    writer.writeheader()
    
    # 分块读取避免内存溢出
    buffer = ''
    for line in input_file:
        buffer += line.strip()
        try:
            # 尝试解析当前缓冲内容
            data = dirtyjson.loads(buffer)
            # 遍历目标结构（根据实际JSON层级调整）
            for item in data.get('rows', []):
                doc = item.get('doc', {})
                filtered_item = {field: doc.get(field, '') for field in desired_fieldnames}
                rows_buffer.append(filtered_item)
                if len(rows_buffer) >= buffer_size:
                    write_buffer(writer, rows_buffer)
                    rows_buffer = []
            buffer = ''
        except dirtyjson.JSONDecodeError:
            # 缓冲内容不足以解析，继续读取下一行
            continue
    # 处理剩余内容
    if buffer:
        try:
            data = dirtyjson.loads(buffer)
            for item in data.get('rows', []):
                doc = item.get('doc', {})
                filtered_item = {field: doc.get(field, '') for field in desired_fieldnames}
                rows_buffer.append(filtered_item)
        except Exception as e:
            print(f"剩余内容解析失败：{e}")
    if rows_buffer:
        write_buffer(writer, rows_buffer)

注意：dirtyjson不支持原生流式解析，通过分块读取可避免内存过载。

3. 临时跳过错误片段（不推荐） #

如果错误是孤立的少量片段，可在ijson遍历过程中捕获错误并跳过：

# 替换原代码中的遍历部分
for item in objects:
    try:
        filtered_item = {field: item.get(field, '') for field in desired_fieldnames}
        rows_buffer.append(filtered_item)
        if len(rows_buffer) >= buffer_size:
            write_buffer(writer, rows_buffer)
            rows_buffer = []
    except ijson.common.IncompleteJSONError as e:
        print(f"跳过错误项：{e}")
        continue

这种方法会丢失错误位置的数据，仅适用于数据完整性要求不高的场景。

内容的提问来源于stack exchange，提问作者Luiz Mário Andrade