SQLite3数据库多次读写后部分数据丢失问题求助

问题背景 #

使用PySide6开发Qt应用，基于SQLite3存储数据，通过Pandas将DataFrame写入数据表，查询数据用于计算展示。多次执行读写操作后，未执行删除语句却出现部分数据丢失，数据表行数最高可达2000万。

数据保存代码片段 #

#Getting connection:
con = sqlite3.connect(f'./xxxx/{name}.db')
cursor = con.cursor()
table_name = "xyz"

#Check if table already exists in the database
check_table_query = f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}';"

#table exists
if table_exists:
  #Get already existing data from the table
  cursor.execute(f"SELECT * FROM '{table_name}'")
  existing_data = pd.DataFrame(cursor.fetchall())
  
  #no data in table, new data will all be saved in the table
  if existing_data.empty:
     print("existing data from db is empty")
     table_to_save.to_sql(table_name, con, if_exists='append', index=False)
     # Commit the transaction
     con.commit()

  #new data will be compared to existing data and just new rows will be saved in table
  else:
     existing_data.columns = table_to_save.columns
     table_to_save = table_to_save[~table_to_save.isin(existing_data.to_dict('list')).all(1)]

     #Append the new data to the table
     if not table_to_save.empty:
        table_to_save.to_sql(table_name, con, if_exists='append', index=False)
        # Commit the transaction
        con.commit()
     else:
        print("Data already in db, no saving needed.")


#table does not exist, table will be created and new data will be saved in this table
else:
   table_to_save.to_sql(table_name, con, if_exists='append', index=False)
   #Commit the transaction
   con.commit()

#Close the connection
con.close()

问题原因分析 #

• 全量读取触发内存溢出：当表达到2000万行时，SELECT * FROM '{table_name}'会将全部数据加载到内存生成DataFrame，极易触发内存不足，导致程序崩溃或写入操作中断，表现为数据丢失。
• 重复行判断逻辑失效：~table_to_save.isin(existing_data.to_dict('list')).all(1)的判断逻辑在存在NaN值时会失效，导致新数据被错误过滤，或重复数据被写入，加剧后续内存压力和数据异常。
• 事务与连接管理不规范：虽有commit操作，但未处理异常场景的回滚；若写入过程中出现错误，未提交的事务会丢失，且连接关闭前未确保所有操作完成。
• SQLite配置未优化：默认的日志模式（DELETE）在大写入量下易出现数据不一致；单写模式下若多线程共享连接，会引发写入冲突，导致部分操作静默失败。

解决办法 #

1. 替换全量读取，改用数据库层面去重 #

避免将所有数据加载到内存，通过SQL唯一约束或INSERT OR IGNORE实现去重：

# 首次建表时添加唯一索引（根据实际重复判断字段调整）
cursor.execute(f"CREATE UNIQUE INDEX IF NOT EXISTS idx_unique ON {table_name}(col1, col2)")

# 写入时直接追加，数据库自动忽略重复行
table_to_save.to_sql(table_name, con, if_exists='append', index=False, method='multi')
con.commit()

2. 修复重复行判断逻辑（若需内存处理） #

用merge替代isin，同时兼容NaN值：

merged = pd.merge(table_to_save, existing_data, on=list(table_to_save.columns), how='left', indicator=True)
table_to_save = merged[merged['_merge'] == 'left_only'].drop('_merge', axis=1)

3. 优化SQLite配置 #

# 连接时开启WAL模式、设置超时、禁用线程检查（多线程需用独立连接）
con = sqlite3.connect(f'./xxxx/{name}.db', detect_types=sqlite3.PARSE_DECLTYPES, timeout=30, check_same_thread=False)
cursor = con.cursor()
# 开启WAL提升并发和安全性
cursor.execute("PRAGMA journal_mode=WAL;")
# 调整缓存大小（单位为页，-20000表示20MB，根据内存调整）
cursor.execute("PRAGMA cache_size=-20000;")

4. 规范事务与连接管理 #

用try-except-finally确保事务安全：

try:
    # 所有数据库操作逻辑
    con.commit()
except Exception as e:
    con.rollback()
    print(f"写入失败：{str(e)}")
finally:
    con.close()

5. 多线程场景隔离连接 #

多线程应用中，每个线程创建独立的数据库连接，禁止共享连接；或使用连接池管理连接。

6. 分批写入大DataFrame #

将大DataFrame拆分为小批次写入，降低内存占用：

batch_size = 10000
for i in range(0, len(table_to_save), batch_size):
    batch = table_to_save.iloc[i:i+batch_size]
    batch.to_sql(table_name, con, if_exists='append', index=False)
con.commit()

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