批量把数据库数据交付成 Excel,手工复制在几百张表、几十万行时基本不可行。Python 适合这条链路:sqlalchemy 建连接,pandas 取数并整理 DataFrame,openpyxl 控制 Excel 的 Sheet、列宽、冻结行和表头样式。本文围绕“数据库到 Excel”的批量导出,拆解连接、查询、类型转换、写入、格式和排查。
一、适用场景与技术选型
业务方每月要订单明细,50 万行、50 多字段,还要按省份拆子表、附汇总 Sheet;或一次导出 12 个月,每月一个 Excel、内部多 Sheet。这类需求本质是流水线:连接数据库→按规则取数→整理 DataFrame→写 Excel→校验产物。Navicat、MySQL Workbench 能导出结果集,但按省份/日期/品类拆分、列宽、冻结窗格、多级表头、公式和自动调度较弱。商业 ETL 工具能做自动化但偏重。Python 脚本轻量、透明、可复用,能和定时任务、监控告警放在同一技术栈。
二、环境准备与连接串
核心库:pandas 负责数据处理和 to_excel,sqlalchemy 用 create_engine 做连接抽象,openpyxl 负责 Excel 底层读写和格式,PyMySQL/psycopg2/cx_Oracle 按数据库选择,tqdm 用于批量进度提示。安装:- pip install pandas openpyxl sqlalchemy pymysql tqdm
复制代码 常用 Python 3.10 到 3.12、pandas 2.x、openpyxl 3.1.x。pandas 2.x 对 openpyxl 的依赖行为有变化,从 1.x 升级后要重新验证写 Excel,尤其是 datetime 格式和列宽设置。建议虚拟环境,系统 Python 里旧版 openpyxl 可能触发 AttributeError: 'Workbook' object has no attribute 'add_chart' 一类版本冲突。
MySQL 连接串示例:- from sqlalchemy import create_engine
- engine = create_engine(
- 'mysql+pymysql://用户名:密码@主机:3306/数据库名?charset=utf8mb4'
- )
复制代码 密码含 @、#、% 等特殊字符时,用 urllib.parse.quote_plus 编码;端口不是 3306 要写明确;MySQL 显式加 charset=utf8mb4,否则中文或 emoji 可能乱码。长查询可加 connect_args={'connect_timeout': 30, 'read_timeout': 3600},避免中途掉线。
三、从数据库读数据
pd.read_sql_query 会一次性把结果加载到内存。几万行可以全量;几十万行且字段多时内存会很难看。处理方式一是用 chunksize 分批:- chunks = pd.read_sql_query(sql, engine, chunksize=20000)
- for chunk in chunks:
- process(chunk)
复制代码 chunksize 通过游标分批取数,可读一批写一批。方式二是把聚合、排序、类型转换尽量放进 SQL,Python 只接收结果。经验量级:10 万以下全量读;10 万到 100 万视字段和内存分批;超过 100 万优先导出 CSV 或按条件拆多个 Excel,因为 Excel 单 Sheet 上限是 1048576 行。
SQL 优化:只 select 需要的字段;分页不要用大偏移量 limit offset,改用 where id > 上次最大 id order by id limit 50000 的滚动键;导出期间可用只读事务或一致性快照;过滤字段建索引。
类型转换要统一处理:日期时间读进来通常是 datetime64[ns],可写标准日期;若 SQL 用 varchar 存时间,要在 Python 转一次,否则 Excel 把日期当文本。金额 decimal/numeric 转成 float64 会精度丢失,例如 0.1+0.2 的浮点误差,财务对账场景建议转字符串或 SQL 里 CAST(amount AS CHAR)。BIGINT 超过 15 位会被 Excel 显示成 1.23457E+14,导出前转字符串或设置单元格文本格式。
四、写入 Excel 与格式控制
pandas.to_excel 方便,但列宽、冻结表头、表头底色等控制弱。常用组合是 pandas 写数据,openpyxl 打开 writer.book 调格式。多 Sheet:- with pd.ExcelWriter('订单数据.xlsx', engine='openpyxl') as writer:
- for name, df in sheet_items:
- df.to_excel(writer, sheet_name=name, index=False)
复制代码 Sheet 名不要超过 31 个字符,不能包含斜杠、问号、星号、方括号等特殊字符,重复名要先判断 writer.sheets 并加后缀。一个 Sheet 先汇总后明细可用 startrow:- df_summary.to_excel(writer, sheet_name='数据汇总', startrow=0, index=False)
- df_detail.to_excel(writer, sheet_name='数据汇总', startrow=len(df_summary)+3, index=False)
复制代码 格式控制用 openpyxl:- from openpyxl.utils import get_column_letter
- wb = writer.book
- ws = writer.sheets['订单明细']
- for i, col in enumerate(df.columns, start=1):
- max_len = max(df[col].astype(str).map(len).max(), len(col)) + 2
- ws.column_dimensions[get_column_letter(i)].width = min(max_len, 40)
- ws.freeze_panes = 'A2'
- from openpyxl.styles import Font, PatternFill, Alignment
- for cell in ws[1]:
- cell.font = Font(bold=True)
- cell.fill = PatternFill(start_color='DDEBF7', end_color='DDEBF7', fill_type='solid')
- cell.alignment = Alignment(horizontal='center', vertical='center')
- ws.auto_filter.ref = ws.dimensions
复制代码 列宽不要只按字符数算,中文字符占两个字节,按长度×2估算再 clamp 到合理宽度。自动筛选让业务方可以直接下拉过滤。
五、完整批量导出代码
下面以 MySQL 为例,导出最近三个月订单数据,按月份拆 3 个文件,每个文件包含“每日汇总”和“订单明细”两个 Sheet。- import pandas as pd
- from sqlalchemy import create_engine
- from openpyxl.utils import get_column_letter
- from openpyxl.styles import Font, PatternFill, Alignment
- from pathlib import Path
- from tqdm import tqdm
- def get_engine():
- from urllib.parse import quote_plus
- user = 'your_user'
- password = quote_plus('your_pass_with_@special')
- host = '10.0.0.10'
- port = '3306'
- dbname = 'your_db'
- engine = create_engine(
- f'mysql+pymysql://{user}:{password}@{host}:{port}/{dbname}?charset=utf8mb4',
- connect_args={'connect_timeout': 30, 'read_timeout': 3600},
- pool_recycle=3600,
- )
- return engine
- def fetch_data(sql, engine):
- df = pd.read_sql_query(sql, engine)
- for col in df.columns:
- if 'id' in col.lower() or 'code' in col.lower():
- df[col] = df[col].astype(str)
- return df
- def write_excel_with_style(filepath, sheet_specs):
- with pd.ExcelWriter(filepath, engine='openpyxl') as writer:
- for spec in sheet_specs:
- df = spec['df']
- sheet_name = spec['name'][:31]
- df.to_excel(writer, sheet_name=sheet_name, index=False)
- ws = writer.sheets[sheet_name]
- for i, col in enumerate(df.columns, start=1):
- col_len = len(col) * 2
- max_len = max(
- df[col].astype(str).map(lambda x: sum(2 if ord(c) > 127 else 1 for c in x)).max(),
- col_len
- ) + 2
- ws.column_dimensions[get_column_letter(i)].width = min(max_len, 40)
- ws.freeze_panes = 'A2'
- for cell in ws[1]:
- cell.font = Font(bold=True)
- cell.fill = PatternFill(start_color='DDEBF7', end_color='DDEBF7', fill_type='solid')
- cell.alignment = Alignment(horizontal='center', vertical='center')
- ws.auto_filter.ref = ws.dimensions
- def run_export():
- engine = get_engine()
- export_dir = Path('./export')
- export_dir.mkdir(exist_ok=True)
- months = ['2024-01-01', '2024-02-01', '2024-03-01']
- for i in range(len(months) - 1):
- start = months[i]
- end = months[i + 1]
- year_month = start[:7]
- print(f'正在处理 {year_month} ...')
- sql_detail = f'''
- SELECT order_id, user_id, product_id, amount, status, create_time
- FROM order_table
- WHERE create_time >= '{start}' AND create_time < '{end}'
- ORDER BY create_time
- '''
- df_detail = fetch_data(sql_detail, engine)
- if df_detail.empty:
- print(f'{year_month} 无数据,跳过')
- continue
- sql_summary = f'''
- SELECT DATE(create_time) AS day, COUNT(*) AS order_cnt, SUM(amount) AS total_amount
- FROM order_table
- WHERE create_time >= '{start}' AND create_time < '{end}'
- GROUP BY DATE(create_time)
- '''
- df_summary = fetch_data(sql_summary, engine)
- filepath = export_dir / f'订单数据_{year_month}.xlsx'
- write_excel_with_style(filepath, [
- {'name': '每日汇总', 'df': df_summary},
- {'name': '订单明细', 'df': df_detail},
- ])
- print(f'已生成 {filepath}')
- if __name__ == '__main__':
- run_export()
复制代码 这段代码把 SQL 当成配置:调整日期范围、字段、汇总逻辑,改 SQL 和循环列表即可。运行后 export 目录生成订单数据_2024-01.xlsx 等文件,“每日汇总”在前,“订单明细”在后,冻结首行、表头底色、列筛选都已加上。实际跑几十万行、20 个字段时,时间主要在数据库查询和 Excel 写入。超大 Sheet 可考虑 xlsxwriter,写入更快,但格式 API 不同;openpyxl 的后期样式调整更灵活。
六、常见问题与排查
1. 内存溢出与超时。症状是进程被杀或数据库报 Lock wait timeout exceeded。先用 EXPLAIN 看 SQL 是否走索引;单表大就缩小范围,按小时或 ID 段分批;在读取前、读取后、写 Excel 前打印行数和内存日志。循环跑几十张表时,用完的 DataFrame 及时 del,必要时 gc.collect()。
2. 特殊字符与时间格式。字符串里有换行符时,写进 Excel 会在单元格内显示多行,若不需要可清洗掉换行和回车。日期 pandas 读出来是 2024-01-01 00:00:00,Excel 可能显示成 2024/1/1 0:00 或数字。写 Excel 前用 dt.strftime('%Y-%m-%d') 转字符串,显示稳定;若要真日期,用 openpyxl 的 number_format 设为 yyyy-mm-dd hh:mm:ss。单个单元格最多 32767 个字符,长文本要预处理或提示。
3. 连接中断与权限。长批量导出容易断开,可加重试机制、查询失败等待几秒重试三次,连接串加 pool_pre_ping=True 让 sqlalchemy 取连接前检查有效性。权限报错常见 Access denied for user 或 1044,需要 DBA 开权限。生产环境导出尽量走只读副本或从库,避免压主库。排查时先看报错栈在 SQLAlchemy 层还是 PyMySQL 层,连接问题常在 pymysql.err,权限问题常在 sqlalchemy.exc.OperationalError。
最后,每次导出前先写一条 SQL 统计各表行数,再和 Excel 行数比对。这个校验只有几行代码,但能避免“文件生成了但少了一半数据”。数据交付信任建立很慢,崩塌只要一次。建议先从单表导出跑通,再叠加格式、拆分和多 Sheet。 |