当前位置： 首页 > news >正文

Python实战：用pandas+SQLAlchemy高效导入Excel到MySQL

news 2026/7/16 11:35:23

1. 项目概述

最近接手了一个数据迁移项目，需要将大量Excel表格中的业务数据导入到MySQL数据库中。作为Python老手，我决定用pandas+SQLAlchemy这套黄金组合来完成这个任务。这种场景在实际工作中非常常见，比如财务系统数据对接、CRM客户资料迁移、电商订单批量导入等场景。

相比手动复制粘贴或使用Excel自带的导出功能，用Python脚本处理有明显优势：可以自动处理几十上百个文件、支持复杂的数据清洗逻辑、能够记录完整的导入日志，最重要的是可以复用代码。下面我就把这次实战中的完整方案和踩坑经验分享给大家。

2. 核心工具选型与配置

2.1 Python库的选择

经过对比测试，我最终确定了以下工具链：

  • pandas：数据读取和预处理的核心库，特别适合表格类数据处理
  • openpyxl/xlrd：Excel文件读取引擎（注意xlrd已停止维护xlsx格式支持）
  • SQLAlchemy：数据库ORM工具，支持多种数据库方言
  • pymysql：MySQL的Python驱动

安装命令如下：

pip install pandas sqlalchemy pymysql openpyxl

注意：如果处理xls格式文件需要额外安装xlrd，但要注意版本不要超过2.0.0，因为新版移除了对Excel的支持

2.2 数据库连接配置

使用SQLAlchemy创建数据库连接字符串的规范格式：

from sqlalchemy import create_engine # 基本格式 db_url = "mysql+pymysql://用户名:密码@服务器地址:端口/数据库名?charset=utf8mb4" # 实际示例 engine = create_engine( "mysql+pymysql://admin:Pass1234@192.168.1.100:3306/business_data", pool_size=5, # 连接池大小 max_overflow=10, # 最大溢出连接数 pool_timeout=30, # 获取连接超时时间(秒) echo=False # 是否输出SQL日志 )

3. Excel数据读取与预处理

3.1 基础读取方法

pandas读取Excel的核心方法：

import pandas as pd # 读取单个sheet df = pd.read_excel("sales_data.xlsx", sheet_name="2023订单") # 读取整个文件所有sheet all_sheets = pd.read_excel("multi_sheet.xlsx", sheet_name=None) # 带特殊参数读取 df = pd.read_excel( "complex_data.xlsx", header=3, # 从第4行开始读取 usecols="B:E,H", # 只读取B-E列和H列 dtype={"订单号": str, "金额": float}, # 指定列数据类型 na_values=["NA", "NULL"] # 自定义空值标识 )

3.2 常见数据清洗操作

实际业务数据往往需要清洗后才能入库：

# 处理空值 df["客户名称"].fillna("未知客户", inplace=True) # 日期格式标准化 df["下单时间"] = pd.to_datetime(df["下单时间"], format="%Y年%m月%d日") # 金额清洗 df["金额"] = df["金额"].str.replace("¥", "").str.replace(",", "").astype(float) # 去重处理 df.drop_duplicates(subset=["订单号"], keep="last", inplace=True) # 列名规范化 df.columns = df.columns.str.strip().str.lower().str.replace(" ", "_")

4. 数据库写入方案

4.1 基础写入方法

最简单的全量写入方式：

df.to_sql( name="sales_records", # 表名 con=engine, # 数据库连接 if_exists="append", # 存在表时追加数据 index=False, # 不写入索引列 chunksize=1000 # 分批写入 )

4.2 高级写入策略

对于大数据量或需要更新的场景：

from sqlalchemy import text # 分批写入+事务控制 with engine.begin() as conn: for chunk in pd.read_excel("large_file.xlsx", chunksize=5000): chunk.to_sql("big_table", conn, if_exists="append", index=False) # 存在则更新策略 def upsert(df, table_name, engine, match_columns): temp_table = f"temp_{table_name}" df.to_sql(temp_table, engine, if_exists="replace", index=False) columns = list(df.columns) update_set = ", ".join([f"{c}=VALUES({c})" for c in columns if c not in match_columns]) with engine.begin() as conn: conn.execute(text( f"INSERT INTO {table_name} ({', '.join(columns)}) " f"SELECT {', '.join(columns)} FROM {temp_table} " f"ON DUPLICATE KEY UPDATE {update_set}" )) conn.execute(text(f"DROP TABLE IF EXISTS {temp_table}")) # 使用示例 upsert(df, "products", engine, ["product_id"])

5. 完整实战案例

5.1 场景描述

假设我们需要将市场部门的"客户调研数据.xlsx"导入到数据库，该文件有以下特点：

  • 包含3个sheet：基础信息、消费记录、满意度评价
  • 数据分布在非标准位置（从第5行开始）
  • 包含合并单元格和空行
  • 需要关联已有数据库中的客户ID

5.2 实现代码

import pandas as pd from sqlalchemy import create_engine from datetime import datetime def import_survey_data(file_path): # 1. 建立数据库连接 engine = create_engine("mysql+pymysql://user:password@localhost/survey_db") # 2. 读取各sheet数据 base_info = pd.read_excel( file_path, sheet_name="基础信息", header=4, # 从第5行开始读取 usecols="B:F,I" ) purchase_records = pd.read_excel( file_path, sheet_name="消费记录", header=4, converters={"消费日期": lambda x: datetime.strptime(x, "%Y-%m-%d")} ) # 3. 数据清洗 base_info.rename(columns={ "客户名称": "name", "联系电话": "phone", "所在地区": "region" }, inplace=True) # 4. 关联已有客户ID with engine.connect() as conn: existing_clients = pd.read_sql( "SELECT client_id, phone FROM clients", conn ) merged = pd.merge( base_info, existing_clients, how="left", on="phone" ) # 5. 写入数据库 merged.to_sql( "survey_base_info", engine, if_exists="append", index=False ) # 6. 记录日志 with engine.begin() as conn: conn.execute( "INSERT INTO import_logs (file_name, record_count, import_time) " "VALUES (%s, %s, %s)", (file_path, len(merged), datetime.now()) ) if __name__ == "__main__": import_survey_data("客户调研数据.xlsx")

6. 性能优化技巧

6.1 大数据量处理方案

当处理超过50万行的Excel文件时：

  1. 使用chunksize参数分块读取
chunk_iter = pd.read_excel("huge_file.xlsx", chunksize=10000) for chunk in chunk_iter: process_chunk(chunk)
  1. 禁用DataFrame类型推断
df = pd.read_excel("data.xlsx", dtype=str) # 全部按字符串读取
  1. 使用低内存模式
df = pd.read_excel( "data.xlsx", engine="openpyxl", read_only=True )

6.2 数据库写入优化

  1. 临时禁用索引和约束
with engine.begin() as conn: conn.execute("ALTER TABLE target_table DISABLE KEYS") df.to_sql("target_table", conn, index=False) conn.execute("ALTER TABLE target_table ENABLE KEYS")
  1. 使用LOAD DATA INFILE替代INSERT（需要文件权限）
df.to_csv("temp.csv", index=False) with engine.begin() as conn: conn.execute(""" LOAD DATA LOCAL INFILE 'temp.csv' INTO TABLE target_table FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' IGNORE 1 ROWS """)

7. 常见问题与解决方案

7.1 编码问题

典型错误：

  • 中文乱码
  • 特殊字符报错

解决方案：

  1. 确保数据库使用utf8mb4字符集
  2. 在连接字符串中添加charset参数
engine = create_engine( "mysql+pymysql://user:pass@host/db?charset=utf8mb4" )

7.2 日期格式问题

常见场景：

  • Excel中的日期被读取为数字
  • 时区不一致

处理方法：

# 显式指定日期列 df = pd.read_excel( "data.xlsx", parse_dates=["order_date", "delivery_date"], date_parser=lambda x: pd.to_datetime(x, format="%d/%m/%Y") ) # 时区处理 df["timestamp"] = df["timestamp"].dt.tz_localize("Asia/Shanghai")

7.3 内存不足问题

症状：

  • 读取大文件时程序崩溃
  • 系统内存占用飙升

应对策略：

  1. 使用chunksize分块处理
  2. 指定dtype减少内存占用
dtype_map = { "id": "int32", "price": "float32", "description": "category" } df = pd.read_excel("data.xlsx", dtype=dtype_map)

8. 扩展应用场景

8.1 定时自动化导入

结合Windows任务计划或Linux cron实现自动导入：

# auto_import.py import schedule import time def job(): try: import_survey_data("/data/share/new_survey.xlsx") log_success() except Exception as e: send_alert_email(str(e)) # 每天凌晨2点执行 schedule.every().day.at("02:00").do(job) while True: schedule.run_pending() time.sleep(60)

8.2 多文件批量处理

处理目录下所有Excel文件：

from pathlib import Path def batch_import(folder_path): excel_files = Path(folder_path).glob("*.xlsx") for file in excel_files: try: print(f"Processing {file.name}...") import_survey_data(str(file)) file.rename(file.with_suffix(".done")) # 标记已处理 except Exception as e: print(f"Failed to process {file.name}: {str(e)}") file.rename(file.with_suffix(".error"))

8.3 数据验证与报告生成

导入后自动生成数据质量报告：

def generate_quality_report(df, table_name): report = { "total_records": len(df), "missing_values": df.isnull().sum().to_dict(), "duplicates": df.duplicated().sum(), "data_types": df.dtypes.astype(str).to_dict() } with engine.begin() as conn: conn.execute( "INSERT INTO quality_reports VALUES (%s, %s, %s, %s)", (table_name, datetime.now(), str(report), "pending") )

在实际项目中，我发现在写入前添加数据校验步骤可以避免80%的后期数据修复工作。比如检查必填字段、金额范围、日期合理性等，这些验证规则应该根据具体业务需求定制。

http://www.cnnetsun.cn/news/3422293.html

相关文章：

  • 基于通义千问与Live2D的智能数字人开发实践
  • AI 应用的命门：你的「产物」长什么样
  • 平板电脑 SRRC 认证实操经验：材料准备避坑与申请全流程
  • Linux进程管理与计划任务核心命令详解
  • Windows窗口置顶神器：AlwaysOnTop完整使用指南，免费提升工作效率300%
  • RHEL 9.6架构解析与性能优化实践
  • 免费开源表情包神器：三步打造你的二次元专属贴纸
  • 抖音无水印下载终极指南：三步完成批量下载与智能管理
  • 会议记录整理慢写不完？智能会议记录生成器帮你高效产出规范纪要
  • AI原型工具企业需求分析：私有化部署与安全协作选购指南
  • 5分钟掌握：鸣潮自动化工具ok-ww终极指南
  • 公考机构捆绑销售陷阱
  • JX3Toy：解放双手的剑网3智能宏工具终极指南
  • Linux进程管理工具与性能优化实战
  • 如何快速提升游戏性能：SGuard限制器终极使用指南
  • 【办公自动化本地 AI 工具】 OpenClaw 完整部署与指令案例分享（含安装包）
  • Linux系统调用与errno错误处理机制详解
  • 调岗调拨资产信息滞后？四大主流资产系统变更能力拆解
  • Windows服务开发与管理全指南：从原理到实践
  • ESD保护电路设计：原理、选型与PCB布局实战
  • 图胜千言·懒人版
  • 告别 C 盘红条！完整图文教你把安装程序转移到 D/E 盘
  • Windows API哈希扫描技术原理与实现
  • AI编程工作流真相：四大IDE底层认知维度解析
  • Cherry Studio本地部署避坑指南：AI知识库安装与向量化实战
  • Java异常体系精讲：一文吃透九大常见异常成因、场景与代码实战
  • 45-代码审查工作流
  • Angry IP Scanner：3分钟掌握专业网络探测的5个核心价值
  • 51单片机电子时钟设计与实现
  • HsMod：基于BepInEx的炉石传说终极增强插件完整技术指南