JSON行处理循环仅14000 it/s，如何提升DuckDB数据导入效率？

我编写了一个读取JSON文件的函数，文件中每行JSON包含约78个键值对，通过字典推导式筛选所需键值对：

data={key:data.get(key,"") for key in keys} 

随后将数据导入DuckDB，完整代码如下：

def upload(file,conn):
    n=0
    temp=[]
    keys=['x1','x2','x3','x4','x5','x6','x7','x8','x9','x10','x11','x12','x13','x14','x15','x16','x17','x18','x19','x20','x21','x22','x23','x24','x25','x26','x27','x28','x29','x30','x31','x32','x33','x34','x35']
    for line in tqdm(file):
        data=json.loads(line)
        data={key:data.get(key,"") for key in keys}
        temp.append(data)
    df=pd.DataFrame(temp)
    conn.execute('INSERT INTO Main SELECT * FROM df')
    temp=[]

目前DuckDB的导入速度很快，但循环处理每行的速度仅约14000 it/s，想知道如何提升这部分的处理速度。

优化方案 #

1. 换用更快的JSON解析库 #

标准库json的解析性能一般，换成orjson或者ujson这类专门做性能优化的库，解析速度能提升2到5倍。安装后直接替换json.loads即可，代码改动极小。

示例（用orjson）：

import orjson

def upload(file,conn):
    temp=[]
    keys=['x1','x2',...,'x35']
    for line in tqdm(file):
        data=orjson.loads(line)
        data={key:data.get(key,"") for key in keys}
        temp.append(data)
    df=pd.DataFrame(temp)
    conn.execute('INSERT INTO Main SELECT * FROM df')

2. 批量处理减少循环开销 #

逐行处理的循环本身就有不少开销，改成每次读取几十到几百行作为一个批次，批量解析和筛选，能显著降低循环带来的性能损耗。

示例：

import orjson

def upload(file,conn):
    keys=['x1','x2',...,'x35']
    chunk_size = 1000  # 可根据内存情况调整
    while True:
        chunk = file.readlines(chunk_size)
        if not chunk:
            break
        # 批量解析所有行
        data_list = [orjson.loads(line) for line in chunk]
        # 批量筛选键值对
        filtered = [{key:d.get(key,"") for key in keys} for d in data_list]
        df = pd.DataFrame(filtered)
        conn.execute('INSERT INTO Main SELECT * FROM df')

3. 用生成器分批插入，降低内存压力 #

如果文件特别大，把所有数据先存到列表里会占大量内存，拖慢处理速度。改用生成器逐步生成数据，攒够一批就插入DuckDB，既能减少内存占用，也能让CPU和IO更均衡。

示例：

import orjson

def process_lines(file, keys):
    for line in file:
        data = orjson.loads(line)
        yield {key:data.get(key,"") for key in keys}

def upload(file,conn):
    keys=['x1','x2',...,'x35']
    batch_size = 2000
    batch = []
    for item in tqdm(process_lines(file, keys)):
        batch.append(item)
        if len(batch) >= batch_size:
            df = pd.DataFrame(batch)
            conn.execute('INSERT INTO Main SELECT * FROM df')
            batch = []
    # 处理最后剩下的少量数据
    if batch:
        df = pd.DataFrame(batch)
        conn.execute('INSERT INTO Main SELECT * FROM df')

4. 直接让DuckDB处理JSON文件（最优方案） #

DuckDB原生支持读取换行分隔的JSON文件，还能直接指定要提取的列，完全跳过Python层面的循环处理，这是速度最快的方式——相当于把数据处理的工作交给了更高效的DuckDB引擎，避免了Python和数据库之间的数据转换开销。

示例：

def upload(file_path, conn):
    keys=['x1','x2',...,'x35']
    # 构造SQL查询，用COALESCE给缺失的字段设默认值
    select_cols = ", ".join([f"COALESCE({key}, '') AS {key}" for key in keys])
    conn.execute(f"""
        INSERT INTO Main
        SELECT {select_cols}
        FROM read_json_auto('{file_path}', format='newline_delimited')
    """)

注意这里要传入文件路径，而不是打开的文件对象，DuckDB会自己高效读取和解析文件。

5. 微小优化：提前绑定方法减少属性查找 #

每次循环里调用data.get都会做一次属性查找，提前把dict.get绑定成变量，能减少这部分微小的开销，积少成多也能提升一点速度。

示例：

import orjson

def upload(file,conn):
    temp=[]
    keys=['x1','x2',...,'x35']
    get = dict.get  # 提前绑定get方法
    for line in tqdm(file):
        data=orjson.loads(line)
        data={key:get(data, key, "") for key in keys}
        temp.append(data)
    df=pd.DataFrame(temp)
    conn.execute('INSERT INTO Main SELECT * FROM df')

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