用 Python 批量导出数据库数据到 Excel 的完整指南

你是不是也遇到过这种情况：运营同事突然丢过来一句“帮我导一份上个月所有订单的Excel”，你打开数据库客户端，点导出，等上十分钟，好不容易导出成功，结果文件大到打开就卡死。更别提还要手动挑字段、调格式、处理公司名里的特殊符号。这种活儿偶尔干一次还能忍，要是每周、每天都有人找你“批量导数据”，那就必须认真考虑用自动化方案了。

这篇文章要聊的就是如何用 Python 批量导出数据库数据至 Excel 文件。我会把从零搭脚本到处理常见坑的完整经验拆开讲，包括库怎么选、连接串怎么配、为什么分批读、多张表怎么导进同一个 Excel、大数据量怎么拆文件。无论你用的是 MySQL、PostgreSQL 还是 SQLite，思路都是通用的。如果你经常给业务方提供数据、做报表，或者需要定时备份数据，这篇内容可以直接照着用。

1. 需求场景与方案选型：为什么是 Python 批量导出而不是手动另存为

1.1 “又要导数据？”——手动导出 Excel 的五个典型痛点

先来对号入座，看看你是否有过下面这些经历。第一，数据量一大，客户端导出经常超时。比如一次导 50 万行，很多图形化数据库工具会直接卡在进度条上，或者导出后打开 Excel 需要半分钟。第二，多表数据需要合并到一个 Excel 里。客户要看“用户 + 订单 + 支付”三张表的数据，手动一次一次导，再手动复制粘贴到同一个文件里，很容易漏数据、覆盖错列。第三，导出的文件格式总被业务方吐槽。日期变成字符串、长数字变成科学计数法、表头没有加粗，这些在业务眼里都是“不专业”。第四，重复劳动。如果数据每周都要导一次，每次都手动操作，其实是在浪费生命。第五，权限限制。很多生产库不希望你用客户端直连，但通过脚本走受控账号，反而更容易操作和追溯。

“批量导出”这四个字，本质上包含了三个层面的诉求：多张表批量导出、大数据量按批次导出、以及多个时间周期定期自动导出。把它们拆开了，才能设计出一套稳定的脚本。

1.2 技术方案对比：pandas、openpyxl、xlsxwriter 和 CSV，选哪个

先看一张对比表，可以帮你快速决策：

个人实测的经验是：小数据量用 pandas 三行搞定，大数据量优先考虑 xlsxwriter。如果你需要给导出的 Excel 做条件格式、列宽、冻结窗格这些，xlsxwriter 更顺手。openpyxl 的强项是“既能读也能写”，但写入性能比 xlsxwriter 差一些，导大数据时差距会很直观。

为什么优先选 Python？因为生态成熟，数据库驱动、数据处理、Excel 写入都有现成的库，而且很容易挂到定时任务里。相比手动导出，脚本可以自动跑、自动归档，还能在跑完后推送通知。

2. 依赖准备与工程搭建：先跑通最小环境

2.1 安装依赖库与虚拟环境创建

我习惯给每个数据导出项目单独建虚拟环境，避免依赖冲突污染全局环境。你可以这样初始化：

python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate

然后安装核心依赖：

pip install pymysql pandas sqlalchemy openpyxl xlsxwriter

这里逐个说明一下它们的作用。pymysql是连接 MySQL 的驱动，如果你用的是 PostgreSQL，可以换成psycopg2-binary，SQL Server 则用pyodbc。SQLAlchemy是一个数据库连接抽象层，主要价值是让 pandas 的read_sql能直接使用，后续换数据库类型时改动更小。pandas负责读取数据和基础转换，但不是所有场景都非用不可。openpyxl和xlsxwriter都是写 Excel 的库，前者适合小数据量、需要改样式的场景，后者适合大批量高性能写入。

安装完成后，可以在 Python 里验证一下：

import pymysql, sqlalchemy, openpyxl, xlsxwriter print("依赖安装正常")

能正常打印，说明环境没问题。

2.2 数据库连接串与配置文件怎么写

很多刚接触数据库导出的人，会把账号密码直接硬编码在脚本里，这是个大忌。一方面，脚本一旦分享出去，密码就泄露了；另一方面，生产环境换密码时，你得在多个脚本里找。更好的做法是把连接信息放进配置文件。

我常用一个config.ini：

[mysql] host = 192.168.1.10 port = 3306 user = export_user password = your_password database = demo_db charset = utf8mb4

然后在 Python 里用configparser读取：

import configparser from sqlalchemy import create_engine config = configparser.ConfigParser() config.read("config.ini", encoding="utf-8") mysql_conf = config["mysql"] engine = create_engine( f"mysql+pymysql://{mysql_conf['user']}:{mysql_conf['password']}" f"@{mysql_conf['host']}:{mysql_conf['port']}/{mysql_conf['database']}" f"?charset={mysql_conf['charset']}" )

连接串里charset=utf8mb4这个参数很关键。如果漏掉，导出的中文容易出现乱码。utf8mb4是 MySQL 上对 Emoji 和生僻字的完整支持，比utf8更稳妥。如果你连的是 PostgreSQL，连接串大概是postgresql+psycopg2://user:password@host:port/dbname，思路一样。

2.3 目录结构：让脚本可维护

导数据的需求通常会从“导一次”演变成“每周导一次”、“每天凌晨跑一次”。所以一开始就把目录结构理清楚，后面能省很多事。我喜欢这样组织：

data_exporter/ ├── config.ini ├── requirements.txt ├── exporter.py └── output/ └── 2024-01-01_orders.xlsx

config.ini放连接配置，exporter.py放主逻辑，output存放生成的文件，文件名里带日期和业务标识。这样做的好处是，定时任务执行时能直接按日期归档，脚本本身不需要处理多余的文件管理逻辑。

3. 核心实现思路：从三行原型到可靠的分批导出

3.1 五秒原型：pandas 直接导出一张表

先写一个能立刻看到效果的版本，帮你确认环境没问题。用 pandas 导出一张表，只需要三行核心代码：

import pandas as pd from sqlalchemy import create_engine import configparser config = configparser.ConfigParser() config.read("config.ini", encoding="utf-8") mysql_conf = config["mysql"] engine = create_engine( f"mysql+pymysql://{mysql_conf['user']}:{mysql_conf['password']}" f"@{mysql_conf['host']}:{mysql_conf['port']}/{mysql_conf['database']}" f"?charset={mysql_conf['charset']}" ) df = pd.read_sql("SELECT * FROM orders LIMIT 1000", engine) df.to_excel("output/test_orders.xlsx", index=False, sheet_name="订单")

这里read_sql返回一个 DataFrame，to_excel负责写入 Excel。如果你的数据只有几千行，这个方案完全够用，代码也是最容易理解的。但要注意，read_sql默认是一次性把查询结果全部加载到内存，数据量一旦到几十万、上百万行，内存占用就会直线上升。所以在生产环境里，我几乎不会直接用这个原型。

3.2 为什么必须分批读：内存估算与三种分批方案

很多人不理解“分批读取”到底在解决什么问题。我们简单算一下：假设一行数据有 10 个字段，平均每个字段 100 字节，那一行大约是 1KB。100 万行就是 1GB 内存占用，还不算 pandas 内部创建 DataFrame 的额外开销。如果表里有TEXT类型的长文本，单行几 KB 甚至几十 KB 都很正常，内存很快就爆了。

分批读取有几种常见思路。第一种是分页查询，比如LIMIT 10000 OFFSET 100000。这个写法虽然简单，但数据量大了之后，偏移量越深，数据库扫描越慢，性能会越来越难看。第二种是时间窗口，比如按天或按月切分，适用于有明显时间字段的表，但对时间分布不均匀的数据效果不稳定。第三种是我最推荐的键集分页，也就是基于主键或唯一键定位：

SELECT id, order_no, amount, created_at FROM orders WHERE id > 上一次读取的最大id ORDER BY id ASC LIMIT 5000

这种方式每次只取主键递增区间内的数据，不需要扫描前面已经读过的行，性能稳定，也不会重复读取。

3.3 用流式游标加键集分页实现稳定导出

如果不想在应用层手动维护id边界，也可以直接使用数据库的流式游标。以 MySQL 为例，pymysql连接时可以指定SSCursor，让数据库把结果一条一条地发给客户端，而不是一次性把所有数据塞进内存。

import pymysql conn = pymysql.connect( host="192.168.1.10", port=3306, user="export_user", password="your_password", database="demo_db", charset="utf8mb4", cursorclass=pymysql.cursors.SSDictCursor, ) with conn.cursor() as cursor: cursor.execute("SELECT id, order_no, amount, created_at FROM orders") for row in cursor: # 逐行处理，内存占用很少 print(row)

流式游标适合纯读取场景，但你要注意，在使用过程中连接会一直保持打开，不能在同一连接上执行其他查询。所以更稳妥的批量导出方案，还是结合前面说的键集分页。

下面是一个完整的“分批读取 + xlsxwriter 写入”示例：

import pymysql import xlsxwriter conn = pymysql.connect( host="192.168.1.10", port=3306, user="export_user", password="your_password", database="demo_db", charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, ) workbook = xlsxwriter.Workbook("output/orders.xlsx") worksheet = workbook.add_worksheet("订单") headers = ["id", "order_no", "amount", "created_at"] for col, name in enumerate(headers): worksheet.write(0, col, name) batch_size = 5000 last_id = 0 row_idx = 1 cursor = conn.cursor() while True: sql = """ SELECT id, order_no, amount, created_at FROM orders WHERE id > %s ORDER BY id ASC LIMIT %s """ cursor.execute(sql, (last_id, batch_size)) rows = cursor.fetchall() if not rows: break for row in rows: worksheet.write(row_idx, 0, row["id"]) worksheet.write(row_idx, 1, row["order_no"]) worksheet.write(row_idx, 2, row["amount"]) worksheet.write(row_idx, 3, str(row["created_at"])) row_idx += 1 last_id = rows[-1]["id"] print(f"已导出 {row_idx - 1} 行，当前批次最后 ID = {last_id}") cursor.close() conn.close() workbook.close()

这里用WHERE id > %s ORDER BY id ASC LIMIT %s分批，每一批 5000 行，每一批内部逐行写入 Excel。如果你的表主键不是id，改成唯一递增字段即可。如果表里没有单列主键，可以用联合主键里的第一列，或者用时间戳配合其他条件。

4. 实操案例：单表、多表与自定义条件下的批量导出

4.1 单表全量导出完整脚本

当你需要完整导出一张业务表时，思路就是“确定分批字段 -> 循环读取 -> 写入 Excel -> 统计行数”。我用一个可复用的函数来封装这个逻辑：

import pymysql import xlsxwriter def export_table_to_excel(table_name, output_path, batch_size=5000): conn = pymysql.connect( host="192.168.1.10", port=3306, user="export_user", password="your_password", database="demo_db", charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, ) workbook = xlsxwriter.Workbook(output_path) worksheet = workbook.add_worksheet(table_name[:31]) # 先查询一次表结构，得到字段名 cursor = conn.cursor() cursor.execute(f"SELECT * FROM {table_name} LIMIT 0") headers = [desc[0] for desc in cursor.description] for col, name in enumerate(headers): worksheet.write(0, col, name) # 找到主键列用于分批 cursor.execute( """ SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_KEY = 'PRI' ORDER BY ORDINAL_POSITION LIMIT 1 """, (table_name,), ) pk_row = cursor.fetchone() if not pk_row: raise RuntimeError(f"表 {table_name} 没有主键，无法使用键集分页，请选择其他分批字段。") pk_col = pk_row["COLUMN_NAME"] last_id = 0 row_idx = 1 total = 0 while True: sql = f"SELECT * FROM {table_name} WHERE {pk_col} > %s ORDER BY {pk_col} LIMIT %s" cursor.execute(sql, (last_id, batch_size)) rows = cursor.fetchall() if not rows: break for row in rows: for col, header in enumerate(headers): value = row[header] if isinstance(value, (bytes, bytearray)): value = value.decode("utf-8", errors="replace") worksheet.write(row_idx, col, value) row_idx += 1 total += 1 last_id = rows[-1][pk_col] print(f"{table_name} 已导出 {total} 行，最后 {pk_col} = {last_id}") cursor.close() conn.close() workbook.close() print(f"{table_name} 导出完成，共 {total} 行，文件：{output_path}")

要注意的是，我没有把表名直接拼进 SQL 里做参数化，因为表名无法作为参数绑定。这里table_name是从配置或白名单里传来的，不是用户输入，因此可控。如果表名来自外部输入，必须严格校验，否则会有安全风险。

4.2 多表导出到一个 Excel 的多个 Sheet

多表批量导出的需求非常常见。业务方往往希望一个 Excel 文件里包含多个 Sheet，Sheet 名分别是用户表、订单表、商品表。实现方式其实很简单，在同一个Workbook里按表名循环创建工作表。

tables = ["users", "orders", "products"] output_path = "output/multi_tables.xlsx" workbook = xlsxwriter.Workbook(output_path) for table_name in tables: worksheet = workbook.add_worksheet(table_name[:31]) conn = pymysql.connect(...) # 复用你项目里的连接逻辑 cursor = conn.cursor(pymysql.cursors.DictCursor) # 写表头 cursor.execute(f"SELECT * FROM {table_name} LIMIT 0") headers = [desc[0] for desc in cursor.description] for col, name in enumerate(headers): worksheet.write(0, col, name) cursor.execute(f"SELECT * FROM {table_name}") row_idx = 1 for row in cursor: for col, header in enumerate(headers): worksheet.write(row_idx, col, row[header]) row_idx += 1 cursor.close() conn.close() print(f"Sheet「{table_name}」写入完成，共 {row_idx - 1} 行") workbook.close()

这里有两个容易踩的坑。第一，Excel 的 Sheet 名最长是 31 个字符，且不能包含: \ / ? * [ ]这些字符。我调用table_name[:31]就长截断，如果表名本身含非法字符，还需要做一次清洗。第二，如果表数据量很大，逐行fetchall会占用大量内存，这时应该像前面一样，给每张表单独做分批查询。

4.3 按日期范围与指定字段做参数化导出

很多时候你不需要整张表，只需要某个时间段、某些字段的数据。这类“自定义导出”需求，用参数化查询是很自然的实现方式。

import pandas as pd from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://export_user:your_password@192.168.1.10:3306/demo_db?charset=utf8mb4") start_date = "2024-01-01" end_date = "2024-02-01" sql = """ SELECT id, order_no, user_name, amount, created_at FROM orders WHERE created_at >= %s AND created_at < %s """ df = pd.read_sql(sql, engine, params=(start_date, end_date)) df.to_excel("output/orders_2024_01.xlsx", index=False, sheet_name="订单")

注意，read_sql的params参数可以避免 SQL 注入，这是必须养成的习惯。使用%s占位符而不是手工拼字符串，能有效防止日期值里出现特殊字符导致的 SQL 错误。

如果你要继续用pymysql + xlsxwriter做同样的事，只需要把WHERE条件放进cursor.execute的第二个参数：

cursor.execute( """ SELECT id, order_no, user_name, amount, created_at FROM orders WHERE created_at >= %s AND created_at < %s """, (start_date, end_date), )

我在实际项目中，会把日期范围、目标表名、输出文件路径都放在一个小的配置里，这样业务方提新需求的时候，我只需要改配置，不需要动代码。如果你的脚本要给别人用，这个设计尤为重要。

5. 性能优化与 Excel 格式细节

5.1 让导出文件更好用的格式化技巧

一份交付给业务方的 Excel，格式是否用心，一眼就能看出来。我通常会做四件事：表头加粗、冻结首行、列宽自适应、日期列设置显示格式。用xlsxwriter做这些非常简单。

workbook = xlsxwriter.Workbook("output/orders_format.xlsx") worksheet = workbook.add_worksheet("订单") header_format = workbook.add_format({"bold": True, "bg_color": "#D9E1F2", "border": 1}) date_format = workbook.add_format({"num_format": "yyyy-mm-dd hh:mm:ss"}) headers = ["id", "order_no", "amount", "created_at"] for col, name in enumerate(headers): worksheet.write(0, col, name, header_format) worksheet.freeze_panes(1, 0) # 写入数据时，日期列使用 date_format worksheet.write_datetime(row_idx, 3, datetime_value, date_format)

列宽自适应可以写一个小函数，根据每列的最大字符数设置宽度。要注意，中文字符的实际显示宽度约为英文字符的两倍，算宽度时可以做简单折算。这一层优化不是“为了好看而好看”，而是为了减少业务方打开文件后的二次整理工作。

5.2 超过单 Sheet 上限时自动拆分为多个文件

Excel 2007 之后的 .xlsx 格式，单个工作表最多支持 1048576 行。如果你导出的表超过这个量，不拆分的话会直接报错。所以我通常设定一个比上限低很多的安全阈值，比如每个文件写 50 万行，超过就自动切到下一个文件。

import os os.makedirs("output", exist_ok=True) max_rows_per_file = 500000 file_idx = 1 row_count_in_current_file = 0 current_workbook = None current_worksheet = None while True: rows = fetch_next_batch() if current_workbook is None: current_workbook = xlsxwriter.Workbook(f"output/orders_part_{file_idx}.xlsx") current_worksheet = current_workbook.add_worksheet("订单") # 写表头... if row_count_in_current_file >= max_rows_per_file: current_workbook.close() file_idx += 1 current_workbook = xlsxwriter.Workbook(f"output/orders_part_{file_idx}.xlsx") current_worksheet = current_workbook.add_worksheet("订单") row_count_in_current_file = 0 # 重新写表头... for row in rows: current_worksheet.write(...) row_count_in_current_file += len(rows)

这里的关键参数是max_rows_per_file。50 万行是一个比较稳妥的值，文件体积通常在 50MB 以内，Excel 打开不会太卡。如果你要继续用 Excel 做透视表，可能还要考虑单文件体积别太大。

5.3 关于导出速度的量化经验和调优方向

导出速度主要取决于三个环节：数据库查询速度、数据从数据库传到 Python 进程的网络耗时、以及 Excel 写入速度。我实测过一份 100 万行的订单表，字段大概 10 个，使用xlsxwriter分批写入，整个过程大约需要 1 到 2 分钟，其中大部分时间花在数据库查询和网络传输上。

优化方向有四条：第一，在数据库层面，对排序字段加索引，否则ORDER BY id在数据量大时会慢很多。第二，减少导出字段，不要无脑SELECT *，只查业务需要的列。第三，提升批大小，比如从 5000 调整到 20000，测试后发现网络往返次数减少，整体速度会有明显提升。第四，如果允许，尽量在业务低峰期执行导出，避免影响线上服务。

6. 常见问题与排查技巧实录

6.1 导出的 Excel 打开乱码或中文变问号

这个问题我碰到过很多次。常见原因有两个：一是 MySQL 连接的charset没有设置成utf8mb4，导致读取时字符串已经损坏；二是数据库表中本身的编码和连接编码不一致。解决办法是，在连接串里明确加上charset=utf8mb4，并在写入 Excel 前确认 Python 侧的字符串是正常的str类型，而不是bytes。如果是从文本文件读配置，也要保证config.ini是以 UTF-8 编码保存的。

6.2 数字列变成科学计数法

Excel 对超过 11 位的数字默认用科学计数法显示，最典型的是身份证号、订单号、流水号。解决方法是，把这些列当作字符串写入。在xlsxwriter里，可以给列单独设置num_format为0，或者在写入前将值转为str。我更推荐前者，因为转成字符串后，Excel 单元格里会有一个绿色的小三角提示“文本类型”，业务看着也会困惑。

6.3 日期时间格式丢失

直接写入datetime类型时，如果没设置单元格格式，Excel 可能显示成一串数字序列。解决办法是使用xlsxwriter的write_datetime方法，或者写一个格式对象并应用给目标列。给日期列加上"yyyy-mm-dd hh:mm:ss"这样的格式，是最稳妥的。

6.4 脚本跑到一半连接断开

批量导出通常要跑几分钟甚至更长，如果数据库的wait_timeout设置较短，连接可能中途断开。应对方法有几种：调大数据库连接池和客户端的超时时间；在循环中捕获OperationalError后重连；或者对耗时很长的任务，拆成多个子任务执行。另一个技巧是定期打印进度日志，这样即使断开，你也能知道是断在哪个分区段，从断点附近继续导，而不是全量重来。

6.5 特殊字符让 SQL 报错

数据里如果包含单引号、反斜杠、换行符，直接拼接 SQL 时会特别容易出错。这也是我强调参数化查询的原因。使用cursor.execute(sql, params)或read_sql(sql, engine, params=params)之后，这些特殊字符都由数据库驱动处理，不需要你手动转义。数据导入 Excel 时，遇到\0、超长文本或控制字符，也可能导致xlsxwriter报错，我的经验是写入前做一次清洗，把\x00替换成空字符串。

下面是这些问题的一个快速速查表：

我在实际排查中还有一个心得：遇到导出文件异常时，不要先怀疑 Python 代码，先看原始 SQL 在数据库客户端里执行是否正常。很多时候错的不是导出程序，而是查询语句本身，比如表里存在NULL值、字段类型差异、权限不足等等。先确认 SQL 结果集，再查写入逻辑，能少走很多弯路。

最后分享两个对我来说特别有用的习惯。第一个，导出前先执行一次SELECT COUNT(*)，快速估算数据量，这样你能提前决定要不要拆文件、批大小设多少。第二个，把“查询并写入一批数据”的逻辑封装成一个通用函数，后续不管是加字段、加条件，还是换输出格式，都只需要改参数。踩过几次坑之后，你就会明白，导出数据这件事看似简单，但把每个环节控制好，才能真正做到“给我数据，我给你文件”。