1. 项目背景与需求场景在日常数据处理工作中我们经常需要将数据库中的大量记录导出到Excel文件进行二次处理或分发。作为数据工程师我每周都要处理几十次这样的需求市场部门需要客户数据做分析、财务部门需要交易记录对账、运营团队需要用户行为数据生成报表...传统的手工操作方式存在明显痛点通过数据库客户端工具导出时每次都需要重复设置查询条件和导出参数数据量超过百万行时GUI工具经常卡死或崩溃需要定期执行的导出任务无法自动化不同数据库系统的导出操作差异大学习成本高Python正好能完美解决这些问题。最近我用PyMySQLopenpyxl组合实现了一套自动化导出方案单脚本可处理MySQL百万级数据导出还能自动拆分Excel文件避免超过104万行限制。下面分享具体实现方法和踩坑经验。2. 技术方案选型2.1 数据库连接方案比较对于Python连接数据库主流有几种方案DB-API标准接口优点标准化接口代码可移植性强缺点需要针对不同数据库安装特定驱动代表库PyMySQL(MySQL)、psycopg2(PostgreSQL)、cx_Oracle(Oracle)ORM框架优点面向对象操作自动防SQL注入缺点性能损耗学习曲线陡峭代表库SQLAlchemy、DjangoORM专用连接器优点厂商官方支持功能完整缺点依赖特定数据库代表mysql-connector-python提示对于纯导出场景推荐使用DB-API方案。ORM在简单查询场景会产生15-20%的性能开销2.2 Excel操作库选型处理Excel文件的Python库主要有库名称读写支持大文件处理公式支持样式调整适用场景openpyxl读写一般完善完善需要修改样式的情况xlsxwriter只写优秀基础完善大数据量导出pandas读写优秀无有限数据分析场景pyxlsb读写优秀无无处理二进制xlsb实测百万行数据导出openpyxl耗时约210秒内存占用1.2GBxlsxwriter耗时约95秒内存占用300MB3. 完整实现方案3.1 基础版本代码import pymysql from openpyxl import Workbook def export_to_excel(host, user, password, db, sql, output_path): # 建立数据库连接 connection pymysql.connect( hosthost, useruser, passwordpassword, databasedb, cursorclasspymysql.cursors.DictCursor ) try: with connection.cursor() as cursor: print(Executing query...) cursor.execute(sql) # 创建Excel工作簿 wb Workbook() ws wb.active # 写入表头 if cursor.description: headers [desc[0] for desc in cursor.description] ws.append(headers) # 分批写入数据 batch_size 10000 while True: rows cursor.fetchmany(batch_size) if not rows: break for row in rows: ws.append(list(row.values())) print(fProcessed {len(rows)} rows) # 保存文件 wb.save(output_path) print(fFile saved to {output_path}) finally: connection.close()3.2 生产环境增强版实际使用时需要考虑更多因素内存优化- 使用生成器分批处理def batch_fetch(cursor, size10000): while True: rows cursor.fetchmany(size) if not rows: break yield rows多Sheet支持- 避免Excel行数限制MAX_ROWS_PER_SHEET 1000000 # Excel限制 sheet_count 1 current_row 0 ws wb.create_sheet(fData_{sheet_count}) for batch in batch_fetch(cursor): for row in batch: if current_row MAX_ROWS_PER_SHEET: sheet_count 1 current_row 0 ws wb.create_sheet(fData_{sheet_count}) ws.append(headers) ws.append(list(row.values())) current_row 1类型处理- 处理datetime等特殊类型from datetime import datetime def format_value(value): if isinstance(value, datetime): return value.strftime(%Y-%m-%d %H:%M:%S) return str(value) if value is not None else 4. 性能优化技巧4.1 数据库层面优化使用SS游标(Server Side Cursor)connection pymysql.connect( ..., cursorclasspymysql.cursors.SSCursor )添加查询超时设置cursor.execute(SET SESSION max_execution_time300000) # 5分钟超时只查询必要字段避免SELECT *明确列出所需字段4.2 Excel写入优化禁用openpyxl自动计算wb Workbook(write_onlyTrue)使用xlsxwriter的常量内存模式import xlsxwriter workbook xlsxwriter.Workbook( large.xlsx, {constant_memory: True} )关闭自动过滤worksheet.autofilter False5. 常见问题与解决方案5.1 内存溢出问题现象处理大数据量时Python进程被Killed解决方案使用SSCursor游标减小batch_size(建议5000-10000)换用xlsxwriter库5.2 中文乱码问题现象导出的Excel打开中文显示为乱码解决方法# 连接数据库时指定编码 connection pymysql.connect( ..., charsetutf8mb4 ) # 保存Excel时指定编码 wb.save(output_path, encodingutf-8)5.3 日期格式问题现象数据库中的datetime导出后变成数字解决方法from openpyxl.styles import numbers for cell in ws[C]: # 假设C列是日期列 if cell.row 1: # 跳过表头 continue cell.number_format numbers.FORMAT_DATE_DATETIME6. 进阶功能实现6.1 多线程导出from concurrent.futures import ThreadPoolExecutor def export_table(table_name): sql fSELECT * FROM {table_name} output f{table_name}.xlsx export_to_excel(..., sql, output) with ThreadPoolExecutor(max_workers4) as executor: tables [users, orders, products] executor.map(export_table, tables)6.2 定时自动导出使用APScheduler实现定时任务from apscheduler.schedulers.blocking import BlockingScheduler sched BlockingScheduler() sched.scheduled_job(cron, hour2) # 每天凌晨2点执行 def daily_export(): export_to_excel(...) sched.start()6.3 命令行参数支持import argparse parser argparse.ArgumentParser() parser.add_argument(--host, requiredTrue) parser.add_argument(--user, requiredTrue) parser.add_argument(--output, defaultoutput.xlsx) args parser.parse_args() export_to_excel( hostargs.host, userargs.user, ... output_pathargs.output )7. 完整生产级代码示例#!/usr/bin/env python3 数据库导出Excel工具 - 生产环境版本 支持功能 1. 多线程分表导出 2. 自动拆分大文件 3. 完善的错误处理 4. 命令行参数支持 import argparse import logging from concurrent.futures import ThreadPoolExecutor from datetime import datetime from typing import Iterator, List, Dict import pymysql from openpyxl import Workbook from openpyxl.styles import numbers # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) logger logging.getLogger(__name__) MAX_ROWS_PER_SHEET 1000000 # Excel单Sheet最大行数 DEFAULT_BATCH_SIZE 5000 # 每次从数据库读取的行数 class DatabaseExporter: def __init__(self, host: str, user: str, password: str, database: str, port: int 3306): self.connection_params { host: host, user: user, password: password, database: database, port: port, cursorclass: pymysql.cursors.SSCursor, charset: utf8mb4 } def _execute_query(self, sql: str) - Iterator[List[Dict]]: 执行SQL查询并返回生成器 conn pymysql.connect(**self.connection_params) cursor conn.cursor(pymysql.cursors.DictCursor) try: logger.info(fExecuting query: {sql[:100]}...) cursor.execute(sql) while True: rows cursor.fetchmany(DEFAULT_BATCH_SIZE) if not rows: break yield rows finally: cursor.close() conn.close() def _format_value(self, value) - str: 格式化特殊类型数据 if isinstance(value, datetime): return value.strftime(%Y-%m-%d %H:%M:%S) return str(value) if value is not None else def export_to_excel(self, sql: str, output_path: str) - None: 主导出函数 wb Workbook(write_onlyTrue) sheet_count 1 current_row 0 headers None # 创建第一个Sheet ws wb.create_sheet(titlefSheet_{sheet_count}) for batch in self._execute_query(sql): # 首次获取数据时提取表头 if headers is None and batch: headers list(batch[0].keys()) ws.append(headers) for row in batch: # 检查是否需要新建Sheet if current_row MAX_ROWS_PER_SHEET: sheet_count 1 current_row 0 ws wb.create_sheet(titlefSheet_{sheet_count}) ws.append(headers) # 格式化并写入行数据 formatted_row [self._format_value(v) for v in row.values()] ws.append(formatted_row) current_row 1 logger.info(fProcessed {len(batch)} rows, total: {current_row}) # 保存工作簿 wb.save(output_path) logger.info(fSuccessfully exported to {output_path}) def main(): 命令行入口 parser argparse.ArgumentParser( descriptionExport database data to Excel file) parser.add_argument(--host, requiredTrue, helpDatabase host) parser.add_argument(--user, requiredTrue, helpDatabase user) parser.add_argument(--password, requiredTrue, helpDatabase password) parser.add_argument(--database, requiredTrue, helpDatabase name) parser.add_argument(--port, typeint, default3306, helpDatabase port) parser.add_argument(--sql, helpSQL query to execute) parser.add_argument(--table, helpExport entire table if specified) parser.add_argument(--output, requiredTrue, helpOutput Excel file path) parser.add_argument(--threads, typeint, default1, helpNumber of parallel threads) args parser.parse_args() exporter DatabaseExporter( hostargs.host, userargs.user, passwordargs.password, databaseargs.database, portargs.port ) if args.table: # 导出整个表 exporter.export_to_excel( sqlfSELECT * FROM {args.table}, output_pathargs.output ) elif args.sql: # 执行自定义SQL exporter.export_to_excel( sqlargs.sql, output_pathargs.output ) else: # 批量导出所有表 def export_table(table: str): output f{table}_{args.output} exporter.export_to_excel( sqlfSELECT * FROM {table}, output_pathoutput ) with ThreadPoolExecutor(max_workersargs.threads) as executor: # 获取所有表名 tables [row[Tables_in_db] for row in exporter._execute_query(SHOW TABLES)] executor.map(export_table, tables) if __name__ __main__: main()8. 实际应用中的经验分享连接池的使用对于高频导出任务建议使用DBUtils等连接池工具。实测连接池可以将频繁导出场景的性能提升3-5倍。超时设置复杂查询务必设置合理的超时时间。我曾经遇到过没有超时设置的导出任务运行了18小时最终因网络中断失败。断点续传对于超大数据量导出可以实现记录已导出行数的机制。示例代码# 记录导出进度 progress_file f{output_path}.progress last_exported_id 0 if os.path.exists(progress_file): with open(progress_file) as f: last_exported_id int(f.read()) sql fSELECT * FROM big_table WHERE id {last_exported_id} ORDER BY idExcel格式优化金融数据导出时数值列应该设置千分位分隔from openpyxl.styles import numbers for col in [B, C, D]: # 数值列 for cell in ws[col]: if cell.row ! 1: # 跳过表头 cell.number_format numbers.FORMAT_NUMBER_COMMA_SEPARATED1性能监控添加简单的性能统计start_time time.time() total_rows 0 # ...导出过程中... total_rows len(batch) elapsed time.time() - start_time speed total_rows / elapsed if elapsed 0 else 0 logger.info(fSpeed: {speed:.1f} rows/sec)
1. 项目背景与需求场景在日常数据处理工作中我们经常需要将数据库中的大量记录导出到Excel文件进行二次处理或分发。作为数据工程师我每周都要处理几十次这样的需求市场部门需要客户数据做分析、财务部门需要交易记录对账、运营团队需要用户行为数据生成报表...传统的手工操作方式存在明显痛点通过数据库客户端工具导出时每次都需要重复设置查询条件和导出参数数据量超过百万行时GUI工具经常卡死或崩溃需要定期执行的导出任务无法自动化不同数据库系统的导出操作差异大学习成本高Python正好能完美解决这些问题。最近我用PyMySQLopenpyxl组合实现了一套自动化导出方案单脚本可处理MySQL百万级数据导出还能自动拆分Excel文件避免超过104万行限制。下面分享具体实现方法和踩坑经验。2. 技术方案选型2.1 数据库连接方案比较对于Python连接数据库主流有几种方案DB-API标准接口优点标准化接口代码可移植性强缺点需要针对不同数据库安装特定驱动代表库PyMySQL(MySQL)、psycopg2(PostgreSQL)、cx_Oracle(Oracle)ORM框架优点面向对象操作自动防SQL注入缺点性能损耗学习曲线陡峭代表库SQLAlchemy、DjangoORM专用连接器优点厂商官方支持功能完整缺点依赖特定数据库代表mysql-connector-python提示对于纯导出场景推荐使用DB-API方案。ORM在简单查询场景会产生15-20%的性能开销2.2 Excel操作库选型处理Excel文件的Python库主要有库名称读写支持大文件处理公式支持样式调整适用场景openpyxl读写一般完善完善需要修改样式的情况xlsxwriter只写优秀基础完善大数据量导出pandas读写优秀无有限数据分析场景pyxlsb读写优秀无无处理二进制xlsb实测百万行数据导出openpyxl耗时约210秒内存占用1.2GBxlsxwriter耗时约95秒内存占用300MB3. 完整实现方案3.1 基础版本代码import pymysql from openpyxl import Workbook def export_to_excel(host, user, password, db, sql, output_path): # 建立数据库连接 connection pymysql.connect( hosthost, useruser, passwordpassword, databasedb, cursorclasspymysql.cursors.DictCursor ) try: with connection.cursor() as cursor: print(Executing query...) cursor.execute(sql) # 创建Excel工作簿 wb Workbook() ws wb.active # 写入表头 if cursor.description: headers [desc[0] for desc in cursor.description] ws.append(headers) # 分批写入数据 batch_size 10000 while True: rows cursor.fetchmany(batch_size) if not rows: break for row in rows: ws.append(list(row.values())) print(fProcessed {len(rows)} rows) # 保存文件 wb.save(output_path) print(fFile saved to {output_path}) finally: connection.close()3.2 生产环境增强版实际使用时需要考虑更多因素内存优化- 使用生成器分批处理def batch_fetch(cursor, size10000): while True: rows cursor.fetchmany(size) if not rows: break yield rows多Sheet支持- 避免Excel行数限制MAX_ROWS_PER_SHEET 1000000 # Excel限制 sheet_count 1 current_row 0 ws wb.create_sheet(fData_{sheet_count}) for batch in batch_fetch(cursor): for row in batch: if current_row MAX_ROWS_PER_SHEET: sheet_count 1 current_row 0 ws wb.create_sheet(fData_{sheet_count}) ws.append(headers) ws.append(list(row.values())) current_row 1类型处理- 处理datetime等特殊类型from datetime import datetime def format_value(value): if isinstance(value, datetime): return value.strftime(%Y-%m-%d %H:%M:%S) return str(value) if value is not None else 4. 性能优化技巧4.1 数据库层面优化使用SS游标(Server Side Cursor)connection pymysql.connect( ..., cursorclasspymysql.cursors.SSCursor )添加查询超时设置cursor.execute(SET SESSION max_execution_time300000) # 5分钟超时只查询必要字段避免SELECT *明确列出所需字段4.2 Excel写入优化禁用openpyxl自动计算wb Workbook(write_onlyTrue)使用xlsxwriter的常量内存模式import xlsxwriter workbook xlsxwriter.Workbook( large.xlsx, {constant_memory: True} )关闭自动过滤worksheet.autofilter False5. 常见问题与解决方案5.1 内存溢出问题现象处理大数据量时Python进程被Killed解决方案使用SSCursor游标减小batch_size(建议5000-10000)换用xlsxwriter库5.2 中文乱码问题现象导出的Excel打开中文显示为乱码解决方法# 连接数据库时指定编码 connection pymysql.connect( ..., charsetutf8mb4 ) # 保存Excel时指定编码 wb.save(output_path, encodingutf-8)5.3 日期格式问题现象数据库中的datetime导出后变成数字解决方法from openpyxl.styles import numbers for cell in ws[C]: # 假设C列是日期列 if cell.row 1: # 跳过表头 continue cell.number_format numbers.FORMAT_DATE_DATETIME6. 进阶功能实现6.1 多线程导出from concurrent.futures import ThreadPoolExecutor def export_table(table_name): sql fSELECT * FROM {table_name} output f{table_name}.xlsx export_to_excel(..., sql, output) with ThreadPoolExecutor(max_workers4) as executor: tables [users, orders, products] executor.map(export_table, tables)6.2 定时自动导出使用APScheduler实现定时任务from apscheduler.schedulers.blocking import BlockingScheduler sched BlockingScheduler() sched.scheduled_job(cron, hour2) # 每天凌晨2点执行 def daily_export(): export_to_excel(...) sched.start()6.3 命令行参数支持import argparse parser argparse.ArgumentParser() parser.add_argument(--host, requiredTrue) parser.add_argument(--user, requiredTrue) parser.add_argument(--output, defaultoutput.xlsx) args parser.parse_args() export_to_excel( hostargs.host, userargs.user, ... output_pathargs.output )7. 完整生产级代码示例#!/usr/bin/env python3 数据库导出Excel工具 - 生产环境版本 支持功能 1. 多线程分表导出 2. 自动拆分大文件 3. 完善的错误处理 4. 命令行参数支持 import argparse import logging from concurrent.futures import ThreadPoolExecutor from datetime import datetime from typing import Iterator, List, Dict import pymysql from openpyxl import Workbook from openpyxl.styles import numbers # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) logger logging.getLogger(__name__) MAX_ROWS_PER_SHEET 1000000 # Excel单Sheet最大行数 DEFAULT_BATCH_SIZE 5000 # 每次从数据库读取的行数 class DatabaseExporter: def __init__(self, host: str, user: str, password: str, database: str, port: int 3306): self.connection_params { host: host, user: user, password: password, database: database, port: port, cursorclass: pymysql.cursors.SSCursor, charset: utf8mb4 } def _execute_query(self, sql: str) - Iterator[List[Dict]]: 执行SQL查询并返回生成器 conn pymysql.connect(**self.connection_params) cursor conn.cursor(pymysql.cursors.DictCursor) try: logger.info(fExecuting query: {sql[:100]}...) cursor.execute(sql) while True: rows cursor.fetchmany(DEFAULT_BATCH_SIZE) if not rows: break yield rows finally: cursor.close() conn.close() def _format_value(self, value) - str: 格式化特殊类型数据 if isinstance(value, datetime): return value.strftime(%Y-%m-%d %H:%M:%S) return str(value) if value is not None else def export_to_excel(self, sql: str, output_path: str) - None: 主导出函数 wb Workbook(write_onlyTrue) sheet_count 1 current_row 0 headers None # 创建第一个Sheet ws wb.create_sheet(titlefSheet_{sheet_count}) for batch in self._execute_query(sql): # 首次获取数据时提取表头 if headers is None and batch: headers list(batch[0].keys()) ws.append(headers) for row in batch: # 检查是否需要新建Sheet if current_row MAX_ROWS_PER_SHEET: sheet_count 1 current_row 0 ws wb.create_sheet(titlefSheet_{sheet_count}) ws.append(headers) # 格式化并写入行数据 formatted_row [self._format_value(v) for v in row.values()] ws.append(formatted_row) current_row 1 logger.info(fProcessed {len(batch)} rows, total: {current_row}) # 保存工作簿 wb.save(output_path) logger.info(fSuccessfully exported to {output_path}) def main(): 命令行入口 parser argparse.ArgumentParser( descriptionExport database data to Excel file) parser.add_argument(--host, requiredTrue, helpDatabase host) parser.add_argument(--user, requiredTrue, helpDatabase user) parser.add_argument(--password, requiredTrue, helpDatabase password) parser.add_argument(--database, requiredTrue, helpDatabase name) parser.add_argument(--port, typeint, default3306, helpDatabase port) parser.add_argument(--sql, helpSQL query to execute) parser.add_argument(--table, helpExport entire table if specified) parser.add_argument(--output, requiredTrue, helpOutput Excel file path) parser.add_argument(--threads, typeint, default1, helpNumber of parallel threads) args parser.parse_args() exporter DatabaseExporter( hostargs.host, userargs.user, passwordargs.password, databaseargs.database, portargs.port ) if args.table: # 导出整个表 exporter.export_to_excel( sqlfSELECT * FROM {args.table}, output_pathargs.output ) elif args.sql: # 执行自定义SQL exporter.export_to_excel( sqlargs.sql, output_pathargs.output ) else: # 批量导出所有表 def export_table(table: str): output f{table}_{args.output} exporter.export_to_excel( sqlfSELECT * FROM {table}, output_pathoutput ) with ThreadPoolExecutor(max_workersargs.threads) as executor: # 获取所有表名 tables [row[Tables_in_db] for row in exporter._execute_query(SHOW TABLES)] executor.map(export_table, tables) if __name__ __main__: main()8. 实际应用中的经验分享连接池的使用对于高频导出任务建议使用DBUtils等连接池工具。实测连接池可以将频繁导出场景的性能提升3-5倍。超时设置复杂查询务必设置合理的超时时间。我曾经遇到过没有超时设置的导出任务运行了18小时最终因网络中断失败。断点续传对于超大数据量导出可以实现记录已导出行数的机制。示例代码# 记录导出进度 progress_file f{output_path}.progress last_exported_id 0 if os.path.exists(progress_file): with open(progress_file) as f: last_exported_id int(f.read()) sql fSELECT * FROM big_table WHERE id {last_exported_id} ORDER BY idExcel格式优化金融数据导出时数值列应该设置千分位分隔from openpyxl.styles import numbers for col in [B, C, D]: # 数值列 for cell in ws[col]: if cell.row ! 1: # 跳过表头 cell.number_format numbers.FORMAT_NUMBER_COMMA_SEPARATED1性能监控添加简单的性能统计start_time time.time() total_rows 0 # ...导出过程中... total_rows len(batch) elapsed time.time() - start_time speed total_rows / elapsed if elapsed 0 else 0 logger.info(fSpeed: {speed:.1f} rows/sec)
