基于Python将超大型JSON导入Oracle/PostgreSQL数据库

一、超大型JSON的流式解析与遍历 #

针对100-150GB级别的JSON文件，绝对不能用json.load()这类一次性加载整个文件到内存的方法，必须采用流式解析库，推荐ijson——它能逐节点读取JSON，内存占用极低，适合新手快速上手。

核心用法示例 #

1. 安装库：

pip install ijson

2. 定位并遍历目标数组：
   假设你的JSON结构是顶级对象包含13个数组（如{"array1": [...], "array2": [...], ...}），可以用以下代码逐个遍历数组元素：

import ijson

def stream_json_elements(file_path, array_name):
    """流式读取指定数组的所有元素"""
    with open(file_path, 'rb') as f:
        # 定位到目标数组的元素节点
        parser = ijson.items(f, f'{array_name}.item')
        for elem in parser:
            yield elem

# 遍历其中一个数组，比如array1
for item in stream_json_elements('large_data.json', 'array1'):
    # 可在此对单个元素做预处理（如提取指定属性、格式转换）
    processed_item = {k: v for k, v in item.items() if k in ['目标字段1', '目标字段2']}
    # 后续交给数据库插入逻辑

关键注意事项 #

• 用rb模式打开文件，避免编码问题干扰流式解析
• 即使JSON有换行或缩进，ijson仍能正常处理，无需预处理
• 可同时启动多个解析器处理不同数组，但需注意磁盘IO瓶颈

二、Pythonic数据库批量导入方案 #

针对8亿级别的数据，逐条插入是灾难，必须采用批量插入+绑定变量的方式，以下分Oracle（优先）和PostgreSQL两种场景说明：

Oracle 批量导入（基于cx_Oracle） #

1. 安装Oracle驱动：

pip install cx_Oracle

2. 批量插入实现：

import cx_Oracle
from ijson import items

# 数据库连接配置（根据实际情况修改）
DB_CONFIG = {
    'user': '你的用户名',
    'password': '你的密码',
    'dsn': '你的主机:1521/服务名'
}

BATCH_SIZE = 10000  # 每批次插入数量，可根据数据库性能调整

def insert_batch(conn, table_name, columns, data_batch):
    """批量插入数据到Oracle"""
    # 构造带绑定变量的SQL语句
    placeholders = ', '.join([f':{i+1}' for i in range(len(columns))])
    sql = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({placeholders})"
    
    cursor = conn.cursor()
    try:
        cursor.executemany(sql, data_batch)
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cursor.close()

def main():
    # 1. 连接数据库
    conn = cx_Oracle.connect(**DB_CONFIG)
    
    # 2. 处理目标数组（以array1为例，对应表table1）
    table_name = 'table1'
    columns = ['col1', 'col2', 'col3']  # 对应JSON元素的属性名
    data_batch = []
    
    with open('large_data.json', 'rb') as f:
        parser = items(f, 'array1.item')
        for elem in parser:
            # 提取需要的字段，转换为数据库兼容格式
            row = (elem['col1'], elem['col2'], elem['col3'])
            data_batch.append(row)
            
            # 达到批次大小则执行插入
            if len(data_batch) >= BATCH_SIZE:
                insert_batch(conn, table_name, columns, data_batch)
                data_batch = []
        
        # 处理剩余的不足一批的数据
        if data_batch:
            insert_batch(conn, table_name, columns, data_batch)
    
    conn.close()

if __name__ == '__main__':
    main()

PostgreSQL 批量导入（基于psycopg2） #

如果用PostgreSQL，除了executemany，还可以用copy_from进一步提升性能：

import psycopg2
from io import StringIO
from ijson import items

DB_CONFIG = {
    'dbname': '你的数据库名',
    'user': '你的用户名',
    'password': '你的密码',
    'host': '你的主机'
}

def copy_from_stream(conn, table_name, columns, data_generator):
    """用copy_from批量导入数据"""
    buffer = StringIO()
    for row in data_generator:
        buffer.write('\t'.join(map(str, row)) + '\n')
    
    buffer.seek(0)
    cursor = conn.cursor()
    try:
        cursor.copy_from(buffer, table_name, columns=columns)
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cursor.close()

# 后续逻辑类似，将JSON元素转换为行数据生成器，传入copy_from_stream即可

优化建议 #

• 数据库端优化：导入前禁用目标表的索引、外键约束，导入完成后再重建；调整数据库的commit频率、PGA/SGA内存配置
• 并发处理：可以用multiprocessing启动多个进程，每个进程处理一个数组，或者将单个数组拆分后并行处理（注意避免数据库连接数过载）
• 错误处理：添加日志记录失败的行，方便后续补插；可将失败的数据写入临时文件，事后单独处理

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