如何处理2GB级JSONP数据并导入数据库？

问题背景 #

从archive.org获取了2GB的JSONP格式图书数据集，尝试过下载后去除callback()包裹再用json.loads()拆分，但数据存在损坏；编写的Python代码效果不佳，核心需求是：

• 将巨型JSONP拆分为结构完整的小型JSON文件
• 或直接从在线源流式导入数据库，彻底避免内存溢出

原代码的核心问题：一次性加载全量2GB数据到内存，且按固定字节拆分破坏了JSON结构，导致解析失败。

一、正确拆分JSONP为小型JSON文件（流式处理） #

核心思路：不加载全量数据到内存，用流式解析工具跳过JSONP的callback(前缀和)后缀，按指定条数拆分保存，保证每个小文件的JSON结构完整。

使用ijson库（专门处理大JSON的流式解析工具），代码示例：

import os
import json
import ijson
import requests

# 替换为你的JSONP链接
url = "https://archive.org/advancedsearch.php?q=collection%3Ainternetarchivebooks&fl[]=creator&fl[]=format&fl[]=genre&fl[]=language&fl[]=name&fl[]=title&fl[]=type&fl[]=year&sort[]=&sort[]=&sort[]=&rows=100000000&page=1&output=json&callback=callback&save=yes"
batch_size = 10000  # 每个小文件存储1万条数据
dir_name = "data_split"
os.makedirs(dir_name, exist_ok=True)

# 流式请求数据
response = requests.get(url, stream=True)
response.raw.decode_content = True

# 跳过JSONP的`callback(`前缀
while True:
    char = response.raw.read(1).decode()
    if char == '(':
        break

# 流式解析JSON数组中的每个元素，批量保存
count = 0
file_count = 1
current_batch = []
for item in ijson.items(response.raw, 'item'):
    current_batch.append(item)
    count += 1
    if count % batch_size == 0:
        with open(os.path.join(dir_name, f"{file_count}.json"), 'w', encoding='utf-8') as f:
            json.dump(current_batch, f, ensure_ascii=False, indent=2)
        current_batch = []
        file_count += 1

# 处理最后一批剩余数据
if current_batch:
    with open(os.path.join(dir_name, f"{file_count}.json"), 'w', encoding='utf-8') as f:
        json.dump(current_batch, f, ensure_ascii=False, indent=2)

# 跳过JSONP末尾的`)`
while True:
    char = response.raw.read(1).decode()
    if char == ')':
        break

print(f"拆分完成：共生成{file_count}个文件，总计{count}条数据")

关键说明：

• ijson仅解析需要的item节点，内存占用始终极低
• 按数据条数拆分而非字节数，确保每个小文件的JSON结构合法
• 全程流式处理，无需加载2GB全量数据到内存

二、直接流式导入数据库（以SQLite为例） #

如果不需要保存中间JSON文件，可直接流式解析并写入数据库，完全规避内存压力：

import sqlite3
import ijson
import requests

# 替换为你的JSONP链接
url = "https://archive.org/advancedsearch.php?q=collection%3Ainternetarchivebooks&fl[]=creator&fl[]=format&fl[]=genre&fl[]=language&fl[]=name&fl[]=title&fl[]=type&fl[]=year&sort[]=&sort[]=&sort[]=&rows=100000000&page=1&output=json&callback=callback&save=yes"
db_name = "books.db"

# 初始化数据库表
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS books (
    creator TEXT,
    format TEXT,
    genre TEXT,
    language TEXT,
    name TEXT PRIMARY KEY,
    title TEXT,
    type TEXT,
    year TEXT
)
''')
conn.commit()

# 流式请求+解析+批量入库
response = requests.get(url, stream=True)
response.raw.decode_content = True

# 跳过`callback(`前缀
while True:
    char = response.raw.read(1).decode()
    if char == '(':
        break

batch_size = 5000
current_batch = []
for item in ijson.items(response.raw, 'item'):
    # 提取字段，处理缺失值
    record = (
        item.get('creator'),
        item.get('format'),
        item.get('genre'),
        item.get('language'),
        item.get('name'),
        item.get('title'),
        item.get('type'),
        item.get('year')
    )
    current_batch.append(record)
    if len(current_batch) >= batch_size:
        cursor.executemany('''
        INSERT OR IGNORE INTO books VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ''', current_batch)
        conn.commit()
        current_batch = []

# 处理剩余数据
if current_batch:
    cursor.executemany('''
    INSERT OR IGNORE INTO books VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    ''', current_batch)
    conn.commit()

# 跳过末尾的`)`
while True:
    char = response.raw.read(1).decode()
    if char == ')':
        break

conn.close()
print("数据已成功导入数据库")

关键说明：

• 使用executemany批量插入，大幅提升数据库写入效率
• INSERT OR IGNORE避免重复数据（基于name主键）
• 适配MySQL/PostgreSQL等其他数据库，只需替换数据库连接逻辑

三、更简便的优化：直接请求纯JSON而非JSONP #

注意到你的请求链接中已经包含output=json，仅因为加了callback=callback才变成JSONP。直接删除&callback=callback参数，就能获取纯JSON格式数据，省去处理前缀后缀的步骤，ijson可直接解析，代码会更简洁。

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