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

你是不是也遇到过这种情况运营同事突然丢过来一句“帮我导一份上个月所有订单的Excel”你打开数据库客户端点导出等上十分钟好不容易导出成功结果文件大到打开就卡死。更别提还要手动挑字段、调格式、处理公司名里的特殊符号。这种活儿偶尔干一次还能忍要是每周、每天都有人找你“批量导数据”那就必须认真考虑用自动化方案了。这篇文章要聊的就是如何用 Python 批量导出数据库数据至 Excel 文件。我会把从零搭脚本到处理常见坑的完整经验拆开讲包括库怎么选、连接串怎么配、为什么分批读、多张表怎么导进同一个 Excel、大数据量怎么拆文件。无论你用的是 MySQL、PostgreSQL 还是 SQLite思路都是通用的。如果你经常给业务方提供数据、做报表或者需要定时备份数据这篇内容可以直接照着用。1. 需求场景与方案选型为什么是 Python 批量导出而不是手动另存为1.1 “又要导数据”——手动导出 Excel 的五个典型痛点先来对号入座看看你是否有过下面这些经历。第一数据量一大客户端导出经常超时。比如一次导 50 万行很多图形化数据库工具会直接卡在进度条上或者导出后打开 Excel 需要半分钟。第二多表数据需要合并到一个 Excel 里。客户要看“用户 订单 支付”三张表的数据手动一次一次导再手动复制粘贴到同一个文件里很容易漏数据、覆盖错列。第三导出的文件格式总被业务方吐槽。日期变成字符串、长数字变成科学计数法、表头没有加粗这些在业务眼里都是“不专业”。第四重复劳动。如果数据每周都要导一次每次都手动操作其实是在浪费生命。第五权限限制。很多生产库不希望你用客户端直连但通过脚本走受控账号反而更容易操作和追溯。“批量导出”这四个字本质上包含了三个层面的诉求多张表批量导出、大数据量按批次导出、以及多个时间周期定期自动导出。把它们拆开了才能设计出一套稳定的脚本。1.2 技术方案对比pandas、openpyxl、xlsxwriter 和 CSV选哪个先看一张对比表可以帮你快速决策方案代码量适合数据量格式控制典型问题数据库客户端手动导出零几万行以下弱无法批量、无法定时、易卡死导出 CSV 再用 Excel 打开极少几十万行以内很弱中文乱码、长数字变形、多表麻烦pandas to_excel极少十万行左右中等一次性读入内存量大容易 OOMpandas 分批 ExcelWriter中等百万行以内中等需要控制批大小和写入方式openpyxl 逐行写入中等几十万行强写入速度一般适合需要改样式的场景xlsxwriter 批量写入中等十万到百万行强不能读取已有 Excel 文件个人实测的经验是小数据量用 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-binarySQL 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, encodingutf-8) mysql_conf config[mysql] engine create_engine( fmysqlpymysql://{mysql_conf[user]}:{mysql_conf[password]} f{mysql_conf[host]}:{mysql_conf[port]}/{mysql_conf[database]} f?charset{mysql_conf[charset]} )连接串里charsetutf8mb4这个参数很关键。如果漏掉导出的中文容易出现乱码。utf8mb4是 MySQL 上对 Emoji 和生僻字的完整支持比utf8更稳妥。如果你连的是 PostgreSQL连接串大概是postgresqlpsycopg2://user:passwordhost:port/dbname思路一样。2.3 目录结构让脚本可维护导数据的需求通常会从“导一次”演变成“每周导一次”、“每天凌晨跑一次”。所以一开始就把目录结构理清楚后面能省很多事。我喜欢这样组织data_exporter/ ├── config.ini ├── requirements.txt ├── exporter.py └── output/ └── 2024-01-01_orders.xlsxconfig.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, encodingutf-8) mysql_conf config[mysql] engine create_engine( fmysqlpymysql://{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, indexFalse, sheet_name订单)这里read_sql返回一个 DataFrameto_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( host192.168.1.10, port3306, userexport_user, passwordyour_password, databasedemo_db, charsetutf8mb4, cursorclasspymysql.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( host192.168.1.10, port3306, userexport_user, passwordyour_password, databasedemo_db, charsetutf8mb4, cursorclasspymysql.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_size5000): conn pymysql.connect( host192.168.1.10, port3306, userexport_user, passwordyour_password, databasedemo_db, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor, ) workbook xlsxwriter.Workbook(output_path) worksheet workbook.add_worksheet(table_name[:31]) # 先查询一次表结构得到字段名 cursor conn.cursor() cursor.execute(fSELECT * 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 fSELECT * 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, errorsreplace) 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 文件里包含多个 SheetSheet 名分别是用户表、订单表、商品表。实现方式其实很简单在同一个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(fSELECT * 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(fSELECT * 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(fSheet「{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(mysqlpymysql://export_user:your_password192.168.1.10:3306/demo_db?charsetutf8mb4) 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, indexFalse, 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_okTrue) 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(foutput/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(foutput/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导致读取时字符串已经损坏二是数据库表中本身的编码和连接编码不一致。解决办法是在连接串里明确加上charsetutf8mb4并在写入 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, paramsparams)之后这些特殊字符都由数据库驱动处理不需要你手动转义。数据导入 Excel 时遇到\0、超长文本或控制字符也可能导致xlsxwriter报错我的经验是写入前做一次清洗把\x00替换成空字符串。下面是这些问题的一个快速速查表现象可能原因解决思路中文乱码连接字符集没配连接串加charsetutf8mb4长数字变科学计数法Excel 默认格式给列设置数字格式或转字符串日期显示为序列数未设置单元格格式使用write_datetime加格式跑一半连接断开等待超时调大超时、重连、断点续跑SQL 报错特殊字符未转义使用参数化查询内存溢出一次性读取全部数据分批查询我在实际排查中还有一个心得遇到导出文件异常时不要先怀疑 Python 代码先看原始 SQL 在数据库客户端里执行是否正常。很多时候错的不是导出程序而是查询语句本身比如表里存在NULL值、字段类型差异、权限不足等等。先确认 SQL 结果集再查写入逻辑能少走很多弯路。最后分享两个对我来说特别有用的习惯。第一个导出前先执行一次SELECT COUNT(*)快速估算数据量这样你能提前决定要不要拆文件、批大小设多少。第二个把“查询并写入一批数据”的逻辑封装成一个通用函数后续不管是加字段、加条件还是换输出格式都只需要改参数。踩过几次坑之后你就会明白导出数据这件事看似简单但把每个环节控制好才能真正做到“给我数据我给你文件”。