如何用Pandas+SQLAlchemy批量插入多对多关联表数据？

针对百万级数据场景，最优方案是利用数据库层面的批量关联查询插入，避免Python端循环处理带来的性能损耗。以下是两种基于SQLAlchemy/纯SQL的实现方式：

方案一：SQLAlchemy Core + 临时表关联插入 #

结合Pandas临时表导入和SQLAlchemy Core的批量操作，所有关联逻辑在数据库内完成：

1. 从原CSV数据中提取去重的员工-产品关联关系：

import pandas as pd
from sqlalchemy import text, insert, select

# 假设employee_data是你读取的CSV DataFrame
unique_associations = employee_data[['Name', 'key_id', 'product']].drop_duplicates()

2. 将去重后的关联数据导入SQLite临时表：

# 创建临时表，SQLite中临时表仅在当前会话有效，结束后自动销毁
unique_associations.to_sql('temp_associations', engine, if_exists='replace', index=False)

3. 用SQLAlchemy Core构建批量插入语句：

# 关联临时表、员工表、产品表，获取对应主键后插入关联表
insert_stmt = insert(call_has_product_table).from_select(
    columns=['employee_id', 'product_id'],
    selectable=select(
        cv_calls_table.c.id,
        cv_products_table.c.id
    )
    .select_from(
        cv_calls_table.join(
            unique_associations_temp,
            (cv_calls_table.c.name == unique_associations_temp.c.Name) & 
            (cv_calls_table.c.key_id == unique_associations_temp.c.key_id)
        ).join(
            cv_products_table,
            cv_products_table.c.product == unique_associations_temp.c.product
        )
    )
    .distinct()  # 避免插入重复关联记录
)

# 执行并提交
with engine.connect() as conn:
    conn.execute(insert_stmt)
    conn.commit()

方案二：纯SQL INSERT...SELECT语句 #

如果更倾向直接用SQL，效率同样高效：

1. 同样先提取去重关联数据并导入临时表：

unique_associations = employee_data[['Name', 'key_id', 'product']].drop_duplicates()
unique_associations.to_sql('temp_associations', engine, if_exists='replace', index=False)

2. 执行纯SQL完成批量插入：

from sqlalchemy import text

with engine.connect() as conn:
    conn.execute(text("""
        INSERT INTO employee_has_product (employee_id, product_id)
        SELECT DISTINCT e.id, p.id
        FROM temp_associations ta
        JOIN employee e ON ta.Name = e.name AND ta.key_id = e.key_id
        JOIN products p ON ta.product = p.product
    """))
    conn.commit()

方案优势 #

• 性能最优：所有操作在数据库内部完成，避免Python与数据库的多次IO往返，适配百万级数据量。
• 去重保障：通过DISTINCT确保不会插入重复的员工-产品关联记录。
• 无ORM开销：跳过ORM实例化对象的环节，直接操作底层表结构，减少额外损耗。

注意事项 #

• 确保员工表(employee)和产品表(products)已完成数据插入，且name+key_id是员工的唯一标识、product是产品的唯一标识（若不是，需调整关联条件）。
• SQLite临时表仅在当前连接会话中存在，执行完插入后自动销毁，不占用持久化存储。

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