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

1. 项目概述最近接手了一个数据迁移项目需要将大量Excel表格中的业务数据导入到MySQL数据库中。作为Python老手我决定用pandasSQLAlchemy这套黄金组合来完成这个任务。这种场景在实际工作中非常常见比如财务系统数据对接、CRM客户资料迁移、电商订单批量导入等场景。相比手动复制粘贴或使用Excel自带的导出功能用Python脚本处理有明显优势可以自动处理几十上百个文件、支持复杂的数据清洗逻辑、能够记录完整的导入日志最重要的是可以复用代码。下面我就把这次实战中的完整方案和踩坑经验分享给大家。2. 核心工具选型与配置2.1 Python库的选择经过对比测试我最终确定了以下工具链pandas数据读取和预处理的核心库特别适合表格类数据处理openpyxl/xlrdExcel文件读取引擎注意xlrd已停止维护xlsx格式支持SQLAlchemy数据库ORM工具支持多种数据库方言pymysqlMySQL的Python驱动安装命令如下pip install pandas sqlalchemy pymysql openpyxl注意如果处理xls格式文件需要额外安装xlrd但要注意版本不要超过2.0.0因为新版移除了对Excel的支持2.2 数据库连接配置使用SQLAlchemy创建数据库连接字符串的规范格式from sqlalchemy import create_engine # 基本格式 db_url mysqlpymysql://用户名:密码服务器地址:端口/数据库名?charsetutf8mb4 # 实际示例 engine create_engine( mysqlpymysql://admin:Pass1234192.168.1.100:3306/business_data, pool_size5, # 连接池大小 max_overflow10, # 最大溢出连接数 pool_timeout30, # 获取连接超时时间(秒) echoFalse # 是否输出SQL日志 )3. Excel数据读取与预处理3.1 基础读取方法pandas读取Excel的核心方法import pandas as pd # 读取单个sheet df pd.read_excel(sales_data.xlsx, sheet_name2023订单) # 读取整个文件所有sheet all_sheets pd.read_excel(multi_sheet.xlsx, sheet_nameNone) # 带特殊参数读取 df pd.read_excel( complex_data.xlsx, header3, # 从第4行开始读取 usecolsB:E,H, # 只读取B-E列和H列 dtype{订单号: str, 金额: float}, # 指定列数据类型 na_values[NA, NULL] # 自定义空值标识 )3.2 常见数据清洗操作实际业务数据往往需要清洗后才能入库# 处理空值 df[客户名称].fillna(未知客户, inplaceTrue) # 日期格式标准化 df[下单时间] pd.to_datetime(df[下单时间], format%Y年%m月%d日) # 金额清洗 df[金额] df[金额].str.replace(¥, ).str.replace(,, ).astype(float) # 去重处理 df.drop_duplicates(subset[订单号], keeplast, inplaceTrue) # 列名规范化 df.columns df.columns.str.strip().str.lower().str.replace( , _)4. 数据库写入方案4.1 基础写入方法最简单的全量写入方式df.to_sql( namesales_records, # 表名 conengine, # 数据库连接 if_existsappend, # 存在表时追加数据 indexFalse, # 不写入索引列 chunksize1000 # 分批写入 )4.2 高级写入策略对于大数据量或需要更新的场景from sqlalchemy import text # 分批写入事务控制 with engine.begin() as conn: for chunk in pd.read_excel(large_file.xlsx, chunksize5000): chunk.to_sql(big_table, conn, if_existsappend, indexFalse) # 存在则更新策略 def upsert(df, table_name, engine, match_columns): temp_table ftemp_{table_name} df.to_sql(temp_table, engine, if_existsreplace, indexFalse) 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( fINSERT INTO {table_name} ({, .join(columns)}) fSELECT {, .join(columns)} FROM {temp_table} fON DUPLICATE KEY UPDATE {update_set} )) conn.execute(text(fDROP TABLE IF EXISTS {temp_table})) # 使用示例 upsert(df, products, engine, [product_id])5. 完整实战案例5.1 场景描述假设我们需要将市场部门的客户调研数据.xlsx导入到数据库该文件有以下特点包含3个sheet基础信息、消费记录、满意度评价数据分布在非标准位置从第5行开始包含合并单元格和空行需要关联已有数据库中的客户ID5.2 实现代码import pandas as pd from sqlalchemy import create_engine from datetime import datetime def import_survey_data(file_path): # 1. 建立数据库连接 engine create_engine(mysqlpymysql://user:passwordlocalhost/survey_db) # 2. 读取各sheet数据 base_info pd.read_excel( file_path, sheet_name基础信息, header4, # 从第5行开始读取 usecolsB:F,I ) purchase_records pd.read_excel( file_path, sheet_name消费记录, header4, converters{消费日期: lambda x: datetime.strptime(x, %Y-%m-%d)} ) # 3. 数据清洗 base_info.rename(columns{ 客户名称: name, 联系电话: phone, 所在地区: region }, inplaceTrue) # 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, howleft, onphone ) # 5. 写入数据库 merged.to_sql( survey_base_info, engine, if_existsappend, indexFalse ) # 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文件时使用chunksize参数分块读取chunk_iter pd.read_excel(huge_file.xlsx, chunksize10000) for chunk in chunk_iter: process_chunk(chunk)禁用DataFrame类型推断df pd.read_excel(data.xlsx, dtypestr) # 全部按字符串读取使用低内存模式df pd.read_excel( data.xlsx, engineopenpyxl, read_onlyTrue )6.2 数据库写入优化临时禁用索引和约束with engine.begin() as conn: conn.execute(ALTER TABLE target_table DISABLE KEYS) df.to_sql(target_table, conn, indexFalse) conn.execute(ALTER TABLE target_table ENABLE KEYS)使用LOAD DATA INFILE替代INSERT需要文件权限df.to_csv(temp.csv, indexFalse) 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 编码问题典型错误中文乱码特殊字符报错解决方案确保数据库使用utf8mb4字符集在连接字符串中添加charset参数engine create_engine( mysqlpymysql://user:passhost/db?charsetutf8mb4 )7.2 日期格式问题常见场景Excel中的日期被读取为数字时区不一致处理方法# 显式指定日期列 df pd.read_excel( data.xlsx, parse_dates[order_date, delivery_date], date_parserlambda x: pd.to_datetime(x, format%d/%m/%Y) ) # 时区处理 df[timestamp] df[timestamp].dt.tz_localize(Asia/Shanghai)7.3 内存不足问题症状读取大文件时程序崩溃系统内存占用飙升应对策略使用chunksize分块处理指定dtype减少内存占用dtype_map { id: int32, price: float32, description: category } df pd.read_excel(data.xlsx, dtypedtype_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(fProcessing {file.name}...) import_survey_data(str(file)) file.rename(file.with_suffix(.done)) # 标记已处理 except Exception as e: print(fFailed 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%的后期数据修复工作。比如检查必填字段、金额范围、日期合理性等这些验证规则应该根据具体业务需求定制。