分组DataFrame最后一个无法写入MySQL的技术问询

问题描述 #

将主DataFrame按groupBy列（下划线分隔的日期字符串）拆分为4个分组DataFrame，前3个分组数据可正常插入对应MySQL临时表，但最后一个临时表（如tmp_lookup_2024_05_18）结构正常却无数据。

脚本执行步骤 #

1. 按groupBy列拆分主DataFrame为多个分组DataFrame；
2. 为每个分组创建命名格式为tmp_lookup_日期的MySQL临时表；
3. 将分组DataFrame的数据写入对应临时表。

核心代码 #

# Group the DataFrame by 'group_column'
grouped = df.groupby('groupBy')

for group_name, group_data in grouped:      
    # Create temporary table for each dataframe in for statement
    create_temp_table_query = conn_prod.execute(text(f"""CREATE TABLE tmp_lookup_{group_name} as SELECT * FROM staging_lookup limit 0;"""))

    # Insert data from dataframe into temporary table
    group_data.to_sql(f"tmp_lookup_{group_name}", con=conn_prod, if_exists='replace', index=False)

已完成的排查 #

• 主DataFrame数据完整符合预期；
• 打印每个分组DataFrame，数据均正确对应分组；
• 分组DataFrame的行数统计正确；
• 在Python脚本内查询临时表行数显示正常（含最后一个表），但直接在MySQL中查询该表行数为0；
• 切换engine.connect()+to_sql和mysql.connector+cursor两种实现方式，问题依旧。

关键原因分析 #

MySQL临时表是会话级生命周期：临时表仅在创建它的数据库会话中可见，会话关闭后自动销毁。你的问题核心在于：

1. to_sql的if_exists='replace'会先删除原临时表再重建，手动创建的表被替换后，若连接未正确提交或会话提前释放，最后一个表的数据无法在外部会话中被读取；
2. 循环操作中，最后一个分组的写操作可能因事务未提交，导致数据仅存在于脚本会话的内存中，未持久化到数据库的会话可见范围。

修复方案 #

方案1：移除手动建表步骤，让to_sql自动处理结构 #

to_sql可根据DataFrame自动生成表结构，无需手动复制原表，且if_exists='replace'能正确创建临时表并插入数据：

grouped = df.groupby('groupBy')

for group_name, group_data in grouped:      
    group_data.to_sql(
        name=f"tmp_lookup_{group_name}",
        con=conn_prod,
        if_exists='replace',
        index=False,
        # 可选：显式指定字段类型匹配原表，避免自动推断偏差
        dtype={col: sqlalchemy.types.VARCHAR(length=255) for col in group_data.columns}
    )
    # 手动提交事务（若连接未开启自动提交）
    conn_prod.commit()

方案2：保持会话一致性，统一在同一会话内操作 #

若必须手动复制原表结构，需确保所有操作在同一个数据库会话中完成，并显式提交事务：

from sqlalchemy import text

# 使用上下文管理器保持会话一致性
with conn_prod.connect() as session:
    grouped = df.groupby('groupBy')
    for group_name, group_data in grouped:      
        # 创建临时表（显式声明TEMPORARY更规范）
        session.execute(text(f"""CREATE TEMPORARY TABLE tmp_lookup_{group_name} as SELECT * FROM staging_lookup limit 0;"""))
        # 写入数据，使用当前会话且用append模式（避免重建表）
        group_data.to_sql(
            name=f"tmp_lookup_{group_name}",
            con=session,
            if_exists='append',
            index=False
        )
    # 提交所有操作
    session.commit()

方案3：验证临时表可见性 #

临时表仅在创建它的会话中可见，若需在外部MySQL客户端查看数据，需保持脚本的数据库会话不关闭，或改用普通表进行测试（脚本结束后会话关闭，临时表会被自动销毁）。

额外验证点 #

• 检查连接的自动提交设置：若conn_prod.autocommit为False，必须在写入后或循环结束后调用commit()；
• 打印f"tmp_lookup_{group_name}"确认表名无特殊字符解析错误。

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