Streamlit CRUD应用：高效将编辑后数据写回MySQL的最优方案

我使用Streamlit开发CRUD应用（UI部分与代码无关），从MySQL数据库读取数据后，需与另一上传数据（以df2表示，列结构一致）通过Pandas完成合并，再写回MySQL。当前采用先清空数据库所有内容，再写入合并后数据的方式。请问：

1. 这种全删全写的方式是否可行？
2. 如何更便捷地更新已修改数据并新增数据？
3. 是否存在比该批量操作更优的方案？

附上当前核心代码：

import pandas as pd
import streamlit as st
    
# 建立MySQL连接，密钥定义在.streamlit/secrets.toml中
@st.cache_resource
def init_connection():
    return st.experimental_connection('mysql',type='sql')
conn = init_connection()
    
# 缓存数据库数据
@st.cache_data#(ttl=600)
def run_query():
    return conn.query('SELECT * from db;')#, ttl=600)
    
# 获取数据库数据
df = run_query()
    
# 合并数据（结果包含新增行和修改行，已按指定列完成合并逻辑）
df=pd.merge(df,df2,how='outer',on='Par',indicator=True)

# 全删全写逻辑：清空表、重置自增ID、逐行插入合并后数据
with conn.session as s:
    s.execute(text("DELETE FROM `db`"))
    s.execute(text("ALTER TABLE `db` AUTO_INCREMENT = 1"))
    for i in df.T:
        e=dict(PN=df.Par[i],C1=df.C1[i],C2=df.C2[i])
        s.execute(text("INSERT INTO `db` (PN, C1, C2) VALUES (:PN,:C1,:C2)"),params=e)
    s.commit()

一、全删全写方式的可行性分析 #

这种方式在小数据量、无关联表、无并发访问的场景下暂时可用，但存在诸多致命问题：

• 数据一致性风险：操作期间若有其他用户读取数据库，会拿到空表数据；若操作中断（如程序崩溃），数据库会处于空数据状态。
• 关联表冲突：如果db表和其他表存在外键关联，清空操作会触发外键约束错误，导致操作失败。
• 性能瓶颈：数据量较大时，全删+逐行插入的效率极低，耗时会呈指数级增长。
• 历史数据丢失：无法保留数据修改的历史轨迹，后续排查问题或数据回溯会非常困难。
• 自增ID混乱：重置自增ID会破坏ID的连续性，若其他系统依赖该ID关联数据，会引发兼容性问题。

二、增量更新：新增+修改的高效实现 #

最优方案是利用MySQL的INSERT ... ON DUPLICATE KEY UPDATE语法，一次性完成新增数据插入和已有数据更新，无需全删全写。前提是要给Par列设置唯一键约束（或主键），让MySQL能识别重复的记录。

具体实现步骤： #

1. 给db表的Par列添加唯一约束：

   ALTER TABLE `db` ADD UNIQUE KEY `idx_par` (`Par`);
   

2. 优化合并逻辑，确保合并后的df包含最终需要的最新数据（示例优先保留df2的修改值）：

   # 合并时区分新旧数据
   df = pd.merge(df, df2, how='outer', on='Par', suffixes=('_old', '_new'))
   # 整合最新数据：df2有值则用新值，否则保留旧值
   df['C1'] = df['C1_new'].fillna(df['C1_old'])
   df['C2'] = df['C2_new'].fillna(df['C2_old'])
   # 去重并保留需要写入的列
   df_final = df[['Par', 'C1', 'C2']].drop_duplicates(subset='Par')
   

3. 用批量操作替代逐行插入，结合ON DUPLICATE KEY UPDATE写入数据：

   from sqlalchemy import text
   
   with conn.session as s:
       # 准备批量插入的参数列表
       insert_params = [{'Par': row['Par'], 'C1': row['C1'], 'C2': row['C2']} 
                       for _, row in df_final.iterrows()]
       # 执行批量插入+更新
       s.execute(
           text("""
               INSERT INTO `db` (Par, C1, C2) 
               VALUES (:Par, :C1, :C2)
               ON DUPLICATE KEY UPDATE 
                   C1 = VALUES(C1), 
                   C2 = VALUES(C2)
           """),
           insert_params
       )
       s.commit()
   

三、批量操作的进一步优化 #

1. 用Pandas的to_sql简化代码：
   可以直接用df.to_sql配合自定义方法实现ON DUPLICATE KEY UPDATE，代码更简洁：

   def insert_on_duplicate_update(table, conn, keys, data_iter):
       from sqlalchemy.dialects.mysql import insert
       data = [dict(zip(keys, row)) for row in data_iter]
       stmt = insert(table.table).values(data)
       update_stmt = stmt.on_duplicate_key_update(
           {k: stmt.inserted[k] for k in keys if k != 'Par'}
       )
       conn.execute(update_stmt)
   
   # 直接写入数据库，无需手动拼接SQL
   df_final.to_sql(
       name='db',
       con=conn.engine,
       if_exists='append',
       index=False,
       method=insert_on_duplicate_update
   )
   

2. 优化缓存策略：
   如果数据需要实时更新，给@st.cache_data设置合理的ttl（比如60秒），避免用户看到过期数据：

   @st.cache_data(ttl=60)
   def run_query():
       return conn.query('SELECT * from db;')
   

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