如何根据CSV列值将数据插入对应MySQL表？求最优方案

我有一个.csv文件，为简化说明，假设它包含4列。我需要根据第4列的值将对应行插入MySQL表——如果第4列的值为'x'，该行就插入表'x'中。我已经实现了自动为第4列的每个唯一值创建对应表（目前有28个唯一值）。

请问最优的插入方式是什么？
我应该先预处理数据，为每个唯一值创建新的Pandas DataFrame，再逐个插入这些DataFrame？还是在INSERT语句中进行区分？

我刚接触Python，如果您发现其他不当之处，欢迎指出。

import mysql.connector as msql
import numpy as np
import pandas as pd
from mysql.connector import Error
from UniqueRounds import unique_rounds

size = len(unique_rounds)

for i in range(size):
    element = unique_rounds[0]
    unique_rounds = np.delete(unique_rounds, 0)
    unique_rounds = np.append(unique_rounds, "`" + element + "`")

table_data = pd.DataFrame(pd.read_csv('PATH/file.csv',
                                      usecols=[0, 1, 13, 14]))

try:
    conn = msql.connect(host='localhost', user='root',
                        password='pw',
                        database='funding_research')
    if conn.is_connected():
        cursor = conn.cursor()
        cursor.execute("select database();")
        record = cursor.fetchone()
        print("You're connected to database: ", record)

        for i in unique_rounds:
            try:
                cursor.execute(
                "CREATE TABLE " + i + " (Uuid varchar(255), Name varchar(255), FundingRound varchar(255), Amount BIGINT(255) DEFAULT NULL)")
                print("Table is created...")
            except Error as e:
                print("Table already exists")
except Error as e:
    print("Error while connecting to MySQL", e)

一、最优插入方式：分组拆分DataFrame后批量插入 #

针对28个分组的场景，先按第4列（即FundingRound列）拆分DataFrame，再对每个分组批量插入对应表是效率更高的方案：

• Pandas的to_sql支持批量插入，比逐行执行INSERT语句速度快数倍；
• 内存中拆分DataFrame的逻辑清晰，后续维护成本低。

具体实现步骤：

1. 读取CSV时指定列名，避免索引混乱；
2. 用groupby按目标列分组；
3. 遍历分组，调用to_sql批量插入对应表。

示例代码（可衔接原数据库连接逻辑）：

# 需先安装sqlalchemy：pip install sqlalchemy
import sqlalchemy

if conn.is_connected():
    cursor = conn.cursor()
    # 读取数据时直接映射列名，避免依赖列索引
    table_data = pd.read_csv(
        'PATH/file.csv', 
        usecols=[0,1,13,14],
        names=['Uuid', 'Name', 'FundingRound', 'Amount']
    )
    
    # 按FundingRound分组
    grouped = table_data.groupby('FundingRound')
    
    for round_name, group_df in grouped:
        table_name = f"`{round_name}`"
        try:
            # 批量插入，指定if_exists='append'实现追加
            group_df.to_sql(
                name=round_name.replace('`',''),  # 去掉反引号适配to_sql参数要求
                con=conn,
                if_exists='append',
                index=False,
                dtype={
                    'Uuid': sqlalchemy.types.VARCHAR(255),
                    'Name': sqlalchemy.types.VARCHAR(255),
                    'FundingRound': sqlalchemy.types.VARCHAR(255),
                    'Amount': sqlalchemy.types.BIGINT()
                }
            )
            conn.commit()
            print(f"成功插入{len(group_df)}行到表{table_name}")
        except Exception as e:
            conn.rollback()
            print(f"插入表{table_name}失败：{str(e)}")

二、原代码的优化点 #

1. 简化unique_rounds处理逻辑：
   原循环删除追加的写法冗余，可直接用列表推导式实现：

   unique_rounds = [f"`{item}`" for item in unique_rounds]
   

2. 优化表创建语句：
   用CREATE TABLE IF NOT EXISTS可避免捕获“表已存在”的异常，代码更简洁：

   for round_name in unique_rounds:
       create_sql = """
       CREATE TABLE IF NOT EXISTS %s (
           Uuid varchar(255), 
           Name varchar(255), 
           FundingRound varchar(255), 
           Amount BIGINT(255) DEFAULT NULL
       )
       """
       cursor.execute(create_sql, (round_name,))
   

3. 添加数据库连接关闭逻辑：
   原代码未关闭连接，建议在操作结束后补充：

   finally:
       if conn.is_connected():
           cursor.close()
           conn.close()
           print("数据库连接已关闭")
   

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