使用Python Pandas to_sql()向CrateDB批量插入时如何对错误行抛异常

该现象由CrateDB默认的批量插入策略导致：CrateDB的Python DBAPI驱动在执行executemany批量操作时，默认不会因单条数据错误终止整个批次，仅会跳过错误行、插入合法数据，也不会主动抛出异常。pandas的to_sql默认未校验批量插入的返回影响行数，因此无法感知到部分行插入失败的情况。仅当chunksize=1时，单条插入错误会直接触发异常，因此可以正常捕获。

• 前置数据校验（最推荐）
  插入前主动对目标字段做类型强制转换，提前筛选出错误行，避免插入时出错。示例代码：

  # 对目标numeric字段做类型转换，无法转换的值设为NaN
  df['target_numeric_col'] = pd.to_numeric(df['target_numeric_col'], errors='coerce')
  # 筛选出所有包含非法值的错误行，单独处理
  bad_rows = df[df['target_numeric_col'].isna()]
  # 仅保留合法行插入
  valid_df = df.dropna(subset=['target_numeric_col'])
  valid_df.to_sql(tableId, 'crate://xxxxxxx:4200', if_exists='append', index=False, chunksize=20000)
  

  同时建议给to_sql加上dtype参数，明确指定每个字段和CrateDB表对应的类型，避免pandas自动推断类型出错。
• 开启CrateDB批量插入失败快速终止配置
  在CrateDB连接地址中添加fail_fast=true参数，配置后只要批次内存在任意错误行，整个批次都会终止并抛出异常，你可以直接捕获到插入失败的事件：

  pandas.to_sql(tableId, 'crate://xxxxxxx:4200?fail_fast=true', if_exists='append', index=False, chunksize=20000)
  

  该方案的不足是无法直接定位具体错误行，需要拆分出问题批次后逐行排查。
• 自定义批量插入逻辑兼顾性能与错误定位
  不直接使用to_sql的默认逻辑，自行拆分数据批次，每个批次插入后校验返回的影响行数，若行数小于批次大小则逐行插入该批次定位错误行，既保留大批次插入的高性能，又能精准识别坏数据：

  from sqlalchemy import create_engine
  engine = create_engine('crate://xxxxxxx:4200')
  chunk_size = 20000
  bad_rows = []
  # 拆分数据批次
  for i in range(0, len(df), chunk_size):
      chunk = df.iloc[i:i+chunk_size]
      try:
          # 批量插入批次
          rows_affected = chunk.to_sql(tableId, engine, if_exists='append', index=False)
          # 校验插入行数是否匹配
          if rows_affected != len(chunk):
              # 逐行插入排查错误
              for _, row in chunk.iterrows():
                  try:
                      row.to_frame().T.to_sql(tableId, engine, if_exists='append', index=False)
                  except Exception as e:
                      bad_rows.append((row, str(e)))
      except Exception as e:
          # 批次整体失败时逐行排查
          for _, row in chunk.iterrows():
              try:
                  row.to_frame().T.to_sql(tableId, engine, if_exists='append', index=False)
              except Exception as e:
                  bad_rows.append((row, str(e)))
  

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