Python实现MySQL大数据高效导出Excel全攻略

1. 项目概述

在日常数据处理工作中，我们经常需要将数据库中的大量记录导出到Excel文件中进行分析或共享。手动操作不仅效率低下，而且容易出错。Python凭借其强大的数据库连接能力和Excel处理库，可以完美解决这个问题。

我最近接手了一个需要从MySQL数据库导出50万条销售记录到Excel的任务，通过Python脚本实现了全自动化处理。相比传统方法，Python方案将原本需要8小时的手工操作缩短到15分钟完成，且保证了100%的数据准确性。

2. 技术选型与准备

2.1 数据库连接方案

对于数据库连接，Python提供了多种选择：

• MySQL：推荐使用mysql-connector-python或PyMySQL
• PostgreSQL：psycopg2是最佳选择
• SQL Server：pyodbc表现稳定
• Oracle：cx_Oracle是官方推荐驱动

提示：无论选择哪种连接器，都建议使用连接池技术(如DBUtils)来管理数据库连接，特别是在处理大量数据时。

python复制# MySQL连接示例
import mysql.connector
from mysql.connector import pooling

db_config = {
    "host": "localhost",
    "user": "your_username",
    "password": "your_password",
    "database": "your_database"
}

connection_pool = pooling.MySQLConnectionPool(
    pool_name="my_pool",
    pool_size=5,
    **db_config
)

2.2 Excel处理库比较

Python处理Excel的主流库有：

1. openpyxl：功能全面，支持.xlsx格式
2. xlsxwriter：写入性能优异
3. pandas：高层抽象，适合数据分析场景
4. pyexcel：简单易用，支持多种格式

经过实测对比，我推荐以下组合方案：

• 数据量<10万行：pandas（代码最简洁）
• 10-50万行：openpyxl（内存控制更好）

• 50万行：xlsxwriter（性能最优）

3. 核心实现步骤

3.1 数据库查询与分页处理

直接一次性查询大量数据会导致内存溢出，必须采用分页查询技术：

python复制def batch_query(sql, page_size=10000):
    conn = connection_pool.get_connection()
    cursor = conn.cursor(dictionary=True)
    
    offset = 0
    while True:
        paginated_sql = f"{sql} LIMIT {offset}, {page_size}"
        cursor.execute(paginated_sql)
        results = cursor.fetchall()
        
        if not results:
            break
            
        yield results
        offset += page_size
    
    cursor.close()
    conn.close()

3.2 数据写入Excel优化

使用openpyxl的优化写入模式可以显著提升性能：

python复制from openpyxl import Workbook
from openpyxl.utils import get_column_letter

def write_to_excel(data, filename):
    wb = Workbook(write_only=True)  # 启用只写模式
    ws = wb.create_sheet()
    
    # 写入表头
    if data and isinstance(data[0], dict):
        headers = list(data[0].keys())
        ws.append(headers)
    
    # 批量写入数据
    for row in data:
        ws.append(list(row.values()) if isinstance(row, dict) else row)
    
    # 自动调整列宽
    for col in ws.columns:
        max_length = 0
        column = col[0].column_letter
        for cell in col:
            try:
                if len(str(cell.value)) > max_length:
                    max_length = len(str(cell.value))
            except:
                pass
        adjusted_width = (max_length + 2) * 1.2
        ws.column_dimensions[column].width = adjusted_width
    
    wb.save(filename)

3.3 完整流程整合

将各模块组合成完整解决方案：

python复制def export_db_to_excel(sql, filename, page_size=10000):
    # 初始化Excel文件
    from openpyxl import Workbook
    wb = Workbook(write_only=True)
    ws = wb.create_sheet()
    headers_written = False
    
    # 分页查询并写入
    for batch in batch_query(sql, page_size):
        if not headers_written and batch:
            headers = list(batch[0].keys())
            ws.append(headers)
            headers_written = True
        
        for row in batch:
            ws.append(list(row.values()))
    
    wb.save(filename)
    print(f"数据已成功导出到 {filename}")

4. 高级功能实现

4.1 多表关联导出

处理复杂查询时，可能需要导出多个关联表的数据：

python复制def export_related_tables(main_table, relation_config, filename):
    """导出主表及关联表数据"""
    wb = Workbook(write_only=True)
    
    # 导出主表
    main_sql = f"SELECT * FROM {main_table}"
    main_ws = wb.create_sheet(title=main_table)
    main_data = list(batch_query(main_sql))
    if main_data:
        main_ws.append(list(main_data[0][0].keys()))
        for batch in main_data:
            for row in batch:
                main_ws.append(list(row.values()))
    
    # 导出关联表
    for table, config in relation_config.items():
        relation_sql = config['sql']
        relation_ws = wb.create_sheet(title=table)
        relation_data = list(batch_query(relation_sql))
        if relation_data:
            relation_ws.append(list(relation_data[0][0].keys()))
            for batch in relation_data:
                for row in batch:
                    relation_ws.append(list(row.values()))
    
    wb.save(filename)

4.2 数据转换与清洗

在导出过程中进行数据清洗：

python复制def data_cleaner(row):
    """数据清洗函数示例"""
    # 处理空值
    for key in row:
        if row[key] is None:
            row[key] = ""
    
    # 日期格式化
    if 'create_time' in row and isinstance(row['create_time'], datetime):
        row['create_time'] = row['create_time'].strftime('%Y-%m-%d %H:%M:%S')
    
    # 金额单位转换
    if 'amount' in row and isinstance(row['amount'], (int, float)):
        row['amount'] = round(row['amount'] / 100, 2)  # 分转元
    
    return row

5. 性能优化技巧

5.1 内存管理

处理大数据量时的内存优化策略：

1. 使用生成器而非列表存储中间结果
2. 及时关闭数据库游标和连接
3. 采用分批次写入策略
4. 使用write_only模式创建工作簿

python复制# 内存友好的写入方式
def memory_efficient_write(data_iter, filename, batch_size=10000):
    wb = Workbook(write_only=True)
    ws = wb.create_sheet()
    
    batch = []
    for row in data_iter:
        batch.append(row)
        if len(batch) >= batch_size:
            ws.append(batch)
            batch = []
    
    if batch:
        ws.append(batch)
    
    wb.save(filename)

5.2 多线程处理

对于I/O密集型操作，可以使用多线程加速：

python复制from concurrent.futures import ThreadPoolExecutor

def parallel_export(tables, output_dir):
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = []
        for table in tables:
            filename = f"{output_dir}/{table}.xlsx"
            future = executor.submit(
                export_db_to_excel,
                f"SELECT * FROM {table}",
                filename
            )
            futures.append(future)
        
        for future in futures:
            future.result()  # 等待所有任务完成

6. 常见问题与解决方案

6.1 编码问题处理

数据库和Excel之间的编码转换：

python复制def safe_str(value):
    """处理各种编码问题"""
    if value is None:
        return ""
    if isinstance(value, bytes):
        try:
            return value.decode('utf-8')
        except UnicodeDecodeError:
            return value.decode('latin-1')
    return str(value)

6.2 大数据量导出策略

当数据量超过100万行时的处理方案：

1. 按时间范围分批导出
2. 使用多个Excel文件分割数据
3. 考虑先导出为CSV再转换
4. 使用专业ETL工具如Apache Airflow

python复制def large_export(sql, filename_pattern, date_column, start_date, end_date):
    """按日期范围分批导出"""
    current = start_date
    delta = timedelta(days=7)  # 每周一个文件
    
    while current <= end_date:
        batch_end = min(current + delta, end_date)
        where_clause = (
            f"WHERE {date_column} BETWEEN '{current}' AND '{batch_end}'"
        )
        filename = filename_pattern.format(
            start=current.strftime('%Y%m%d'),
            end=batch_end.strftime('%Y%m%d')
        )
        export_db_to_excel(f"{sql} {where_clause}", filename)
        current = batch_end + timedelta(days=1)

7. 完整示例代码

结合所有最佳实践的完整实现：

python复制import mysql.connector
from mysql.connector import pooling
from openpyxl import Workbook
from datetime import datetime, timedelta
import logging

class DatabaseExporter:
    def __init__(self, db_config):
        self.pool = self._create_pool(db_config)
        self.logger = logging.getLogger(__name__)
    
    def _create_pool(self, config):
        return pooling.MySQLConnectionPool(
            pool_name="export_pool",
            pool_size=5,
            **config
        )
    
    def batch_query(self, sql, params=None, page_size=10000):
        conn = self.pool.get_connection()
        cursor = conn.cursor(dictionary=True)
        
        offset = 0
        while True:
            paginated_sql = f"{sql} LIMIT {offset}, {page_size}"
            try:
                cursor.execute(paginated_sql, params or ())
                results = cursor.fetchall()
                
                if not results:
                    break
                    
                yield results
                offset += page_size
            except Exception as e:
                self.logger.error(f"查询出错: {e}")
                break
            finally:
                cursor.close()
                conn.close()
    
    def export_to_excel(self, sql, filename, page_size=10000, transform_fn=None):
        wb = Workbook(write_only=True)
        ws = wb.create_sheet()
        headers_written = False
        
        try:
            for batch in self.batch_query(sql, page_size=page_size):
                if not headers_written and batch:
                    headers = list(batch[0].keys())
                    ws.append(headers)
                    headers_written = True
                
                for row in batch:
                    if transform_fn:
                        row = transform_fn(row)
                    ws.append(list(row.values()))
            
            wb.save(filename)
            self.logger.info(f"成功导出到 {filename}")
            return True
        except Exception as e:
            self.logger.error(f"导出失败: {e}")
            return False

# 使用示例
if __name__ == "__main__":
    db_config = {
        "host": "localhost",
        "user": "user",
        "password": "password",
        "database": "sales_db"
    }
    
    exporter = DatabaseExporter(db_config)
    
    # 简单导出
    exporter.export_to_excel(
        "SELECT * FROM orders WHERE status='completed'",
        "orders.xlsx"
    )
    
    # 带数据转换的导出
    def transform_order(row):
        row['order_date'] = row['order_date'].strftime('%Y-%m-%d')
        row['amount'] = f"¥{row['amount']/100:.2f}"
        return row
    
    exporter.export_to_excel(
        "SELECT * FROM large_orders",
        "large_orders.xlsx",
        transform_fn=transform_order
    )

8. 扩展应用场景

8.1 定时自动导出

结合APScheduler实现定时导出：

python复制from apscheduler.schedulers.blocking import BlockingScheduler

def setup_scheduled_export():
    scheduler = BlockingScheduler()
    
    # 每天凌晨1点执行
    @scheduler.scheduled_job('cron', hour=1)
    def daily_export():
        exporter = DatabaseExporter(db_config)
        exporter.export_to_excel(
            "SELECT * FROM orders WHERE DATE(create_time)=CURDATE()-1",
            f"orders_{datetime.now().strftime('%Y%m%d')}.xlsx"
        )
    
    scheduler.start()

8.2 与邮件系统集成

导出后自动发送邮件：

python复制import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders

def send_email_with_attachment(to, subject, body, filename):
    msg = MIMEMultipart()
    msg['From'] = "export@company.com"
    msg['To'] = to
    msg['Subject'] = subject
    
    msg.attach(MIMEText(body))
    
    with open(filename, 'rb') as f:
        part = MIMEBase('application', 'octet-stream')
        part.set_payload(f.read())
        encoders.encode_base64(part)
        part.add_header(
            'Content-Disposition',
            f'attachment; filename="{filename}"'
        )
        msg.attach(part)
    
    with smtplib.SMTP('smtp.company.com') as server:
        server.send_message(msg)

在实际项目中，我建议将这些功能模块化，形成可复用的数据导出工具库。根据我的经验，一个好的导出系统应该具备以下特性：

1. 可配置的数据源连接
2. 灵活的数据转换管道
3. 完善的错误处理和日志记录
4. 可扩展的输出格式支持
5. 性能监控和优化提示

最后分享一个实用技巧：在处理特别大的数据导出时，可以先用EXPLAIN分析查询性能，确保数据库端已经优化。我曾经通过添加合适的索引，将一个需要2小时的导出任务缩短到15分钟。