通过pymysql读取数据库中表格并保存到excel(实用篇)

一、背景与问题

在数据处理场景中，我们经常需要将数据库中的结构化数据导出为Excel文件。这种需求常见于：

数据分析前的数据准备
系统间数据迁移
数据审计和存档
业务报表生成

传统方案通常采用以下流程：

使用pymysql连接MySQL数据库
查询目标数据
通过openpyxl或pandas等库生成Excel文件
将文件保存到指定路径

本篇文章将深入探讨该技术方案的实现细节，涵盖性能优化、安全考量、多库对比等关键内容。

二、基本原理

1. 数据库连接机制

pymysql通过MySQLdb库实现与MySQL的通信，其核心流程如下：

import pymysql

# 创建连接
connection = pymysql.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

# 创建游标
with connection.cursor() as cursor:
    # 执行SQL
    cursor.execute("SELECT * FROM test_table")
    # 获取结果
    results = cursor.fetchall()

2. Excel文件生成原理

使用openpyxl库时，其核心操作包括：

创建Workbook对象
添加工作表
写入数据（支持多种格式）
保存文件

from openpyxl import Workbook

wb = Workbook()
ws = wb.active
ws.title = "Sheet1"

# 写入表头
ws.append(['ID', 'Name', 'Created'])

# 写入数据
for row in data:
    ws.append([row['id'], row['name'], row['created']])
    
wb.save('output.xlsx')

3. 数据类型映射

MySQL字段类型与Excel格式的映射关系：

MySQL类型	Excel格式	备注
VARCHAR	文本	需要处理特殊字符转义
INT	数字	支持科学计数法
DATETIME	日期时间	自动格式化为Excel日期格式
BLOB	二进制数据	需要特殊处理

三、环境准备

确保安装以下依赖：

pip install pymysql openpyxl pandas

推荐版本：

pymysql 1.0.2
openpyxl 3.9.5
pandas 1.5.3

四、核心实现

1. 基础数据导出(无格式)

import pymysql
from openpyxl import Workbook

def export_to_excel():
    # 数据库连接参数
    config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'test_db',
        'charset': 'utf8mb4',
        'cursorclass': pymysql.cursors.DictCursor
    }
    
    # 创建连接
    connection = pymysql.connect(**config)
    
    try:
        with connection.cursor() as cursor:
            # 查询语句
            sql = "SELECT id, name, created FROM test_table"
            cursor.execute(sql)
            
            # 获取结果
            results = cursor.fetchall()
            
            # 创建Excel文件
            wb = Workbook()
            ws = wb.active
            ws.title = "Data"
            
            # 写入表头
            ws.append(['ID', 'Name', 'Created'])
            
            # 写入数据
            for row in results:
                ws.append([row['id'], row['name'], row['created']])
            
            # 保存文件
            wb.save('output.xlsx')
            
    finally:
        connection.close()

关键点解析：

使用DictCursor获取字典类型结果
明确指定字符集防止乱码
使用with语句确保连接正确关闭
日期字段自动转换为Excel日期格式

2. 分页处理(大数据量)

def export_large_data():
    config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'test_db',
        'charset': 'utf8mb4',
        'cursorclass': pymysql.cursors.DictCursor
    }
    
    connection = pymysql.connect(**config)
    
    try:
        with connection.cursor() as cursor:
            # 查询语句
            sql = "SELECT id, name, created FROM test_table ORDER BY id"
            cursor.execute(sql)
            
            # 获取总记录数
            total = cursor.fetchone()['id']
            
            # 分页参数
            page_size = 1000
            pages = (total // page_size) + 1
            
            # 创建Excel文件
            wb = Workbook()
            ws = wb.active
            ws.title = "LargeData"
            
            # 写入表头
            ws.append(['ID', 'Name', 'Created'])
            
            # 分页查询
            for page in range(pages):
                offset = page * page_size
                cursor.execute(f"{sql} LIMIT {page_size} OFFSET {offset}")
                results = cursor.fetchall()
                
                # 写入数据
                for row in results:
                    ws.append([row['id'], row['name'], row['created']])
            
            wb.save('large_data.xlsx')
            
    finally:
        connection.close()

3. 带格式导出(复杂场景)

import pandas as pd

def export_with_format():
    config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'test_db',
        'charset': 'utf8mb4'
    }
    
    connection = pymysql.connect(**config)
    
    try:
        with connection.cursor() as cursor:
            sql = "SELECT id, name, created FROM test_table"
            cursor.execute(sql)
            results = cursor.fetchall()
            
            # 转换为DataFrame
            df = pd.DataFrame(results, columns=['id', 'name', 'created'])
            
            # 格式化处理
            df['created'] = pd.to_datetime(df['created']).dt.strftime('%Y-%m-%d %H:%M:%S')
            
            # 保存为Excel
            df.to_excel('formatted_data.xlsx', index=False)
            
    finally:
        connection.close()

五、完整案例：用户数据导出

1. 业务场景

某电商平台需要将用户信息导出为Excel，包含以下字段：

用户ID
用户名
注册时间
最后登录时间
注册IP
状态

2. 实现代码

import pymysql
from openpyxl import Workbook
from datetime import datetime

def export_users():
    config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'ecommerce',
        'charset': 'utf8mb4',
        'cursorclass': pymysql.cursors.DictCursor
    }
    
    connection = pymysql.connect(**config)
    
    try:
        with connection.cursor() as cursor:
            # 查询语句
            sql = """
                SELECT 
                    id AS 用户ID,
                    name AS 用户名,
                    created AS 注册时间,
                    last_login AS 最后登录时间,
                    ip AS 注册IP,
                    status AS 状态
                FROM users
                ORDER BY created DESC
            """
            cursor.execute(sql)
            
            # 获取结果
            results = cursor.fetchall()
            
            # 创建Excel文件
            wb = Workbook()
            ws = wb.active
            ws.title = "用户数据"
            
            # 写入表头
            ws.append(['用户ID', '用户名', '注册时间', '最后登录时间', '注册IP', '状态'])
            
            # 写入数据
            for row in results:
                ws.append([
                    row['用户ID'],
                    row['用户名'],
                    row['注册时间'],
                    row['最后登录时间'],
                    row['注册IP'],
                    row['状态']
                ])
            
            # 设置列宽
            for col in range(1, ws.max_column+1):
                ws.column_dimensions[col].width = 20
                
            # 设置格式
            for row in ws.iter_rows(min_row=2):
                for cell in row:
                    cell.alignment = openpyxl.styles.Alignment(horizontal='center')
            
            # 保存文件
            wb.save(f'users_{datetime.now().strftime("%Y%m%d")}.xlsx')
            
    finally:
        connection.close()

六、源码解析

1. 数据库连接

connection = pymysql.connect(**config)

使用**config解包字典参数
cursorclass指定游标类型
charset设置字符集防止乱码

2. 数据查询

cursor.execute(sql)
results = cursor.fetchall()

fetchall()获取所有结果
使用DictCursor返回字典类型结果
可通过cursor.rowcount获取返回行数

3. Excel写入

ws.append(['用户ID', '用户名', '注册时间', '最后登录时间', '注册IP', '状态'])

使用append()方法逐行写入
ws.max_column获取最大列数
设置列宽和对齐方式

七、进阶使用

1. 性能优化

分页处理

避免一次性获取大量数据
使用LIMIT和OFFSET分页查询
按时间排序可使用WHERE id > last_id方式

批量写入

from openpyxl import Workbook

wb = Workbook()
ws = wb.active
ws.title = "Data"

# 批量写入
ws.append(['ID', 'Name'])
for row in data:
    ws.append([row['id'], row['name']])

使用pandas

df.to_excel('data.xlsx', index=False)

2. 安全考量

SQL注入防护

sql = "SELECT * FROM users WHERE id = %s"
cursor.execute(sql, (user_id,))

敏感信息处理

密码字段应加密存储
导出文件应限制访问权限
使用mysql-connector替代直接连接

3. 多库对比

库	优点	缺点
openpyxl	支持多种格式	文件较大
pandas	简化数据处理	依赖C库，需安装
xlsxwriter	支持复杂格式	需要额外安装

八、性能与工程实践

1. 性能优化策略

场景	优化方法	效果
大数据量	分页处理 + 批量写入	提升50%
频繁导出	缓存查询结果	提升30%
跨库导出	使用中间数据表	提升20%
多字段排序	使用索引	提升50%

2. 异常处理

try:
    connection = pymysql.connect(**config)
except pymysql.MySQLError as e:
    print(f"连接失败: {e}")
    exit(1)

3. 安全增强

使用ssl_verify=True连接加密
设置read_default_file使用配置文件
限制导出字段数量

九、常见问题与踩坑

1. 常见错误

错误类型	原因	解决方案
ConnectionError	数据库未启动	检查服务状态
UnicodeEncodeError	字符集不匹配	设置`charset='utf8mb4'`
TypeError	非字典类型字段	使用`DictCursor`
FileNotFoundError	文件写入路径错误	检查路径权限
MemoryError	大数据量处理	使用分页处理

2. 高级问题

日期格式问题

from datetime import datetime

def format_date(date_str):
    return datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S').strftime('%Y/%m/%d')

特殊字符处理

import re

def sanitize_value(value):
    return re.sub(r'[^a-zA-Z0-9\s\-\_]', '', value)

十、最佳实践

1. 推荐方案

使用pandas处理复杂数据转换
对大数据量使用分页处理
设置合理的连接超时和重试机制
采用配置文件管理数据库参数
对敏感字段进行脱敏处理

2. 推荐代码结构

project/
├── config/
│   └── db_config.py
├── utils/
│   └── excel_utils.py
├── core/
│   └── data_export.py
└── main.py

3. 推荐代码规范

使用上下文管理器处理连接
分离数据库逻辑和导出逻辑
对关键字段进行类型校验
使用日志记录关键操作

十一、总结

通过pymysql读取数据库并保存到Excel的方案，需要综合考虑多个技术要素。本文深入分析了该方案的实现原理，提供了三种不同复杂度的代码示例，并给出了完整的业务场景案例。在实际开发中，应根据具体需求选择合适的实现方式：

小数据量场景：使用基础方案
中等数据量：采用分页处理
大数据量场景：结合pandas进行优化

需要注意的是，该方案适用于数据导出需求，但不适合实时数据处理场景。对于需要频繁导出或处理大量数据的系统，建议采用更专业的ETL工具。在实际应用中，还需注意数据库连接安全、数据格式处理和异常处理等关键点。