Python轻松实现将Excel数据批量导入数据库_Python

在日常数据处理工作中，将 excel 文件内容导入数据库是一个常见需求。python 生态中虽然有 pandas、openpyxl 等成熟方案，但当遇到超大型 excel 文件或需要精细控制单元格格式时，借助专用组件往往能提升开发效率。

本文基于轻量级 excel 处理库完成 excel 文件解析，结合 python 内置的 sqlite 数据库（无需独立部署），实现多工作表自动识别、动态创建表结构、批量数据入库的完整方案。

一、应用场景与方案优势

适用场景

企业 excel 报表数据迁移至数据库持久化存储；
自动化办公：定期将 excel 导出数据同步到数据库；
轻量级数据中台：多 excel 文件整合入库，方便后续查询分析； 4.测试数据构造：快速将 excel 测试数据导入数据库。

方案核心优势

无环境依赖：无需安装 microsoft office/wps，纯 python 库解析 excel；
多工作表适配：自动遍历 excel 所有 sheet，无需手动指定；
动态建表：根据 excel 表头自动生成数据库表结构；
安全稳定：参数化 sql 防注入，事务管理保证数据一致性；
轻量免费：适用于中小型 excel 文件处理，无额外成本。

二、环境准备

仅需安装 excel 解析库（free spire.xls for python），sqlite 为 python 内置库，无需额外安装：

pip install freespire.xls

三、核心执行流程

整个程序分为 5 个核心步骤，数据流转清晰无冗余：

加载excel文件 → 连接数据库 → 遍历工作表 → 读取表头+动态建表 → 逐行数据插入 → 提交事务+释放资源

3.1 完整代码

from spire.xls import workbook
import sqlite3

def excel_to_sqlite(excel_path, db_path):
    # 1. 加载 excel 文件
    workbook = workbook()
    workbook.loadfromfile(excel_path)

    # 2. 连接数据库
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # 3. 遍历每个工作表
    for sheet_index in range(workbook.worksheets.count):
        sheet = workbook.worksheets.get_item(sheet_index)
        sheet_name = sheet.name.replace(" ", "")          # 表名中去掉空格

        # 4. 读取表头（第一行）
        header = []
        for col in range(sheet.allocatedrange.columncount):
            raw_value = sheet.range[1, col + 1].value
            # 字段名中去掉空格，并避免空字段
            field_name = raw_value.replace(" ", "") if raw_value else f"col_{col}"
            header.append(field_name)

        # 5. 创建数据库表（所有字段暂定为 text 类型）
        create_sql = f"""
        create table if not exists {sheet_name} (
            {', '.join([f'[{h}] text' for h in header])}
        )
        """
        cursor.execute(create_sql)

        # 6. 逐行插入数据（跳过表头行）
        for row in range(1, sheet.allocatedrange.rowcount):  # row=1 对应 excel 第二行
            row_data = []
            for col in range(sheet.allocatedrange.columncount):
                cell_value = sheet.range[row + 1, col + 1].value
                row_data.append(cell_value)
            
            # 使用参数化查询防止 sql 注入
            placeholders = ','.join(['?' for _ in row_data])
            insert_sql = f"insert into {sheet_name} ({','.join(header)}) values ({placeholders})"
            cursor.execute(insert_sql, row_data)

    # 7. 提交并清理
    conn.commit()
    conn.close()
    workbook.dispose()

if __name__ == "__main__":
    excel_to_sqlite("sample.xlsx", "output/report.db")

3.2 关键点解析

1. 工作表遍历与表名清洗

workbook.worksheets.count 获取工作表总数，get_item(s) 按索引获取。工作表名称可能包含空格、特殊字符，直接用作 sqlite 表名会导致语法错误，因此使用 .replace(" ", "") 去除空格。更严谨的做法可增加正则过滤，只保留字母数字和下划线。

2. 动态建表与字段类型

示例将所有字段定义为 text 类型，适配 excel 中字符串、数字、日期等通用格式（可根据业务修改数据类型）。

3. 数据读取的范围

sheet.allocatedrange 返回已使用的单元格区域（包含数据的最大矩形），比直接遍历全部行列更高效。注意 rowcount 和 columncount 是基于 1 的计数。

4. 参数化插入

使用 ? 占位符配合 cursor.execute(insert_sql, row_data) 能自动处理字符串转义，避免因 excel 单元格内容包含单引号导致的 sql 错误，同时防范注入风险。

四、扩展：适配其他数据库

只需修改数据库连接部分，即可迁移到 mysql、postgresql 等。注意不同数据库的标识符引用符不同（mysql 用反引号 `，postgresql 用双引号 "），以及字段类型映射的差异。例如连接 mysql：

import pymysql
conn = pymysql.connect(host='localhost', user='root', password='123456', db='test')
cursor = conn.cursor()
# 建表时将 [field] 改为 `field`

五、注意事项与最佳实践

检查 excel 列名与数据库表字段名是否匹配。to_sql 默认使用 dataframe 的列名作为数据库字段名，如果数据库表已存在且字段名不同，要么重命名 dataframe 列，要么设置 if_exists='replace' 重建表。
处理超大 excel 文件：pd.read_excel 会将整个文件加载到内存，如果 excel 文件超过几百 mb，可以改用 openpyxl 的只读模式或分块读取。但更推荐将大 excel 拆分成多个小文件，或者使用 dask。
事务处理：to_sql 默认在自动提交模式下运行，如果中途出错，已写入的数据不会回滚。要保证原子性，可以手动管理事务（使用 engine.begin() 或连接对象的 begin()）。
性能对比：对于十万行以下的数据，to_sql 默认方式已经足够；对于百万行级别，建议使用 chunksize=5000 并配合数据库的批量提交优化。
日志与错误处理：建议用 try...except 捕获异常，并记录失败的行或文件位置。

六、知识扩展

1.使用 executemany 手动批量插入

如果你需要更精细的控制（例如在插入前做复杂转换），也可以手动使用 cursor.executemany：

import pymysql
import pandas as pd
df = pd.read_excel('data.xlsx')
records = df.to_dict('records')  # 转换为字典列表
conn = pymysql.connect(host='localhost', user='root', password='123456', database='test')
cursor = conn.cursor()
sql = "insert into table_name (col1, col2) values (%s, %s)"
values = [(r['col1'], r['col2']) for r in records]
cursor.executemany(sql, values)
conn.commit()
cursor.close()
conn.close()

这种方式和 to_sql 的 chunksize 本质类似，但你可以自定义 sql 语句。

2.从 excel 批量导入 mysql

import pandas as pd
from sqlalchemy import create_engine
# 读取 excel
df = pd.read_excel('sales.xlsx')
# 确保日期列为 datetime
df['sale_date'] = pd.to_datetime(df['sale_date'])
# 连接 mysql
engine = create_engine('mysql+pymysql://root:123456@localhost:3306/testdb')
# 写入数据库（如果表存在则替换）
df.to_sql('sales', engine, if_exists='replace', index=false)
print("数据导入完成！")