Python自动化批量导出数据库数据到Excel的实践

1. 项目背景与需求分析

在日常数据处理工作中，我们经常需要将数据库中的大量数据导出到Excel文件进行二次处理或分享。手动操作不仅效率低下，而且容易出错。Python作为数据处理利器，配合适当的库可以轻松实现自动化批量导出。

这个项目主要解决三个痛点：

• 多表数据需要按业务逻辑批量导出
• 导出过程需要保持数据完整性和格式规范
• 导出的Excel文件需要符合业务部门的使用习惯

2. 技术方案选型

2.1 核心工具链选择

经过对比测试，我最终确定的工具组合是：

python复制import pandas as pd
import sqlalchemy as sa
from openpyxl import Workbook

选择理由：

1. Pandas提供强大的DataFrame数据结构，完美衔接数据库查询结果
2. SQLAlchemy作为ORM工具，支持多种数据库引擎
3. Openpyxl提供精细的Excel文件控制能力

2.2 数据库连接配置

以MySQL为例的连接配置模板：

python复制def create_db_engine():
    return sa.create_engine(
        "mysql+pymysql://user:password@host:port/dbname",
        pool_recycle=3600,
        echo=False
    )

重要提示：生产环境务必使用配置文件存储凭证，不要硬编码在代码中

3. 核心实现逻辑

3.1 批量查询与数据转换

基础查询函数实现：

python复制def query_to_dataframe(engine, sql):
    with engine.connect() as conn:
        return pd.read_sql(sql, conn)

进阶版本支持分块查询：

python复制def batch_query(engine, sql, chunk_size=10000):
    chunks = []
    with engine.connect() as conn:
        for chunk in pd.read_sql(sql, conn, chunksize=chunk_size):
            chunks.append(chunk)
    return pd.concat(chunks)

3.2 Excel导出优化

多sheet导出实现：

python复制def export_to_excel(data_dict, filepath):
    with pd.ExcelWriter(filepath, engine='openpyxl') as writer:
        for sheet_name, df in data_dict.items():
            df.to_excel(writer, sheet_name=sheet_name, index=False)

格式增强版：

python复制def export_with_format(data_dict, filepath):
    with pd.ExcelWriter(filepath, engine='openpyxl') as writer:
        for sheet_name, df in data_dict.items():
            df.to_excel(writer, sheet_name=sheet_name[:30], index=False)
            worksheet = writer.sheets[sheet_name[:30]]
            # 设置列宽自适应
            for column in worksheet.columns:
                max_length = max(len(str(cell.value)) for cell in column)
                worksheet.column_dimensions[column[0].column_letter].width = min(max_length + 2, 50)

4. 完整工作流实现

4.1 主流程控制

python复制def main():
    # 初始化
    engine = create_db_engine()
    output_file = "export_{}.xlsx".format(datetime.now().strftime("%Y%m%d"))
    
    # 查询配置
    queries = {
        "用户数据": "SELECT * FROM users WHERE status=1",
        "订单数据": "SELECT * FROM orders WHERE create_time > '2023-01-01'",
        "商品数据": "SELECT id,name,price FROM products"
    }
    
    # 执行导出
    data = {name: query_to_dataframe(engine, sql) for name, sql in queries.items()}
    export_to_excel(data, output_file)
    
    print(f"导出完成，文件保存在：{output_file}")

4.2 异常处理增强

python复制def safe_export():
    try:
        main()
    except sa.exc.SQLAlchemyError as e:
        print(f"数据库错误：{str(e)}")
        # 记录错误日志
        with open("export_error.log", "a") as f:
            f.write(f"{datetime.now()} - {str(e)}\n")
    except PermissionError:
        print("文件写入权限不足，请检查输出目录")
    except Exception as e:
        print(f"未知错误：{str(e)}")

5. 高级功能扩展

5.1 定时自动导出

结合APScheduler实现定时任务：

python复制from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job('cron', hour=2, minute=30)
def daily_export():
    safe_export()

if __name__ == '__main__':
    scheduler.start()

5.2 邮件自动发送

使用smtplib自动发送导出文件：

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

def send_email_with_attachment(filepath):
    msg = MIMEMultipart()
    msg['From'] = 'sender@example.com'
    msg['To'] = 'receiver@example.com'
    msg['Subject'] = '数据自动导出报告'
    
    part = MIMEBase('application', "octet-stream")
    with open(filepath, 'rb') as file:
        part.set_payload(file.read())
    encoders.encode_base64(part)
    part.add_header('Content-Disposition', f'attachment; filename="{filepath}"')
    msg.attach(part)
    
    with smtplib.SMTP('smtp.example.com', 587) as server:
        server.starttls()
        server.login('username', 'password')
        server.send_message(msg)

6. 性能优化技巧

6.1 内存优化方案

对于超大数据集（>100万行）：

1. 使用分块查询+分块写入
2. 禁用DataFrame索引
3. 指定列数据类型

优化后代码示例：

python复制def export_large_data(engine, sql, filepath, chunk_size=50000):
    with pd.ExcelWriter(filepath, engine='openpyxl') as writer:
        with engine.connect() as conn:
            for i, chunk in enumerate(pd.read_sql(sql, conn, chunksize=ch_size)):
                chunk.to_excel(writer, sheet_name=f"Part_{i+1}", index=False)
                writer.save()  # 分段保存

6.2 多线程导出

python复制from concurrent.futures import ThreadPoolExecutor

def parallel_export(query_list):
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = []
        for query in query_list:
            futures.append(executor.submit(query_to_dataframe, engine, query))
        results = [f.result() for f in futures]
    return dict(zip([q['name'] for q in query_list], results))

7. 常见问题排查

7.1 典型错误与解决方案

7.2 调试技巧

1. 先测试小数据集
2. 打印SQL日志（设置echo=True）
3. 使用head()检查DataFrame结构
4. 临时导出为CSV检查数据完整性

8. 项目部署建议

8.1 环境配置

推荐使用conda创建独立环境：

bash复制conda create -n data_export python=3.8
conda install pandas sqlalchemy openpyxl

8.2 日志记录增强

python复制import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('export.log'),
        logging.StreamHandler()
    ]
)

在实际项目中，我发现将数据库连接池大小设置为5-10个连接，配合适当的超时设置（如30秒），可以在大多数场景下获得最佳性能。对于特别复杂的查询，建议先在数据库客户端中测试执行计划，确保查询本身已经优化。