Excel 才十几万行，导入接口还没跑完，容器内存先冲上去了。
这种问题我一般不先查数据库，先看代码是不是把整张 Excel 一次性塞进了内存。尤其是直接用 pandas.read_excel()，写起来确实省事，但文件一大，DataFrame、类型转换、中间对象会一起吃内存。
EasyExcel 解决的正是这个问题。
不过有个地方得先纠正：EasyExcel 是阿里开源的 Java 组件，Python 项目不能直接安装使用。它的核心思路倒不复杂：流式读取 Excel，每读到一行就处理一行，攒够一批再写数据库。
在 Python 里，我一般用 openpyxl 的只读模式，照着这个思路做。
假设现在要导入一份商品表，Excel 里有商品编码、商品名称、价格和库存四列。数据库中的商品编码有唯一索引，重复导入时执行更新。
先把 Excel 按行读出来，不要一次性加载整张表。
from decimal import Decimal, InvalidOperation
from openpyxl import load_workbook
REQUIRED_COLUMNS = ("商品编码", "商品名称", "价格", "库存")
defiter_goods(file_path: str):
workbook = load_workbook(
file_path,
read_only=True,
data_only=True,
)
sheet = workbook.active
rows = sheet.iter_rows(values_only=True)
try:
header = next(rows)
column_index = {
str(name).strip(): index
for index, name in enumerate(header)
if name isnotNone
}
missing = [
name for name in REQUIRED_COLUMNS
if name notin column_index
]
if missing:
raise ValueError(f"Excel 缺少字段：{','.join(missing)}")
for line_no, row in enumerate(rows, start=2):
ifnot any(value isnotNonefor value in row):
continue
yield line_no, {
name: row[column_index[name]]
for name in REQUIRED_COLUMNS
}
finally:
workbook.close()
这里我特意用了 read_only=True。
普通模式会把单元格对象留在内存里，数据量一上来，内存占用很难看。只读模式更接近 EasyExcel 的流式读取方式，代码拿到的是当前行，不需要把整个工作簿抱在内存里。
读取只是第一步。批量导入最容易出问题的地方，其实是脏数据。
价格里混进“12元”，库存填成“十个”，商品编码前后带空格，这些数据直接扔给数据库，最后通常是一整批一起失败。
我会在入库前先清洗。
defnormalize_goods(line_no: int, row: dict) -> tuple:
goods_code = str(row["商品编码"] or"").strip()
goods_name = str(row["商品名称"] or"").strip()
ifnot goods_code:
raise ValueError(f"第 {line_no} 行商品编码为空")
ifnot goods_name:
raise ValueError(f"第 {line_no} 行商品名称为空")
try:
price = Decimal(str(row["价格"])).quantize(Decimal("0.01"))
stock = int(row["库存"])
except (InvalidOperation, TypeError, ValueError):
raise ValueError(f"第 {line_no} 行价格或库存格式错误")
if price < 0or stock < 0:
raise ValueError(f"第 {line_no} 行价格或库存不能为负数")
return goods_code, goods_name, price, stock
数据库写入不要一行提交一次。
一万行数据就提交一万次事务，这种代码看着老实，实际全耗在网络往返和事务提交上了。我通常按几百行一批处理，具体批次大小再根据字段数量和数据库连接情况调整。
import logging
import pymysql
INSERT_SQL = """
INSERT INTO goods(goods_code, goods_name, price, stock)
VALUES (%s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
goods_name = VALUES(goods_name),
price = VALUES(price),
stock = VALUES(stock)
"""
defsave_batch(connection, batch: list[tuple]) -> None:
try:
with connection.cursor() as cursor:
cursor.executemany(INSERT_SQL, batch)
connection.commit()
except Exception:
connection.rollback()
raise
defimport_goods(file_path: str) -> dict:
connection = pymysql.connect(
host="127.0.0.1",
port=3306,
user="app_user",
password="change_me",
database="shop",
charset="utf8mb4",
)
batch = []
errors = []
success_count = 0
batch_size = 500
try:
for line_no, row in iter_goods(file_path):
try:
batch.append(normalize_goods(line_no, row))
except ValueError as exc:
errors.append(str(exc))
continue
if len(batch) < batch_size:
continue
save_batch(connection, batch)
success_count += len(batch)
logging.info(
"excel_import committed=%s bad_rows=%s",
success_count,
len(errors),
)
batch.clear()
if batch:
save_batch(connection, batch)
success_count += len(batch)
return {
"success": success_count,
"failed": len(errors),
"errors": errors,
}
finally:
connection.close()
这段代码没有把整个导入任务包在一个大事务里。
几万行数据放进同一个事务，看着能保证“全部成功或者全部失败”，但事务跑几分钟，锁和连接一直不释放，线上并发请求很容易跟着排队。分批提交更稳，某一批失败时，前面已经成功的数据也不会白跑。
当然，分批提交以后要考虑重复执行。这里使用商品编码唯一索引，再配合 ON DUPLICATE KEY UPDATE，同一个文件重新上传不会插出重复商品。
这也是批量导入里经常被漏掉的一点：导入接口不能只考虑第一次成功，还得考虑用户点了两次、接口超时后重试、任务执行到一半重跑。
EasyExcel 真正好用的地方，不是 API 少写了几行，而是它把处理方式带对了：按行读取、监听回调、对象映射、批量处理，避免开发者顺手把整张表加载进内存。
换成 Python，组件名字变了，判断没变。
小文件可以直接用 Pandas，代码短，数据分析也方便。到了业务导入，文件可能有十万行，还要校验、记录错误行、分批提交数据库，我更愿意用 openpyxl 的只读模式自己控制流程。
导入任务最怕的不是慢一点，而是跑到百分之八十，内存没了，事务也回滚了，日志里还看不出来到底坏在哪一行。
流式读取、提前校验、分批提交、唯一键兜底，这几处处理好，Excel 导入基本就不会太难收拾。
