使用SQLAlchemy将DataFrame列插入不同表及重复条目错误解决

嘿，我来帮你梳理下这个问题，从代码里能看出你刚接触SQLAlchemy和Python，别担心，咱们一步步来解决～

1. 先搞定烦人的「Duplicate Entry」错误 #

你的核心问题在于插入数据的逻辑完全错了：你现在是循环给每个字段单独插入一行，但timestamp被设为主键，第一次插入timestamp后，后面再插入system_id、simulated_yield这些值时，又会生成新的行，而主键要求唯一，自然触发重复错误。

举个直观的例子：你先插入(timestamp='2024-01-01', system__id=None, ...)，然后又插入(timestamp=None, system__id=123, ...)——要么timestamp为空（不符合主键非空要求），要么数据库尝试用默认值，但之前已经有相同的timestamp，直接报错。

正确的逻辑是：一条完整的记录要一次性插入所有字段，而不是拆分到多个insert语句里。

2. 优化DataFrame写入数据库的最优姿势 #

你现在的循环逐行插入效率极低，还容易出错。其实Pandas本身就支持直接把DataFrame写入SQLAlchemy连接的数据库，用df.to_sql()方法，这才是专业玩家的做法。

步骤1：先整理你的数据结构 #

首先要把对应每个系统的所有字段（timestamp、system_id、simulated_yield、global_irradiance）合并成一个完整的DataFrame，而不是分散在多个独立的DataFrame里。比如对于系统index，你需要把：

• timestamp_df的timestamp_utc列
• sys_ids_df的对应system_id
• pr_daily_df的pr_x列
• rad_daily_df的rad_x列

整合成一行对应一条完整记录的DataFrame。

步骤2：用to_sql批量写入（还能处理重复值） #

直接用to_sql批量写入，同时可以通过参数或额外逻辑处理主键重复问题。

改写后的示例代码 #

import pandas as pd
import numpy as np
from sqlalchemy import create_engine, Table, Column, DateTime, Integer, Float, MetaData

# 假设你的数据库连接已经创建好，比如MySQL：
# engine = create_engine('mysql+pymysql://用户名:密码@主机地址/数据库名')
meta = MetaData()

x = 0
for index in a_id:
    table_name = f'simulated_for_sys_{np.int_(index)}'
    
    # 1. 定义表结构
    table_sim = Table(
        table_name, meta,
        Column('timestamp', DateTime, primary_key=True),
        Column('system__id', Integer),
        Column('simulated_yield_in_kWh', Float),
        Column('global_irradiance_tilted_in_kWh_per_m2', Float)
    )
    
    # 2. 创建表（如果不存在）
    if not engine.dialect.has_table(engine, table_name):
        meta.create_all(engine)
        print(f"表 {table_name} 已创建")
    else:
        print(f"表 {table_name} 已存在...")
    
    # 3. 整理对应这个系统的完整数据（确保所有列长度一致）
    # 如果system_id是单一值，记得重复填充到对应长度
    system_data = pd.DataFrame({
        'timestamp': timestamp_df['timestamp_utc'].dt.to_pydatetime(),
        'system__id': sys_ids_df['system_id'],  # 若为单一值可改为：[sys_ids_df['system_id'].iloc[0]] * len(timestamp_df)
        'simulated_yield_in_kWh': pr_daily_df[f'pr_{x}'],
        'global_irradiance_tilted_in_kWh_per_m2': rad_daily_df[f'rad_{x}']
    })
    
    # 4. 批量写入，处理主键重复
    try:
        system_data.to_sql(
            name=table_name,
            con=engine,
            if_exists='append',  # 追加数据，而非覆盖
            index=False,
            dtype={
                'timestamp': DateTime,
                'system__id': Integer,
                'simulated_yield_in_kWh': Float,
                'global_irradiance_tilted_in_kWh_per_m2': Float
            }
        )
        print(f"数据已成功写入表 {table_name}")
    except Exception as e:
        # 捕获主键重复错误，过滤已存在的记录后再写入
        if "Duplicate entry" in str(e):
            # 查询数据库中已有的timestamp
            existing_timestamps = pd.read_sql(f"SELECT timestamp FROM {table_name}", con=engine)['timestamp']
            # 过滤掉已存在的行
            new_data = system_data[~system_data['timestamp'].isin(existing_timestamps)]
            # 重新写入新数据
            new_data.to_sql(
                name=table_name,
                con=engine,
                if_exists='append',
                index=False,
                dtype={
                    'timestamp': DateTime,
                    'system__id': Integer,
                    'simulated_yield_in_kWh': Float,
                    'global_irradiance_tilted_in_kWh_per_m2': Float
                }
            )
            print(f"跳过重复记录，新数据已写入表 {table_name}")
        else:
            # 其他错误直接抛出
            raise e
    
    x += 1

3. 关于「将DataFrame不同列写入不同表」的需求 #

如果你的场景是要把单个/多个DataFrame的指定列分别写入不同表（比如把pr_daily_df的pr_0写入表A，pr_1写入表B），最优方式还是用to_sql：

• 提取需要的列生成新的DataFrame（比如df_subset = pr_daily_df[['pr_0']]）
• 重命名列以匹配目标表的字段（比如df_subset.rename(columns={'pr_0': 'simulated_yield_in_kWh'}, inplace=True)）
• 调用df_subset.to_sql()写入对应表

这种方式比循环逐行插入高效数倍，代码也更简洁易维护。

额外优化建议 #

• 避免重复创建数据库连接：你之前在循环里多次调用engine.connect()，其实to_sql会自动处理连接，不需要手动重复创建。
• 批量写入效率更高：to_sql默认是批量插入，还可以通过chunksize参数调整每次写入的行数，远快于逐行插入。
• 数据一致性检查：确保合并的所有列长度一致，否则会出现数据错位的问题。

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