1. 项目背景与需求解析
在日常数据处理工作中，我们经常需要将数据库中的大量记录导出到Excel文件中进行分析或共享。手动逐条导出不仅效率低下，而且容易出错。这个Python脚本就是为了解决这个痛点而设计的。
我最近接手了一个客户项目，需要每周从MySQL数据库导出近万条销售记录到Excel报表。最初尝试用Navicat等GUI工具手动导出，发现存在三个明显问题：一是导出速度慢，二是字段映射容易出错，三是无法灵活处理数据转换。于是决定用Python开发一个自动化解决方案。
2. 技术方案选型
2.1 核心组件对比
经过评估，最终选择了以下技术组合：
- 数据库连接：SQLAlchemy + PyMySQL
- Excel操作：openpyxl库
- 异步处理：concurrent.futures.ThreadPoolExecutor
为什么不选其他方案？
- 相比pymysql直接连接，SQLAlchemy提供了更好的ORM支持和连接池管理
- 测试发现openpyxl在大数据量导出时比xlwt/xlrd更稳定
- 线程池比多进程更轻量，适合IO密集型任务
2.2 架构设计
脚本采用生产者-消费者模式：
- 主线程从数据库分页查询数据
- 工作线程负责Excel写入
- 通过队列实现数据传递
python复制from concurrent.futures import ThreadPoolExecutor
from queue import Queue
data_queue = Queue(maxsize=1000)
def producer(page_size=500):
# 数据库分页查询逻辑
pass
def consumer():
# Excel写入逻辑
pass
with ThreadPoolExecutor(max_workers=4) as executor:
executor.submit(producer)
executor.submit(consumer)
3. 核心实现细节
3.1 数据库连接优化
使用连接池避免频繁创建连接：
python复制from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
engine = create_engine(
'mysql+pymysql://user:pass@host/db',
poolclass=QueuePool,
pool_size=5,
max_overflow=10,
pool_recycle=3600
)
3.2 Excel写入技巧
- 批量写入：每积累1000条记录写入一次
- 样式预定义：提前创建单元格样式对象
- 内存优化：使用openpyxl的write_only模式
python复制from openpyxl import Workbook
from openpyxl.styles import Font, Alignment
wb = Workbook(write_only=True)
ws = wb.create_sheet()
# 预定义样式
header_font = Font(bold=True)
cell_alignment = Alignment(horizontal='center')
# 批量写入数据
for row in data_chunk:
ws.append(row)
3.3 数据类型处理
数据库与Excel类型映射表：
| 数据库类型 | Python类型 | Excel处理方式 |
|---|---|---|
| DATETIME | datetime | 转换为字符串 |
| DECIMAL | Decimal | float()转换 |
| TEXT | str | 直接写入 |
| BLOB | bytes | Base64编码 |
4. 性能优化实战
4.1 分页查询策略
采用游标分页代替LIMIT OFFSET：
python复制def get_records(last_id=0, batch_size=500):
return session.execute(
"SELECT * FROM orders WHERE id > :last_id ORDER BY id LIMIT :limit",
{'last_id': last_id, 'limit': batch_size}
).fetchall()
4.2 内存监控
添加内存使用监控逻辑：
python复制import psutil
def check_memory():
if psutil.virtual_memory().percent > 90:
logging.warning("Memory usage over 90%")
return False
return True
5. 异常处理与日志
5.1 错误重试机制
实现指数退避重试：
python复制from time import sleep
def query_with_retry(query, max_retries=3):
for attempt in range(max_retries):
try:
return session.execute(query)
except Exception as e:
wait = 2 ** attempt
sleep(wait)
continue
raise Exception("Max retries exceeded")
5.2 日志配置
python复制import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('export.log'),
logging.StreamHandler()
]
)
6. 完整代码结构
code复制db_to_excel/
├── __init__.py
├── config.py # 配置文件
├── connectors.py # 数据库连接
├── excel_writer.py # Excel操作
├── main.py # 主程序
└── utils.py # 工具函数
7. 实际应用案例
7.1 电商订单导出
配置示例：
python复制{
"database": {
"dialect": "mysql",
"host": "127.0.0.1",
"port": 3306,
"dbname": "ecommerce",
"table": "orders",
"columns": ["id", "order_no", "amount", "create_time"]
},
"excel": {
"output_path": "/reports/",
"filename_prefix": "orders_export_",
"batch_size": 2000
}
}
7.2 导出结果示例
生成的文件包含：
- 自动添加的表头
- 格式化的日期字段
- 金额千分位分隔
- 冻结首行方便查看
8. 性能对比测试
测试环境：
- 数据库：MySQL 8.0，100万条测试数据
- 硬件：4核CPU/8GB内存
| 导出方式 | 10万条耗时 | CPU占用 | 内存峰值 |
|---|---|---|---|
| 本方案 | 42s | 75% | 1.2GB |
| 原生导出 | 3m28s | 35% | 2.8GB |
| ORM导出 | 2m15s | 90% | 1.8GB |
9. 常见问题解决
9.1 内存溢出处理
症状：导出大文件时Python进程被终止
解决方案：
- 减小batch_size参数
- 使用--memory-limit启动参数
- 考虑分多个文件输出
9.2 编码问题
错误：UnicodeEncodeError
处理方法：
python复制# 在连接字符串中添加charset参数
'mysql+pymysql://user:pass@host/db?charset=utf8mb4'
# Excel写入时指定编码
with open(filename, 'w', encoding='utf-8-sig') as f:
wb.save(f)
9.3 日期格式混乱
解决方法：
python复制from datetime import datetime
def format_datetime(value):
if isinstance(value, datetime):
return value.strftime('%Y-%m-%d %H:%M:%S')
return value
10. 进阶优化方向
- 增量导出：记录上次导出位置，只导出新增数据
- 分布式导出：使用Celery实现多机并行
- Web界面：集成Flask提供可视化操作
- 自动邮件发送：导出完成后发送邮件通知
关键提示：在实际项目中，建议先导出小批量数据验证格式正确性，再处理全量数据。我曾遇到过因为一个字段类型不匹配导致导出5小时后才报错的情况。
这个脚本已经在我们生产环境稳定运行了8个月，每周自动处理超过50万条记录。最大的收获是：一定要做好字段映射的验证和异常情况的处理，数据导出类工具一旦出错往往会造成严重后果。