如何用Python自动根据CSV列创建MySQL表列并导入数据？

当然有可行方案，而且能轻松处理100列以上的CSV文件。下面提供两种实用方案，分别适配不同场景：

方案一：使用Pandas + SQLAlchemy（推荐，简洁高效） #

Pandas可自动推断CSV列的数据类型，结合SQLAlchemy能一键完成表创建与数据导入，对多列文件适配性极强。

操作步骤： #

1. 安装依赖：

pip install pandas sqlalchemy mysql-connector-python

2. 代码示例：

import pandas as pd
from sqlalchemy import create_engine

# 数据库连接配置
db_user = '你的用户名'
db_password = '你的密码'
db_host = 'localhost'
db_name = '目标数据库名'
table_name = '要创建的表名'

# CSV文件路径
csv_path = '你的多列CSV文件.csv'
# 超大文件可设置分块大小，避免内存溢出
chunk_size = 10000

# 创建数据库连接引擎
engine = create_engine(f'mysql+mysqlconnector://{db_user}:{db_password}@{db_host}/{db_name}')

# 分块读取并批量导入
for chunk in pd.read_csv(csv_path, chunksize=chunk_size):
    chunk.to_sql(
        name=table_name,
        con=engine,
        if_exists='append',  # 可选'replace'覆盖表/'fail'存在则报错
        index=False,
        # 若需手动指定列类型，添加dtype参数，例如：dtype={'id': sqlalchemy.types.INTEGER()}
    )

print("数据导入完成")

核心优势： #

• 自动处理列名映射与数据类型推断，无需手动编写建表语句
• 分块读取支持GB级超大文件，避免内存耗尽
• 代码极简，大幅减少多列场景下的手动工作量

方案二：使用原生CSV模块 + MySQL-Connector（底层可控） #

如果需要精细控制表结构或数据清洗逻辑，可通过原生CSV模块读取内容，手动生成建表语句并批量插入数据。

操作步骤： #

1. 安装依赖：

pip install mysql-connector-python

2. 代码示例：

import csv
import mysql.connector
from mysql.connector import Error

def infer_data_type(value):
    # 自定义数据类型推断逻辑，可按需扩展
    try:
        int(value)
        return 'INT'
    except ValueError:
        try:
            float(value)
            return 'FLOAT'
        except ValueError:
            # 字符串类型默认设为255长度，长文本可改为TEXT
            return 'VARCHAR(255)'

# 数据库连接配置
db_config = {
    'user': '你的用户名',
    'password': '你的密码',
    'host': 'localhost',
    'database': '目标数据库名'
}

csv_path = '你的多列CSV文件.csv'
table_name = '要创建的表名'

try:
    # 建立数据库连接
    connection = mysql.connector.connect(**db_config)
    cursor = connection.cursor()

    # 读取CSV表头与首行数据，用于推断列类型
    with open(csv_path, 'r', encoding='utf-8') as csv_file:
        csv_reader = csv.reader(csv_file)
        headers = next(csv_reader)
        first_row = next(csv_reader)

        # 生成CREATE TABLE语句，用反引号包裹列名避免关键字冲突
        columns_def = []
        for header, val in zip(headers, first_row):
            col_name = f"`{header}`"
            col_type = infer_data_type(val)
            columns_def.append(f"{col_name} {col_type}")
        create_table_sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join(columns_def)})"
        cursor.execute(create_table_sql)

        # 批量插入数据，每1000条提交一次提升效率
        csv_file.seek(0)
        next(csv_reader)  # 跳过表头
        insert_sql = f"INSERT INTO {table_name} ({', '.join([f'`{h}`' for h in headers])}) VALUES ({', '.join(['%s']*len(headers))})"
        batch_size = 1000
        batch_data = []
        for row in csv_reader:
            batch_data.append(row)
            if len(batch_data) == batch_size:
                cursor.executemany(insert_sql, batch_data)
                connection.commit()
                batch_data = []
        # 插入剩余数据
        if batch_data:
            cursor.executemany(insert_sql, batch_data)
            connection.commit()

    print("表创建及数据导入完成")

except Error as e:
    print(f"数据库操作错误: {e}")
finally:
    if connection.is_connected():
        cursor.close()
        connection.close()

核心优势： #

• 完全可控数据类型推断与表结构定义
• 批量插入优化性能，适配超大规模数据集
• 无额外依赖，轻量灵活

关键注意事项 #

• 表头处理：若CSV表头包含MySQL关键字（如user、date），必须用反引号`包裹列名
• 数据类型调整：自动推断的类型可能存在偏差（如长文本需用TEXT而非VARCHAR(255)），可根据实际需求修改推断逻辑
• 编码一致性：确保CSV文件与数据库编码统一（推荐UTF-8）
• 权限配置：确保MySQL用户拥有目标数据库的CREATE TABLE和INSERT权限

内容的提问来源于stack exchange，提问作者Dwip Ranjan Pal