1. 项目背景与核心需求最近在整理客户历史订单数据时遇到了一个典型的数据搬运需求——需要将MySQL数据库中近三年的交易记录导出到Excel文件并按月份拆分存储。手工操作不仅效率低下还容易出错。于是用Python写了个自动化脚本没想到后来这个工具成了团队里的高频使用工具。数据库到Excel的数据搬运是数据分析、报表生成等场景的基础操作。无论是财务对账、销售统计还是用户行为分析都需要将数据库中的结构化数据导出为更易读的Excel格式。传统的手工导出存在三个痛点多表关联查询结果无法直接导出大数据量导出时容易内存溢出需要定期执行的导出任务缺乏自动化2. 技术方案选型2.1 数据库连接方式比较常见的Python数据库连接库主要有三种MySQLdb最原始的MySQL连接器性能好但不支持Python3PyMySQL纯Python实现的替代方案兼容性好SQLAlchemyORM工具支持多种数据库对于简单的导出任务PyMySQL足够轻量。如果项目已使用SQLAlchemy也可以直接沿用。这里选择PyMySQL作为演示import pymysql conn pymysql.connect( hostlocalhost, userroot, passwordyourpassword, databasesales_db, charsetutf8mb4 )2.2 Excel写入库选择Python处理Excel的主流库对比库名称优点缺点适用场景openpyxl功能全面支持.xlsx内存占用大需要修改现有文件xlwt/xlrd速度快只支持老版.xls兼容旧系统pandas接口简单依赖其他库数据分析场景xlsxwriter性能好只写不能读纯写入需求考虑到我们只需要写入功能且要处理大量数据选择xlsxwriter作为核心写入引擎。3. 完整实现方案3.1 基础版本实现先实现单表导出的基础功能import pymysql import xlsxwriter from datetime import datetime def export_to_excel(host, user, password, db, table, output_file): # 建立数据库连接 connection pymysql.connect( hosthost, useruser, passwordpassword, databasedb, charsetutf8mb4 ) try: with connection.cursor() as cursor: # 获取数据 sql fSELECT * FROM {table} cursor.execute(sql) results cursor.fetchall() # 获取列名 desc cursor.description column_names [col[0] for col in desc] # 创建Excel文件 workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入表头 for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(results, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()3.2 分页查询优化当数据量超过万条时一次性读取会导致内存暴涨。改用分页查询def batch_export(host, user, password, db, table, output_file, batch_size5000): connection pymysql.connect(...) try: with connection.cursor() as cursor: # 获取总行数 count_sql fSELECT COUNT(*) FROM {table} cursor.execute(count_sql) total cursor.fetchone()[0] # 分页处理 workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 先写入表头 cursor.execute(fSELECT * FROM {table} LIMIT 1) desc cursor.description column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 分批读取数据 for offset in range(0, total, batch_size): sql fSELECT * FROM {table} LIMIT {offset}, {batch_size} cursor.execute(sql) batch cursor.fetchall() # 写入当前批次 for row_num, row in enumerate(batch, offset1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()3.3 多表关联导出实际业务中经常需要关联多表查询def export_join_tables(host, user, password, db, sql, output_file): connection pymysql.connect(...) try: with connection.cursor() as cursor: cursor.execute(sql) results cursor.fetchall() desc cursor.description workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入动态列名 column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(results, 1): for col_num, value in enumerate(row): # 处理datetime类型 if isinstance(value, datetime): value value.strftime(%Y-%m-%d %H:%M:%S) worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()4. 高级功能实现4.1 自动拆分多Sheet当数据需要按类别分组导出时def export_by_category(host, user, password, db, table, category_column, output_file): connection pymysql.connect(...) try: with connection.cursor() as cursor: # 获取所有分类 cursor.execute(fSELECT DISTINCT {category_column} FROM {table}) categories [row[0] for row in cursor.fetchall()] workbook xlsxwriter.Workbook(output_file) for category in categories: # 为每个分类创建sheet sheet_name str(category)[:31] # Excel sheet名最长31字符 worksheet workbook.add_worksheet(sheet_name) # 查询该分类数据 sql fSELECT * FROM {table} WHERE {category_column} %s cursor.execute(sql, (category,)) results cursor.fetchall() desc cursor.description # 写入表头 column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(results, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()4.2 添加样式和格式使用xlsxwriter的样式功能提升可读性def export_with_styles(host, user, password, db, table, output_file): connection pymysql.connect(...) try: with connection.cursor() as cursor: cursor.execute(fSELECT * FROM {table}) results cursor.fetchall() desc cursor.description workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 定义样式 header_style workbook.add_format({ bold: True, bg_color: #4F81BD, font_color: white, border: 1 }) date_style workbook.add_format({num_format: yyyy-mm-dd}) money_style workbook.add_format({num_format: $#,##0.00}) # 写入带样式的表头 column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name, header_style) # 自动调整列宽 worksheet.set_column(col_num, col_num, len(name)2) # 写入数据并应用样式 for row_num, row in enumerate(results, 1): for col_num, (value, col) in enumerate(zip(row, desc)): col_type col[1] # 获取字段类型 if isinstance(value, datetime): worksheet.write_datetime(row_num, col_num, value, date_style) elif col_type pymysql.NUMBER and col[5] 2: # 小数位为2的数值 worksheet.write_number(row_num, col_num, value, money_style) else: worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()5. 性能优化技巧5.1 内存优化方案处理超大数据集时超过10万行使用服务器端游标SScursor减少内存占用分批写入磁盘避免在内存中累积全部数据禁用xlsxwriter的默认字符串缓存优化后的实现def export_large_dataset(host, user, password, db, table, output_file, batch_size10000): connection pymysql.connect( hosthost, useruser, passwordpassword, databasedb, charsetutf8mb4, cursorclasspymysql.cursors.SSCursor # 服务器端游标 ) try: with connection.cursor() as cursor: # 获取列信息 cursor.execute(fSELECT * FROM {table} LIMIT 1) desc cursor.description column_names [col[0] for col in desc] # 创建Excel文件并优化性能 workbook xlsxwriter.Workbook( output_file, {constant_memory: True, strings_to_urls: False} ) worksheet workbook.add_worksheet() # 写入表头 for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 分批读取和写入 cursor.execute(fSELECT * FROM {table}) row_num 1 while True: batch cursor.fetchmany(batch_size) if not batch: break for row in batch: for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) row_num 1 finally: connection.close() workbook.close()5.2 多线程导出对于可以并行导出的多个表from concurrent.futures import ThreadPoolExecutor def parallel_export(config_list): config_list: [{ host: localhost, user: root, password: xxx, db: db1, table: table1, output: output1.xlsx }, ...] with ThreadPoolExecutor(max_workers4) as executor: futures [] for config in config_list: future executor.submit( export_to_excel, hostconfig[host], userconfig[user], passwordconfig[password], dbconfig[db], tableconfig[table], output_fileconfig[output] ) futures.append(future) # 等待所有任务完成 for future in futures: future.result()6. 异常处理与日志记录6.1 健壮性增强完善的异常处理机制import logging from xlsxwriter.exceptions import FileCreateError logging.basicConfig( filenameexport.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) def safe_export(host, user, password, db, table, output_file): try: # 验证输出路径 if not output_file.endswith(.xlsx): raise ValueError(输出文件必须是.xlsx格式) # 记录开始时间 start_time datetime.now() logging.info(f开始导出 {table} 到 {output_file}) # 执行导出 export_to_excel(host, user, password, db, table, output_file) # 记录完成时间 duration (datetime.now() - start_time).total_seconds() logging.info(f导出完成耗时 {duration:.2f} 秒) except pymysql.Error as e: logging.error(f数据库错误: {e}) raise except FileCreateError as e: logging.error(f文件创建失败: {e}) raise except Exception as e: logging.error(f未知错误: {e}) raise6.2 进度反馈添加进度显示功能from tqdm import tqdm def export_with_progress(host, user, password, db, table, output_file): connection pymysql.connect(...) try: with connection.cursor() as cursor: # 获取总行数 cursor.execute(fSELECT COUNT(*) FROM {table}) total cursor.fetchone()[0] workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入表头 cursor.execute(fSELECT * FROM {table} LIMIT 1) desc cursor.description column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 使用tqdm显示进度 cursor.execute(fSELECT * FROM {table}) with tqdm(totaltotal, unitrows) as pbar: for row_num, row in enumerate(cursor, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) pbar.update(1) finally: connection.close() workbook.close()7. 实际应用案例7.1 电商订单导出系统为电商平台设计的自动化导出方案def export_orders(start_date, end_date, output_dir): 导出指定日期范围内的订单数据 sql SELECT o.order_id, o.order_date, c.customer_name, p.product_name, oi.quantity, oi.unit_price, oi.quantity * oi.unit_price AS total FROM orders o JOIN customers c ON o.customer_id c.customer_id JOIN order_items oi ON o.order_id oi.order_id JOIN products p ON oi.product_id p.product_id WHERE o.order_date BETWEEN %s AND %s ORDER BY o.order_date # 按月份拆分导出 current start_date while current end_date: month_start current.replace(day1) month_end (month_start timedelta(days32)).replace(day1) - timedelta(days1) month_end min(month_end, end_date) output_file os.path.join( output_dir, forders_{month_start.strftime(%Y%m)}.xlsx ) conn pymysql.connect(...) try: with conn.cursor() as cursor: cursor.execute(sql, (month_start, month_end)) results cursor.fetchall() desc cursor.description workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入表头 column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(results, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) finally: conn.close() workbook.close() current month_end timedelta(days1)7.2 定时自动导出任务结合Windows任务计划或Linux cron实现自动化# auto_export.py import schedule import time def daily_export(): today datetime.now().strftime(%Y%m%d) output_file f/data/exports/sales_{today}.xlsx export_to_excel( hostprod-db.example.com, userreport_user, passwordsecurepassword, dbsales, tabledaily_sales, output_fileoutput_file ) # 每天凌晨1点执行 schedule.every().day.at(01:00).do(daily_export) while True: schedule.run_pending() time.sleep(60)8. 常见问题与解决方案8.1 编码问题处理处理数据库中的特殊字符# 在连接配置中添加charset参数 conn pymysql.connect( hostlocalhost, userroot, passwordyourpassword, databasesales_db, charsetutf8mb4, # 支持emoji等特殊字符 cursorclasspymysql.cursors.DictCursor ) # 写入Excel前对字符串进行清洗 def clean_string(value): if isinstance(value, str): return value.encode(unicode_escape).decode(ascii) return value8.2 大数据量导出优化当处理百万级数据时的建议使用服务器端游标(SSCursor)增加batch_size到5000-10000考虑先导出为CSV再转换使用--quick参数连接MySQL8.3 性能对比测试不同方案的导出速度对比测试数据10万行20列方案耗时(秒)内存峰值(MB)基础版本45.2780分页查询38.7120服务器游标36.585多线程(4线程)22.12109. 扩展功能思路9.1 支持更多数据库类型通过SQLAlchemy实现多数据库支持from sqlalchemy import create_engine def export_using_sqlalchemy(connection_str, sql, output_file): engine create_engine(connection_str) with engine.connect() as conn: result conn.execute(sql) workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入表头 column_names result.keys() for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(result, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) workbook.close()9.2 添加数据验证在Excel中添加下拉菜单等验证def export_with_validation(output_file): workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 添加数据验证 worksheet.data_validation( B2:B100, { validate: list, source: [High, Medium, Low] } ) # 添加条件格式 format_high workbook.add_format({bg_color: #FFC7CE}) worksheet.conditional_format( C2:C100, { type: cell, criteria: , value: 1000, format: format_high } ) workbook.close()9.3 与Pandas集成利用Pandas简化数据处理import pandas as pd def export_with_pandas(host, user, password, db, sql, output_file): conn pymysql.connect(...) try: df pd.read_sql(sql, conn) # 使用Pandas的ExcelWriter with pd.ExcelWriter( output_file, enginexlsxwriter, datetime_formatyyyy-mm-dd hh:mm:ss ) as writer: df.to_excel(writer, indexFalse) # 获取xlsxwriter对象进行额外设置 workbook writer.book worksheet writer.sheets[Sheet1] # 设置自动筛选 worksheet.autofilter(0, 0, 0, len(df.columns)-1) finally: conn.close()
Python实现MySQL数据高效导出Excel的自动化方案
1. 项目背景与核心需求最近在整理客户历史订单数据时遇到了一个典型的数据搬运需求——需要将MySQL数据库中近三年的交易记录导出到Excel文件并按月份拆分存储。手工操作不仅效率低下还容易出错。于是用Python写了个自动化脚本没想到后来这个工具成了团队里的高频使用工具。数据库到Excel的数据搬运是数据分析、报表生成等场景的基础操作。无论是财务对账、销售统计还是用户行为分析都需要将数据库中的结构化数据导出为更易读的Excel格式。传统的手工导出存在三个痛点多表关联查询结果无法直接导出大数据量导出时容易内存溢出需要定期执行的导出任务缺乏自动化2. 技术方案选型2.1 数据库连接方式比较常见的Python数据库连接库主要有三种MySQLdb最原始的MySQL连接器性能好但不支持Python3PyMySQL纯Python实现的替代方案兼容性好SQLAlchemyORM工具支持多种数据库对于简单的导出任务PyMySQL足够轻量。如果项目已使用SQLAlchemy也可以直接沿用。这里选择PyMySQL作为演示import pymysql conn pymysql.connect( hostlocalhost, userroot, passwordyourpassword, databasesales_db, charsetutf8mb4 )2.2 Excel写入库选择Python处理Excel的主流库对比库名称优点缺点适用场景openpyxl功能全面支持.xlsx内存占用大需要修改现有文件xlwt/xlrd速度快只支持老版.xls兼容旧系统pandas接口简单依赖其他库数据分析场景xlsxwriter性能好只写不能读纯写入需求考虑到我们只需要写入功能且要处理大量数据选择xlsxwriter作为核心写入引擎。3. 完整实现方案3.1 基础版本实现先实现单表导出的基础功能import pymysql import xlsxwriter from datetime import datetime def export_to_excel(host, user, password, db, table, output_file): # 建立数据库连接 connection pymysql.connect( hosthost, useruser, passwordpassword, databasedb, charsetutf8mb4 ) try: with connection.cursor() as cursor: # 获取数据 sql fSELECT * FROM {table} cursor.execute(sql) results cursor.fetchall() # 获取列名 desc cursor.description column_names [col[0] for col in desc] # 创建Excel文件 workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入表头 for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(results, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()3.2 分页查询优化当数据量超过万条时一次性读取会导致内存暴涨。改用分页查询def batch_export(host, user, password, db, table, output_file, batch_size5000): connection pymysql.connect(...) try: with connection.cursor() as cursor: # 获取总行数 count_sql fSELECT COUNT(*) FROM {table} cursor.execute(count_sql) total cursor.fetchone()[0] # 分页处理 workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 先写入表头 cursor.execute(fSELECT * FROM {table} LIMIT 1) desc cursor.description column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 分批读取数据 for offset in range(0, total, batch_size): sql fSELECT * FROM {table} LIMIT {offset}, {batch_size} cursor.execute(sql) batch cursor.fetchall() # 写入当前批次 for row_num, row in enumerate(batch, offset1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()3.3 多表关联导出实际业务中经常需要关联多表查询def export_join_tables(host, user, password, db, sql, output_file): connection pymysql.connect(...) try: with connection.cursor() as cursor: cursor.execute(sql) results cursor.fetchall() desc cursor.description workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入动态列名 column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(results, 1): for col_num, value in enumerate(row): # 处理datetime类型 if isinstance(value, datetime): value value.strftime(%Y-%m-%d %H:%M:%S) worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()4. 高级功能实现4.1 自动拆分多Sheet当数据需要按类别分组导出时def export_by_category(host, user, password, db, table, category_column, output_file): connection pymysql.connect(...) try: with connection.cursor() as cursor: # 获取所有分类 cursor.execute(fSELECT DISTINCT {category_column} FROM {table}) categories [row[0] for row in cursor.fetchall()] workbook xlsxwriter.Workbook(output_file) for category in categories: # 为每个分类创建sheet sheet_name str(category)[:31] # Excel sheet名最长31字符 worksheet workbook.add_worksheet(sheet_name) # 查询该分类数据 sql fSELECT * FROM {table} WHERE {category_column} %s cursor.execute(sql, (category,)) results cursor.fetchall() desc cursor.description # 写入表头 column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(results, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()4.2 添加样式和格式使用xlsxwriter的样式功能提升可读性def export_with_styles(host, user, password, db, table, output_file): connection pymysql.connect(...) try: with connection.cursor() as cursor: cursor.execute(fSELECT * FROM {table}) results cursor.fetchall() desc cursor.description workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 定义样式 header_style workbook.add_format({ bold: True, bg_color: #4F81BD, font_color: white, border: 1 }) date_style workbook.add_format({num_format: yyyy-mm-dd}) money_style workbook.add_format({num_format: $#,##0.00}) # 写入带样式的表头 column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name, header_style) # 自动调整列宽 worksheet.set_column(col_num, col_num, len(name)2) # 写入数据并应用样式 for row_num, row in enumerate(results, 1): for col_num, (value, col) in enumerate(zip(row, desc)): col_type col[1] # 获取字段类型 if isinstance(value, datetime): worksheet.write_datetime(row_num, col_num, value, date_style) elif col_type pymysql.NUMBER and col[5] 2: # 小数位为2的数值 worksheet.write_number(row_num, col_num, value, money_style) else: worksheet.write(row_num, col_num, value) finally: connection.close() workbook.close()5. 性能优化技巧5.1 内存优化方案处理超大数据集时超过10万行使用服务器端游标SScursor减少内存占用分批写入磁盘避免在内存中累积全部数据禁用xlsxwriter的默认字符串缓存优化后的实现def export_large_dataset(host, user, password, db, table, output_file, batch_size10000): connection pymysql.connect( hosthost, useruser, passwordpassword, databasedb, charsetutf8mb4, cursorclasspymysql.cursors.SSCursor # 服务器端游标 ) try: with connection.cursor() as cursor: # 获取列信息 cursor.execute(fSELECT * FROM {table} LIMIT 1) desc cursor.description column_names [col[0] for col in desc] # 创建Excel文件并优化性能 workbook xlsxwriter.Workbook( output_file, {constant_memory: True, strings_to_urls: False} ) worksheet workbook.add_worksheet() # 写入表头 for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 分批读取和写入 cursor.execute(fSELECT * FROM {table}) row_num 1 while True: batch cursor.fetchmany(batch_size) if not batch: break for row in batch: for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) row_num 1 finally: connection.close() workbook.close()5.2 多线程导出对于可以并行导出的多个表from concurrent.futures import ThreadPoolExecutor def parallel_export(config_list): config_list: [{ host: localhost, user: root, password: xxx, db: db1, table: table1, output: output1.xlsx }, ...] with ThreadPoolExecutor(max_workers4) as executor: futures [] for config in config_list: future executor.submit( export_to_excel, hostconfig[host], userconfig[user], passwordconfig[password], dbconfig[db], tableconfig[table], output_fileconfig[output] ) futures.append(future) # 等待所有任务完成 for future in futures: future.result()6. 异常处理与日志记录6.1 健壮性增强完善的异常处理机制import logging from xlsxwriter.exceptions import FileCreateError logging.basicConfig( filenameexport.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) def safe_export(host, user, password, db, table, output_file): try: # 验证输出路径 if not output_file.endswith(.xlsx): raise ValueError(输出文件必须是.xlsx格式) # 记录开始时间 start_time datetime.now() logging.info(f开始导出 {table} 到 {output_file}) # 执行导出 export_to_excel(host, user, password, db, table, output_file) # 记录完成时间 duration (datetime.now() - start_time).total_seconds() logging.info(f导出完成耗时 {duration:.2f} 秒) except pymysql.Error as e: logging.error(f数据库错误: {e}) raise except FileCreateError as e: logging.error(f文件创建失败: {e}) raise except Exception as e: logging.error(f未知错误: {e}) raise6.2 进度反馈添加进度显示功能from tqdm import tqdm def export_with_progress(host, user, password, db, table, output_file): connection pymysql.connect(...) try: with connection.cursor() as cursor: # 获取总行数 cursor.execute(fSELECT COUNT(*) FROM {table}) total cursor.fetchone()[0] workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入表头 cursor.execute(fSELECT * FROM {table} LIMIT 1) desc cursor.description column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 使用tqdm显示进度 cursor.execute(fSELECT * FROM {table}) with tqdm(totaltotal, unitrows) as pbar: for row_num, row in enumerate(cursor, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) pbar.update(1) finally: connection.close() workbook.close()7. 实际应用案例7.1 电商订单导出系统为电商平台设计的自动化导出方案def export_orders(start_date, end_date, output_dir): 导出指定日期范围内的订单数据 sql SELECT o.order_id, o.order_date, c.customer_name, p.product_name, oi.quantity, oi.unit_price, oi.quantity * oi.unit_price AS total FROM orders o JOIN customers c ON o.customer_id c.customer_id JOIN order_items oi ON o.order_id oi.order_id JOIN products p ON oi.product_id p.product_id WHERE o.order_date BETWEEN %s AND %s ORDER BY o.order_date # 按月份拆分导出 current start_date while current end_date: month_start current.replace(day1) month_end (month_start timedelta(days32)).replace(day1) - timedelta(days1) month_end min(month_end, end_date) output_file os.path.join( output_dir, forders_{month_start.strftime(%Y%m)}.xlsx ) conn pymysql.connect(...) try: with conn.cursor() as cursor: cursor.execute(sql, (month_start, month_end)) results cursor.fetchall() desc cursor.description workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入表头 column_names [col[0] for col in desc] for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(results, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) finally: conn.close() workbook.close() current month_end timedelta(days1)7.2 定时自动导出任务结合Windows任务计划或Linux cron实现自动化# auto_export.py import schedule import time def daily_export(): today datetime.now().strftime(%Y%m%d) output_file f/data/exports/sales_{today}.xlsx export_to_excel( hostprod-db.example.com, userreport_user, passwordsecurepassword, dbsales, tabledaily_sales, output_fileoutput_file ) # 每天凌晨1点执行 schedule.every().day.at(01:00).do(daily_export) while True: schedule.run_pending() time.sleep(60)8. 常见问题与解决方案8.1 编码问题处理处理数据库中的特殊字符# 在连接配置中添加charset参数 conn pymysql.connect( hostlocalhost, userroot, passwordyourpassword, databasesales_db, charsetutf8mb4, # 支持emoji等特殊字符 cursorclasspymysql.cursors.DictCursor ) # 写入Excel前对字符串进行清洗 def clean_string(value): if isinstance(value, str): return value.encode(unicode_escape).decode(ascii) return value8.2 大数据量导出优化当处理百万级数据时的建议使用服务器端游标(SSCursor)增加batch_size到5000-10000考虑先导出为CSV再转换使用--quick参数连接MySQL8.3 性能对比测试不同方案的导出速度对比测试数据10万行20列方案耗时(秒)内存峰值(MB)基础版本45.2780分页查询38.7120服务器游标36.585多线程(4线程)22.12109. 扩展功能思路9.1 支持更多数据库类型通过SQLAlchemy实现多数据库支持from sqlalchemy import create_engine def export_using_sqlalchemy(connection_str, sql, output_file): engine create_engine(connection_str) with engine.connect() as conn: result conn.execute(sql) workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 写入表头 column_names result.keys() for col_num, name in enumerate(column_names): worksheet.write(0, col_num, name) # 写入数据 for row_num, row in enumerate(result, 1): for col_num, value in enumerate(row): worksheet.write(row_num, col_num, value) workbook.close()9.2 添加数据验证在Excel中添加下拉菜单等验证def export_with_validation(output_file): workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet() # 添加数据验证 worksheet.data_validation( B2:B100, { validate: list, source: [High, Medium, Low] } ) # 添加条件格式 format_high workbook.add_format({bg_color: #FFC7CE}) worksheet.conditional_format( C2:C100, { type: cell, criteria: , value: 1000, format: format_high } ) workbook.close()9.3 与Pandas集成利用Pandas简化数据处理import pandas as pd def export_with_pandas(host, user, password, db, sql, output_file): conn pymysql.connect(...) try: df pd.read_sql(sql, conn) # 使用Pandas的ExcelWriter with pd.ExcelWriter( output_file, enginexlsxwriter, datetime_formatyyyy-mm-dd hh:mm:ss ) as writer: df.to_excel(writer, indexFalse) # 获取xlsxwriter对象进行额外设置 workbook writer.book worksheet writer.sheets[Sheet1] # 设置自动筛选 worksheet.autofilter(0, 0, 0, len(df.columns)-1) finally: conn.close()
