如何使用pandas.to_sql()插入Oracle时仅捕获错误行

针对大Chunk插入时因个别重复行导致整段被标记失败的问题，有两种高效解决方案：

方案一：预查询过滤已存在的行（最优选择） #

提前从Oracle表中查询已存在的约束字段数据，在DataFrame中过滤掉重复行，直接插入剩余数据，同时标记过滤出的重复行为失败行。这种方式避免了插入时的错误抛出，处理效率远高于插入时再拆分Chunk。

假设表的唯一约束字段是id，代码示例：

import pandas as pd
from sqlalchemy import create_engine

# 创建数据库连接
engine = create_engine('oracle://user:password@host:port/service_name')
table_name = 'my_table'
new_data = pd.DataFrame({...})  # 你的待插入数据

# 从Oracle查询已存在的约束字段值
with engine.connect() as conn:
    existing_ids = pd.read_sql(f"SELECT id FROM {table_name}", conn)['id'].tolist()

# 拆分失败行和待插入行
failed_data = new_data[new_data['id'].isin(existing_ids)].copy()
failed_data['failed_reason'] = '约束字段已存在'
to_insert_data = new_data[~new_data['id'].isin(existing_ids)]

# 批量插入有效数据
if not to_insert_data.empty:
    to_insert_data.to_sql(table_name, engine, if_exists='append', index=False, chunksize=10000)

# 输出失败行
if not failed_data.empty:
    print('插入失败的行：')
    print(failed_data)

方案二：出错时递归拆分Chunk定位错误行 #

如果无法提前获取约束字段信息，或需要捕获其他类型的IntegrityError，可在Chunk插入失败时，将其递归拆分为更小的子Chunk，直到定位到单个错误行。这种方式比直接用chunk_size=1高效，尤其适合大部分行可成功插入的场景。

代码示例：

import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy.exc import IntegrityError

engine = create_engine('oracle://user:password@host:port/service_name')
table_name = 'my_table'
new_data = pd.DataFrame({...})
failed_rows = []

def insert_chunk(chunk):
    """递归插入Chunk，定位错误行"""
    try:
        chunk.to_sql(table_name, engine, if_exists='append', index=False, chunksize=len(chunk))
    except IntegrityError as e:
        if len(chunk) == 1:
            # 单个行插入失败，记录错误
            failed_row = chunk.copy()
            failed_row['failed_reason'] = str(e)
            failed_rows.append(failed_row)
        else:
            # 拆分Chunk为两部分，递归处理
            mid = len(chunk) // 2
            insert_chunk(chunk.iloc[:mid])
            insert_chunk(chunk.iloc[mid:])

# 按大Chunk拆分后处理
chunk_size = 10000
for i in range(0, len(new_data), chunk_size):
    chunk = new_data.iloc[i:i+chunk_size]
    insert_chunk(chunk)

# 合并失败行并输出
if failed_rows:
    failed_data = pd.concat(failed_rows)
    print('插入失败的行：')
    print(failed_data)

注意事项 #

• 方案一需根据实际约束调整查询字段，比如外键约束需查询关联表的对应字段。
• 方案二可调整拆分粒度（如每次拆为10份），进一步适配不同数据场景的效率需求。

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