企业官网建设流程全解析

上周业务方的同事跑来找我，说能不能把库里近一年的订单明细导成 Excel 他们要自己拉透视表。我第一反应是这还不简单，pandas.read_sql接to_excel，几行代码就收工。等真拿到生产环境的数据量一看，一张表几千万行，直接全量读取内存根本扛不住，Excel 单 Sheet 的行数上限也摆在那，根本不是一个to_excel能解决的问题。

所以专门写了这个 Python 批量导出工具，把数据库里的数据按批次、按分表、按业务维度拆成多个 Excel 文件，或者在一个文件里分多个 Sheet 导出，整个过程带进度、带日志、带断点续传。这篇文章就把我的完整实现思路、代码、还有踩过的坑一次说清楚，给要做报表导出、数据分发、运营取数的朋友一个可以直接落地的参考。

1. 为什么简单的 Excel 导出，落地时却最容易翻车

先说结论：单机、小数据量、一次性脚本，用 pandas 一把梭确实没问题。但数据量一旦上来，或者任务变成每天定时跑，问题就一串串往外冒。

1.1 数据量和 Excel 格式本身的硬限制

Excel 的.xlsx格式有个让人头疼的单 Sheet 上限：1048576 行。很多业务表动辄几百万行，直接全量导出必然失败。就算没到上限，超过 20 万行的 Excel 文件打开和刷新都会变得很卡，业务方的电脑稍微旧一点，打开一次要等半分钟。

另外还有一个大家容易忽略的限制：单元格内文本不能超过 32767 个字符。数据库里如果有大字段、备注、JSON 配置之类的长文本，超长部分会被 Excel 静默截断，业务方拿到手才发现数据不对，这时候再排查就尴尬了。所以在导出逻辑里必须预留字段长度检测和截断策略。

再加上 Excel 对日期时间格式、超长数字（比如雪花 ID、手机号）的"自动科学计数法"问题，整个导出任务要考虑的细节一点不比业务功能少。

1.2 常见方案对比：为什么我选了 SQLAlchemy + pandas + xlsxwriter

写这个工具之前我对比了三种主流方案：

我最终选择的是第三种的组合，但把查询层包了一层pandas，而不是完全手动fetchmany。原因是pandas的分块读取、类型推断、空值处理已经足够成熟，没必要重复造轮子；真正需要自己控制的是"分块大小""写入方式""文件切分策略"这些工程层面的东西。

1.3 批量导出到底在"批量"什么

这个标题里的"批量"其实包含三个维度：

• 批量导出多张表：比如一个库里有订单表、用户表、商品表，一次任务全部导出成独立文件，或者合成一个多 Sheet 的 Excel。
• 单张表数据量过大，批量分块导出：把 500 万行拆成 5 个文件，每个 100 万行，避免触发 Excel 上限。
• 增量任务每天跑：只导出当天新增或变化的数据，而不是每天全量扫全表。

这三个维度在代码设计上是互相叠加的，下面第二、三章会分别展开。

2. 核心实现：从数据库读取到 Excel 落盘的完整代码

这一章直接给完整的可运行代码，并解释每一处关键设计的原因。我会用虚拟的表结构order_records做例子，表里有id、trade_date、user_id、product_name、remark等字段，覆盖了常规数字、日期、文本、长文本几种典型类型。

2.1 数据库连接：用 SQLAlchemy 统一管理

第一步是建立数据库连接。我习惯用 SQLAlchemy 的create_engine，而不是直接用某个数据库驱动裸连，原因有两个：

• 换数据库不用改代码，MySQL、PostgreSQL、SQLite 都能用同一套导出代码。
• pandas.read_sql原生支持 SQLAlchemy 引擎，参数占位符也更安全。

from sqlalchemy import create_engine def build_engine(db_type: str = "mysql", host: str = "127.0.0.1", port: int = 3306, username: str = "your_user", password: str = "your_pass", database: str = "your_db") -> object: """ 构建 SQLAlchemy 数据库引擎。 如果 db_type 是 sqlite，只需传 database 路径即可。 """ if db_type == "sqlite": return create_engine(f"sqlite:///{database}", pool_pre_ping=True) url = f"{db_type}://{username}:{password}@{host}:{port}/{database}?charset=utf8mb4" return create_engine(url, pool_pre_ping=True, pool_size=5, pool_recycle=3600)

注意charset=utf8mb4，这个对 MySQL 很重要。utf8mb4才能完整支持 emoji 和生僻字，如果漏掉，数据里带个 emoji 符号写到 Excel 里就会变成乱码。这是我在真实项目中踩过的坑。

pool_pre_ping=True这个参数的意思是每次从连接池取连接之前先做一次轻量探测，防止拿到一个已经失效的连接。定时任务里数据库连接经常被服务端主动断开，加上这个参数能省掉很多莫名其妙的MySQL server has gone away报错。

2.2 主函数：按配置批量导出多张表到同一个 Excel

这个设计的核心思路是：把"导出什么表""导出哪些列""按什么条件过滤"都放到配置字典里，代码本身保持通用。这样业务上新增一张表的导出需求，只需要加一段配置，不需要改逻辑。

import pandas as pd from sqlalchemy import text def export_tables_to_excel(engine, tables_config: list, output_path: str, chunksize: int = 50000, sheet_limit: int = 500000): """ tables_config: [ { "sheet_name": "订单表", "query": "SELECT id, trade_date, user_id, product_name, remark FROM order_records WHERE trade_date >= :start_date", "params": {"start_date": "2024-01-01"}, "order_by": "id", }, ... ] """ writer = pd.ExcelWriter(output_path, engine="xlsxwriter") for cfg in tables_config: sheet_name = cfg["sheet_name"][:31] # Excel sheet 名最长 31 字符 query = cfg["query"] params = cfg.get("params", {}) order_by = cfg.get("order_by", "") # 构造带分页包裹的查询，后面解释为什么这么做 paged_query = wrap_query_with_row_number(query, order_by) total_exported = 0 sheet_row = 0 # 当前 sheet 已写入行数 # 把分页条件逐段传入 offset = 0 while True: page_query = paged_query + f" LIMIT {chunksize} OFFSET {offset}" df = pd.read_sql(text(page_query), engine, params=params) if df.empty: break # 达到单个 sheet 的行数上限，换下一个 sheet if sheet_row + len(df) > sheet_limit: # 先处理不满一个 sheet 的剩余部分... pass df.to_excel(writer, sheet_name=sheet_name, index=False, startrow=sheet_row + 1, header=(sheet_row == 0)) sheet_row += len(df) total_exported += len(df) offset += chunksize print(f"[{sheet_name}] 已导出 {total_exported} 行，偏移 {offset}") writer.close() print(f"导出完成：{output_path}")

这段代码里wrap_query_with_row_number是我封装的一个分页函数，下面单独说明。先解释几个关键点：

• index=False必须写：DataFrame自带的行索引如果不关掉，会以多出来一列的方式写进 Excel，业务方会看到一列没名字的序号，非常不专业。
• header=(sheet_row == 0)控制列名只写一次：分块导出到同一个 sheet 时，只有第一块数据需要写列名，后面续写只需要追加数据行。这个判断非常关键，不写的话每个分块都会重复一行字段名。
• startrow=sheet_row + 1跳过列名行：第一块从第 1 行开始写（0 是列名行），后续块从已有数据行之后续写。

2.3 稳定分页：用主键排序代替 OFFSET

LIMIT ... OFFSET ...在数据量小的时候没问题，但偏移量越大，数据库扫描越慢。更严重的是，如果trade_date有重复值，而分页排序不稳定，同一批数据可能被重复导出或漏掉。

我在这里用了基于排序键的键集分页，稳定且性能好：

def stable_page_query(base_query: str, order_column: str, last_value, page_size: int): if last_value is None: return f"{base_query} ORDER BY {order_column} LIMIT {page_size}" return f"{base_query} AND {order_column} > :last_value ORDER BY {order_column} LIMIT {page_size}"

每次记录当前页最后一条order_column的值，下一页用WHERE order_column > last_value取数据。这样即使数据在导出过程中被并发写入，也不会出现 OFFSET 分页那种"页与页之间数据错位"的问题。

当然前提是order_column有唯一性，或者至少业务上能接受"排序键相同的数据被一起跨越"的边缘情况。实际操作中我会要求配置里必须带一个order_by字段，并且优先用主键id作为排序键，这是最稳妥的方案。

2.4 数据类型预处理：防止写进 Excel 就变味儿

数据库里的数据类型跟 Excel 单元格格式不是一一对应的，下面这几个处理必须在写入前完成：

def prepare_dataframe(df: pd.DataFrame, datetime_cols: list = None, long_text_cols: list = None) -> pd.DataFrame: """ datetime_cols: 需要显式保留日期时间的列名列表 long_text_cols: 可能超过 32767 字符的长文本列 """ if datetime_cols: for col in datetime_cols: if col in df.columns: df[col] = pd.to_datetime(df[col]).dt.strftime("%Y-%m-%d %H:%M:%S") if long_text_cols: for col in long_text_cols: if col in df.columns: df[col] = df[col].astype(str).str.slice(0, 32000) # 把 id 这类超长数字列显式转成字符串，避免科学计数法 for col in df.select_dtypes(include=["int64", "float64"]).columns: if col.lower().endswith(("id", "no", "num")): df[col] = df[col].astype(str) return df

这里有三件事值得展开：

• 日期转字符串：pandas 写 datetime 到 Excel 时，如果 xlsxwriter 的格式化没配对，单元格显示的是一串数字（Excel 日期本质是序列号）。我选择直接在 SQL 查询或 pandas 里把日期格式化成标准字符串，一劳永逸。
• 超长文本截断：截断到 32000 而不是 32767，留出一点余量，避免多字节字符在边界被截断产生乱码。
• 长数字转字符串：这招防的是手机号、雪花 ID 这类超过 15 位的纯数字，如果不转字符串，它们在 Excel 里会变成科学计数法，看起来像4.10217E+18，精度还丢了。

2.5 完整调用示例

把上面的函数串起来，实际使用长这样：

if __name__ == "__main__": engine = build_engine(database="business_db") tables = [ { "sheet_name": "订单明细", "query": """ SELECT id, trade_date, user_id, product_name, remark FROM order_records WHERE trade_date >= :start_date """, "params": {"start_date": "2024-01-01"}, "order_by": "id", }, { "sheet_name": "用户信息", "query": "SELECT id, user_name, register_date, level FROM user_info", "params": {}, "order_by": "id", }, ] export_tables_to_excel( engine, tables_config=tables, output_path="/data/export/business_2024.xlsx", chunksize=50000, sheet_limit=400000, )

这里我把sheet_limit设成 40 万而不是 Excel 上限 104 万，是有意的。超过 40 万行的 sheet，Excel 打开已经明显卡顿，与其交给业务方一个"能开但很慢"的大文件，不如在导出时就按更保守的阈值切分。

3. 数据量特别大时的优化：分块、并发与内存控制

当单表数据达到千万级别，前面那个"单进程、串行分块"的方案还是不够快。这一章说三个优化手段，按性价比排序。

3.1 内存估算：为什么不能一次性全量读取

很多人写导出脚本的第一版都是pd.read_sql(query, engine)一把梭，几百万行数据全读进内存。我们来粗算一下：一行订单记录假设有 20 个字段，平均每个字段算 20 字节，500 万行就是500万 × 20 × 20 = 2亿字节，约 200 MB。看着还好，但 pandas 为了类型推断会临时复制多份副本，实际峰值可能是这个数的 3 到 4 倍。再加上 DataFrame 索引、字符串对象本身需要的内存，一个 500 万行的 DataFrame 吃 1 GB 内存很正常。

如果服务器上还跑着其他服务，这基本就是 OOM 的前奏了。

所以我坚持在任何生产导出脚本里都做分块读取，既为了控制内存，也为了能在每块之间打印进度。

3.2 分块大小怎么定

分块大小chunksize是个需要平衡的参数。我自己的经验公式：

• 行平均宽度小（比如只有几个整数列），可以设 10 万到 20 万。
• 行平均宽度大（有长文本、JSON、多字符串列），建议设 2 万到 5 万。
• 宁可块数多，也不要让单块内存超过 200 MB。

实际压测时，我会在分块读取的循环里监控一下内存变化：

import tracemalloc tracemalloc.start() # 分块循环读取... current, peak = tracemalloc.get_traced_memory() print(f"当前内存: {current / 1024 / 1024:.1f} MB, 峰值: {peak / 1024 / 1024:.1f} MB")

这里的核心原则是：单块读完了、写完了，内存就该释放。如果发现内存只增不减，优先检查是不是有人把每个分块都append到了一个总 DataFrame 里——这是最常见的隐性内存泄漏。

3.3 多进程并行：不同表并行，单表按 ID 范围切分

如果你的数据库服务器 CPU 核数多，可以并行导出多张独立的表。Python 的多线程在 IO 密集型任务上有用，但这里的主要瓶颈是数据库查询后的 pandas 内存处理，属于 CPU 和内存密集型，所以用multiprocessing更合适。

多进程最简单的写法是concurrent.futures.ProcessPoolExecutor：

from concurrent.futures import ProcessPoolExecutor, as_completed def export_one_table(cfg: dict): """导出单张表，独立进程执行""" engine = build_engine() # 每个进程单独建连接 return export_tables_to_excel( engine, [cfg], cfg["output_path"], chunksize=cfg.get("chunksize", 50000), sheet_limit=cfg.get("sheet_limit", 400000), ) def parallel_export(all_configs: list): with ProcessPoolExecutor(max_workers=4) as pool: futures = [pool.submit(export_one_table, cfg) for cfg in all_configs] for future in as_completed(futures): print(future.result())

注意每个子进程必须自己创建数据库连接和ExcelWriter，不能把父进程的 engine 直接传进去。数据库连接不能跨进程共享，这是很多人踩过坑的地方。

如果是单张表特别大，多进程思路要反过来：按主键id的范围切成 N 段，每个进程导出[id_min, id_max)区间。切分方式先查一下表的最小和最大主键：

SELECT MIN(id), MAX(id), COUNT(*) FROM order_records;

然后根据总数和期望的每段行数算出切割点，再按切割点生成各自的WHERE id BETWEEN ... AND ...查询。这个方案的好处是天然支持断点续传——某个区间跑挂了，重跑那个区间即可。

3.4 写入引擎的选择：xlsxwriter 的 write_only 模式

pandas.DataFrame.to_excel支持传入不同的 engine，openpyxl和xlsxwriter对同一份数据的写入性能差距明显。

xlsxwriter有一个 write-only 模式，专门为大数据量设计：

workbook = xlsxwriter.Workbook(output_path, {"constant_memory": True}) worksheet = workbook.add_worksheet("订单明细")

开启这个模式后，xlsxwriter 不会在内存里维护完整的行数据，而是"边写边落盘"，内存占用大幅下降。唯一的不方便是不能回改已经写过的单元格——但导出任务根本不需要回改，所以完全适配。

在pandas.ExcelWriter里启用方式也简单：

writer = pd.ExcelWriter(output_path, engine="xlsxwriter", engine_kwargs={"options": {"constant_memory": True}})

我之前做过一次对比测试：同样是 80 万行、15 列的数据，openpyxl写入耗时接近 3 分钟，xlsxwriter的 constant_memory 模式只要 40 秒左右。这个差距在生产任务里非常可观。

4. 我把这些 Excel 发给业务方后，踩过的那些坑

代码能跑通只是第一步。真实世界里"能把文件生成出来"和"业务方能正常使用"之间，隔着好几个大坑。这一章记录我实际交付过程中踩过的坑和对应的处理办法。

4.1 单 Sheet 超过 104 万行：不是报错，但比报错更麻烦

有一次我导出一张日志表，数据量 130 多万行，我满怀信心地等着文件生成，结果确实生成了，但 Excel 打开时报"文件损坏"，原因是超过了单 Sheet 上限。

这个坑最坑的地方在于：文件层面已经写坏了，但导出脚本没有任何异常提示。xlsxwriter 会把写到 1048577 行之后的逻辑线强行掐掉，但文件打开前你永远不知道。

我的处理办法是在写入循环里做一个硬检查：

MAX_EXCEL_ROWS = 1048576 if sheet_row + len(df) > MAX_EXCEL_ROWS: # 当前 sheet 已满，另起一个 sheet 或另起文件 ...

更稳妥的做法是：在方案设计阶段就按业务数据的增长趋势预估未来的行数，提前把一个大表的数据按时间维度分文件导出。比如日志表就按天导出，一天一个文件，既避免行数问题，业务方拿数据也更方便。

4.2 数字被显示成科学计数法的三个来源

这个问题太常见了，常见到每个做导出的人都会遇到。

• 超长 ID：19 位的雪花 ID，Excel 显示成4.10217E+18，而且后几位被当成 0，原始精度丢失。
• 手机号：11 位数字理论上不超过 Excel 的 15 位精度限制，但很多手机号第 11 位附近会出现奇怪的浮点误差，导出后末尾变成12300这种。
• 带前导零的编码：比如商品编码001234，Excel 会自作主张把前导零省略。

解决思路分两层：

第一层，在 SQL 查询阶段就转成字符串。比如 MySQL 用CAST(id AS CHAR)，PostgreSQL 用id::text，让数据库返回字符串而不是数字，pandas 拿到手就是字符串，再写入 Excel 就不会触发科学计数法。

第二层，如果数据已经以数字类型到了 pandas 里，就用df[col] = df[col].astype(str)转一次。注意此时如果数字已经是浮点型，直接转字符串会带.0后缀，要先转int64再转str，或者用小数点后 0 位格式化。

4.3 日期显示成数字或 Unix 时间戳

pandas 写 datetime 到 xlsxwriter 时，如果不设置单元格格式，日期会显示成45087.3456这种序列数。我一开始没做特殊处理，业务方拿到文件后发给我一堆问号，"这个数字是什么？"

后来我总结出一个最省心的方案：所有日期列在写入前统一格式化成YYYY-MM-DD HH:MM:SS字符串。

df["trade_date"] = pd.to_datetime(df["trade_date"]).dt.strftime("%Y-%m-%d %H:%M:%S")

字符串日期写入 Excel 后，是左对齐的文本格式，虽然不能直接在 Excel 里做日期运算，但对于"导出给业务方看"这个场景完全够用。如果有分析师需要日期参与计算，他们自己用分列功能转一下批量也很方便。

4.4 乱码问题：谁都可能遇到，但原因各不同

乱码的排查路径我这几年走得很熟练了，按位置分成三种：

• 数据库读出来就是乱码：大概率是连接串没加charset=utf8mb4，或者表字段本身的字符集不对。
• pandas 阶段乱码：检查拿到的 DataFrame 里打印出来的值是正常的，但写出去是乱码，多半是 ExcelWriter 的编码设置被动了，或者用了 CSV 输出时没指定utf-8-sig。
• Excel 打开乱码但代码输出正常：这个最常见，就是 CSV 文件没有 BOM 头。Excel 默认用 GBK 编码读 CSV，没有 BOM 的 UTF-8 文件它识别不出来。

CSV 导出时一定要用df.to_csv(path, index=False, encoding="utf-8-sig")，utf-8-sig会自动带上 BOM，Excel 就能正确识别。

4.5 一个"慢得无法忍受"的文件，性能瓶颈可能在单元格格式

我遇到过这么一次：数据量不大，才 30 万行，但导出耗时非常长，一看代码，循环里对每个单元格调用了worksheet.write_number()，后面还跟了一堆set_format()。

naive 写法的性能杀手是"逐单元格操作 + 格式对象绑定"。优化办法是：

• 能用pandas的批量写入，就不要自己循环写单元格。
• 如果要设置列宽、表头样式，只对列名行做格式化，数据行保持默认。
• 表头样式可以用worksheet.set_row(0, None, header_format)这种批量方式，而不是每个单元格单独设置。

这个优化做完，同样的 30 万行数据，耗时从 4 分多钟降到了 20 秒以内，效果立竿见影。

5. 让导出任务变成可运维的定时任务：断点续传与失败重试

一次性的手工导出很简单，真正的难点在于把这个脚本变成"每天自动跑、挂了能恢复、跑完有日志"的生产级定时任务。这一章是我在交付给其他团队后，根据使用反馈迭代出的工程化改进。

5.1 断点续传：以"已导出的最大主键"为锚点

最朴素的断点续传方案是记录上一次导出的最大主键值。下次任务启动时，SQL 条件从id > last_max_id开始查，而不是从全表开头查。

def get_last_checkpoint(state_file: str) -> int: try: with open(state_file, "r") as f: return int(f.read().strip()) except (FileNotFoundError, ValueError): return 0 def save_checkpoint(state_file: str, last_id: int): with open(state_file, "w") as f: f.write(str(last_id))

这个方案有几个前提：

• 表必须有单调递增的主键（或者至少有一个递增的业务时间字段）。
• 每次任务只追加新数据，不修改历史数据。
• 增量导出的数据范围是连续的区间，不存在"越过空白段后还有新数据"的情况。

如果表结构不满足这些前提，断点续传的意义就大打折扣，这种情况建议改为按时间窗口全量重导，或者把增量标志位交给业务系统维护。

5.2 失败重试：指数退避，而不是立即硬冲

数据库连接抖动、锁竞争、网络闪断，这些都是定时任务最常遇到的故障类型。如果失败后立即重试，很可能在同一瞬间再次失败。

我习惯用带指数退避的重试逻辑：

import time def retry_call(func, retries=3, base_delay=2): for attempt in range(retries): try: return func() except Exception as exc: if attempt == retries - 1: raise delay = base_delay * (2 ** attempt) print(f"第 {attempt + 1} 次失败，{delay}s 后重试: {exc}") time.sleep(delay)

每次失败的间隔按 2 秒、4 秒、8 秒指数增长，给数据库留出恢复窗口。同时记录最后失败位置，让下一次任务可以从断点继续。

5.3 日志是导出任务的灵魂

我接手过别人的导出脚本，代码结尾只有一行print("done")。出问题时完全不知道数据导到哪一步了、出了多少行、卡在哪个表上。

我的日志规范很简单：

• 每个表开始导出前，记录表名、预计数据量、开始时间。
• 每个分块导出后，记录已导出行数、耗时、当前主键边界。
• 任务结束或异常退出，统一写一个 summary：每个表导出了多少行、总耗时、是否有跳过和重试。

import logging import time logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.FileHandler("/data/export/export.log")] ) logger = logging.getLogger("exporter") def export_with_log(engine, cfg): start = time.time() total = 0 logger.info(f"开始导出表 {cfg['sheet_name']}") # ... 分块导出循环 ... logger.info(f"表 {cfg['sheet_name']} 完成，共 {total} 行，耗时 {time.time() - start:.2f}s")

有了这个日志，任何一次任务异常都能快速定位：是数据库查询太慢、某一行的数据格式有问题、还是写 Excel 时文件被占用。日志不求花哨，但必须能让接手的同事在 5 分钟内定位到问题。

5.4 定时调度的落地姿势

我用 cron 表达式做调度，核心命令很简单：

0 2 * * * /usr/bin/python3 /opt/scripts/exporter.py >> /data/export/cron.log 2>&1

凌晨两点跑，业务低谷，数据库压力小，文件生成后也不影响上班时间取数。

这里有个细节：不要直接用python exporter.py，而是写一个启动脚本，先激活虚拟环境再跑。项目依赖如果装在全局环境，某天其他人动了依赖版本，整个定时任务就无声无息地挂了。

我之前为此定了一条规矩：任何导出任务的启动命令都固定为cd /opt/scripts && source .venv/bin/activate && python exporter.py，cron 里写全路径，不要依赖默认环境的任何东西。

定时任务跑成功后，还可以加一个可选的发送通知环节：把导出的文件上传到公司的文件平台或对象存储，然后把下载链接发给业务群。这一步看团队基础设施，有空实现一下价值很大，因为业务方不用每次找运维或者开发要文件了。

6. 一些值得再抠一抠的细节：格式美化与文件命名

导出任务做到这里已经是可用的生产工具了，但要让业务方用得舒服，还有几个小细节值得花时间优化。

6.1 列宽自适应：别让业务方每次手动拉列宽

xlsxwriter 里可以给 Sheet 设置默认列宽，或者针对特定列设置更宽的值：

worksheet.set_column("A:A", 12) # ID 列 worksheet.set_column("B:B", 20) # 日期列 worksheet.set_column("C:C", 30) # 名称列

如果列很多不想一个个指定，可以按列名的最大字符长度估算：

for i, col in enumerate(df.columns): max_len = max(df[col].astype(str).str.len().max(), len(col)) + 2 worksheet.set_column(i, i, min(max_len, 50))

这个操作对使用者体验提升非常明显，值得做。

6.2 表头加粗与冻结首行

冻结首行这个功能几乎是业务方最喜欢的一个。表数据一多，往下拉的时候完全不知道每一列是什么，冻结首行后这个痛点直接消失。

header_format = workbook.add_format({"bold": True, "bg_color": "#D9E1F2", "border": 1}) worksheet.set_row(0, None, header_format) worksheet.freeze_panes(1, 0)

一次设置，永久生效，整个文件的专业度立刻上一个档次。

6.3 文件命名规则：时间戳 + 业务标识

文件命名看起来小，但影响的是整个交付链路。我经常看到有人导出文件叫export.xlsx，结果第二天再导出就把昨天的文件覆盖了，或者业务方同时拿到 5 个export.xlsx根本分不清哪个是哪个。

我的命名规范是：业务日期_数据范围_生成时间戳.xlsx，比如：

order_records_20240101_20240630_20240701020000.xlsx

保证文件名里能看出数据的时间范围、生成的时间点，且不会互相覆盖。如果是定时增量任务，这个习惯能帮你少背很多锅。

6.4 空数据表的处理：宁可给空模板，也别少一个 Sheet

业务方用 Excel 最怕的情况是：今天导出数据恰好是 0 行，脚本直接跳过了这个 sheet，业务方以为程序出 bug 了，实际上只是没数据。

我的处理逻辑是：如果查询结果为空，也创建 Sheet，只写入列名和一行"无数据"说明。这样业务方一眼就明白"这个时间段确实没有数据"，而不是"工具坏了"。

if df.empty: empty_df = pd.DataFrame(columns=cfg["columns"]) empty_df.to_excel(writer, sheet_name=sheet_name, index=False)

小小的一个坚持，省掉了很多不必要的需求沟通。

7. 从"能跑"到"好用"的个人经验总结

最后说说我做完这个导出工具后的几点体会。

导出这个功能看起来是人人都能做的最小功能，但真正能稳定服务业务的导出工具，一定不是pd.read_sql + to_excel几行代码就算完成的。数据量边界、格式兼容性、异常恢复、可观测性、使用体验，缺一不可。尤其是在数据量大、任务频繁的场景下，一个没有分块、没有日志、没有断点的导出脚本，迟早会变成半夜的告警电话来源。

我自己的习惯是：任何导出任务都先问三个问题——数据量级是多少？多久跑一次？谁在用这个文件？数据量决定要不要分块，频率决定要不要断点续传和定时调度，使用者的技术水平决定我要不要在格式上多花心思。这三个问题问清楚了，方案就自然浮出水面。

如果看完这篇你还是只想抄一个最小可用的版本，那至少记得：导出前先做数据类型预处理，导出时分块控制内存，导出后确认 sheet 行数没超上限。这三件事做到位，你的导出脚本就能超过大多数临时脚本，成为别人愿意持续使用的工具。