生成器迭代未触发StopIteration，数据入库收尾失败求助

将超百万条数据的生成器内容导入SQL数据库，采用每10000条存入Pandas DataFrame后通过to_sql提交的方案。填满批次时运行正常，但最后一批处理完后，StopIteration从未触发，程序无响应。调试发现第1384471条数据已处理，再次调用next时既无新数据也不抛出异常，程序停滞。改用for循环后，循环后的收尾代码也未执行。

相关核心代码如下：

import ijson
import pandas as pd
import sqlalchemy as sa

def connect_to_database():
    engine = sa.create_engine(f"postgresql+psycopg2://{USER}:{PWD}@{SERVERNAME}:{PORT}/{DB}")
    return engine

engine = connect_to_database()

def create_content(objects, table_name):
    '''从JSON文件的对象列表中提取内容'''
    item_lst = []
    item_count = 0

    while True:
        try:
            item = next(objects)
            value_conversion(item)
            item_lst.append(item)
            
            if len(item_lst) == 10000:
                item_count += len(item_lst)
                submit_sql(table_name, item_lst)
                item_lst.clear()

        except StopIteration:
            if len(item_lst) > 0:
                item_count += len(item_lst)
                submit_sql(table_name, item_lst)
                item_lst.clear()
                print(item_count)
            break               
        except Exception as e:
            print(e)

def submit_sql(engine, table_name, items):
    '''提交DataFrame内容到SQL数据库'''
    engine = connect_to_database()
    temp_df = pd.DataFrame(items)
    temp_df.to_sql(table_name, engine, if_exists='append', index=False)
    engine.dispose()

with open("C:/Path/To/Json/file.json", "r", encoding="utf-8") as json_file:
    s_objects = ijson.items(json_file, 'ObjectList.item')
    create_content(s_objects, 'tablename')

尝试的for循环版本：

try:
    for item in objects:
        value_conversion(item)
        item_lst.append(item)
        
        if len(item_lst) == 10000:
            item_count += len(item_lst)
            submit_sql(table_name, item_lst)
            item_lst.clear()
            print(item_count)

    if len(item_lst) > 0:
        item_count += len(item_lst)
        submit_sql(table_name, item_lst)
        item_lst.clear()
        print(item_count)

except Exception as e:
    print(e)

1. ijson生成器未正常终止
   ijson采用流式解析JSON，若JSON文件存在格式问题（如末尾多余逗号、未闭合的括号/引号，或隐藏的无效字符），解析器会卡在最后一段数据之后，既不返回新数据也不抛出StopIteration，导致程序无限等待。

2. 数据库连接的冗余操作
   submit_sql函数每次提交都重新创建数据库引擎并立即销毁，频繁的连接建立/销毁会消耗大量资源，不仅降低效率，还可能引发连接池异常，拖慢甚至阻塞最后批次的处理流程。此外，函数定义的engine参数未被使用，属于冗余代码。

3. 循环计数逻辑的潜在问题
   原while循环中item_count仅在批次满10000时累加，若生成器因解析问题阻塞，StopIteration分支的收尾代码根本无法执行；for循环版本的收尾代码同样因生成器未终止而无法触发。

1. 修复JSON解析问题 #

• 先检查JSON文件完整性：用本地工具（如python -m json.tool your_file.json）验证文件格式，修复末尾无效字符、未闭合结构等问题。
• 改用itertools.islice分批处理生成器，更可靠地检测迭代终止。

2. 优化数据库连接逻辑 #

复用数据库引擎，避免频繁创建/销毁连接：

def submit_sql(engine, table_name, items):
    '''提交items到SQL数据库'''
    temp_df = pd.DataFrame(items)
    temp_df.to_sql(table_name, engine, if_exists='append', index=False)

3. 重构批次处理逻辑 #

使用itertools.islice实现更稳定的分批迭代，确保能正确检测生成器终止：

from itertools import islice

def create_content(objects, table_name, engine):
    item_count = 0
    while True:
        # 每次从生成器中取10000条数据
        batch = list(islice(objects, 10000))
        if not batch:
            # 批次为空，说明迭代完全结束
            break
        # 处理批次内的每个item
        for item in batch:
            value_conversion(item)
        # 提交到数据库
        submit_sql(engine, table_name, batch)
        item_count += len(batch)
        print(f"已处理 {item_count} 条数据")

4. 完善调用流程 #

确保数据库引擎在处理完成后正确销毁，同时捕获可能的解析异常：

with open("C:/Path/To/Json/file.json", "r", encoding="utf-8") as json_file:
    s_objects = ijson.items(json_file, 'ObjectList.item')
    engine = connect_to_database()
    try:
        create_content(s_objects, 'tablename', engine)
    except Exception as e:
        print(f"处理过程中出现错误: {str(e)}")
    finally:
        # 确保引擎最终被销毁
        engine.dispose()

额外建议 #

• 在value_conversion函数中添加异常捕获，避免单个数据项的转换错误导致整个流程阻塞：

  def value_conversion(item):
      try:
          # 原转换逻辑
          pass
      except Exception as e:
          print(f"转换数据项失败: {str(e)}, 数据项: {item}")
          # 可选：跳过错误项或记录日志
  

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