第 22 章 数据读写进阶：Excel 复杂操作、json_normalize 与大数据实战

> [第 4 章](04-数据读取与写入.md)讲了基础读写。本篇进阶：**Excel 样式与复杂写盘、多层嵌套 JSON 展开（json_normalize）、大文件分块/追加、数据库优化、Parquet 工程化、网页表格抓取**，覆盖真实项目中的高频读写场景。 --- ## 目录 1. [Excel 进阶：样式、多表、性能](#1-excel-进阶样式多表性能) 2. [json_normalize：展开嵌套 JSON](#2-json_normalize展开嵌套-json) 3. [大 CSV 分块处理与追加](#3-大-csv-分块处理与追加) 4. [Parquet / Feather 工程化](#4-parquet--feather-工程化) 5. [数据库读写优化](#5-数据库读写优化) 6. [网页表格抓取](#6-网页表格抓取) 7. [编码与异常兜底](#7-编码与异常兜底) 8. [读写实战：完整数据管道](#8-读写实战完整数据管道) 9. [常见坑与注意事项](#9-常见坑与注意事项) 10. [本章小结与练习](#10-本章小结与练习) --- ## 1. Excel 进阶：样式、多表、性能 ### 1.1 用 openpyxl 直接写样式 ```python import pandas as pd from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils.dataframe import dataframe_to_rows df = pd.DataFrame({ "产品": ["苹果", "香蕉", "橙子"], "销量": [120, 80, 95], "销售额": [600, 240, 380], }) wb = Workbook() ws = wb.active ws.title = "销售报表" # 写入表头 header_font = Font(bold=True, color="FFFFFF") header_fill = PatternFill(start_color="2E86AB", end_color="2E86AB", fill_type="solid") for col_idx, col_name in enumerate(df.columns, start=1): cell = ws.cell(row=1, column=col_idx, value=col_name) cell.font = header_font cell.fill = header_fill cell.alignment = Alignment(horizontal="center") # 写入数据 for r, row in df.iterrows(): for c, val in enumerate(row, start=1): ws.cell(row=r + 2, column=c, value=val) # 列宽 for col in ws.columns: ws.column_dimensions[col[0].column_letter].width = 12 wb.save("样式报表.xlsx") ``` ### 1.2 用 pandas + openpyxl 引擎配合 ```python # pandas 写多表 + 加总 with pd.ExcelWriter("多表.xlsx", engine="openpyxl") as writer: df.to_excel(writer, sheet_name="明细", index=False) df.groupby("产品")["销量"].sum().to_frame("总销量").to_excel( writer, sheet_name="汇总", index=True ) # 读取时跳过某些行 / 指定区间 df2 = pd.read_excel("多表.xlsx", sheet_name="汇总", header=0) # 读取时处理合并单元格（header 前几行有标题） df3 = pd.read_excel("report.xlsx", header=1) # 跳过第一行合并标题 ``` ### 1.3 Excel 性能注意 | 场景 | 建议 | |------|------| | 大量数据写 Excel | 用 `to_excel` 一次写完，别循环写入单元格 | | 读大 Excel | 指定 `usecols`、`dtype` | | 只需要数据 | 优先 CSV/Parquet，Excel 仅用于交付 | ```python # 大 Excel 读取优化 big = pd.read_excel("big.xlsx", usecols=["id", "金额"], dtype={"id": "int32"}) ``` --- ## 2. json_normalize：展开嵌套 JSON API 返回的 JSON 经常多层嵌套，`pd.json_normalize` 能一键"拍平"： ```python import pandas as pd # 嵌套结构示例：用户 + 地址对象 + 订单列表 data = [ {"id": 1, "name": "张三", "address": {"city": "北京", "district": "朝阳"}, "orders": [{"no": "A1", "amount": 100}, {"no": "A2", "amount": 200}]}, {"id": 2, "name": "李四", "address": {"city": "上海", "district": "浦东"}, "orders": [{"no": "B1", "amount": 300}]}, ] # 拍平一级嵌套（address 展开为列） df_users = pd.json_normalize(data) print(df_users) # id name address.city address.district orders # 0 1 张三 北京 朝阳 [{'no': 'A1', ...}] # 1 2 李四 上海 浦东 [{'no': 'B1', ...}] # 展开订单列表（record_path），并保留用户信息（meta） df_orders = pd.json_normalize( data, record_path="orders", # 要展开的列表字段 meta=["id", "name"], # 保留的外层字段 meta_prefix="user_", # 前缀防冲突 ) print(df_orders) # no amount user_id user_name # 0 A1 100 1 张三 # 1 A2 200 1 张三 # 2 B1 300 2 李四 # 多层嵌套（meta 里再指定路径） df_orders2 = pd.json_normalize( data, record_path="orders", meta=[["address", "city"]], meta_prefix="addr_", ) ``` | 参数 | 作用 | |------|------| | `record_path` | 要展开的列表字段路径 | | `meta` | 保留的外层字段（可嵌套列表路径） | | `meta_prefix` | 外层字段加前缀，防列名冲突 | | `errors` | 缺失路径时的处理 | ### 通用 JSON 展开工具 ```python import json with open("api_response.json", encoding="utf-8") as f: raw = json.load(f) # 常见结构：{"code":0, "data": {...}} if isinstance(raw, dict) and "data" in raw: df = pd.json_normalize(raw["data"]) else: df = pd.json_normalize(raw) ``` --- ## 3. 大 CSV 分块处理与追加 ### 3.1 分块读取 + 增量聚合 ```python import pandas as pd # 逐块读取并聚合 from collections import defaultdict totals = defaultdict(float) counts = defaultdict(int) for chunk in pd.read_csv("big.csv", chunksize=100_000, usecols=["城市", "金额"]): for city, amt in chunk.groupby("城市")["金额"].sum().items(): totals[city] += amt counts[city] += 1 result = pd.DataFrame({"总金额": totals, "块数": counts}) print(result) ``` ### 3.2 分块写出 / 追加 ```python # 方式 1：分批追加写（mode='a'） for chunk in pd.read_csv("big.csv", chunksize=100_000): chunk.to_csv("out.csv", mode="a", header=not __file__.endswith("0"), index=False) # 方式 2：concat 后一次写（小数据量） frames = [pd.read_csv(f) for f in ["a.csv", "b.csv"]] pd.concat(frames, ignore_index=True).to_csv("merged.csv", index=False) # 方式 3：追加到已有文件（注意 header 控制） df1.to_csv("all.csv", index=False) df2.to_csv("all.csv", mode="a", header=False, index=False) ``` > **坑**：追加写入时第二次起必须 `header=False`，否则表头重复。 --- ## 4. Parquet / Feather 工程化 ```python # 安装：pip install pyarrow # 写读 df.to_parquet("data.parquet", index=False) df = pd.read_parquet("data.parquet") # 压缩方式 df.to_parquet("data_snappy.parquet", compression="snappy") # 默认，快 df.to_parquet("data_gzip.parquet", compression="gzip") # 压缩率高 df.to_parquet("data_zstd.parquet", compression="zstd") # 均衡 # 只读部分列（列式存储优势） df_part = pd.read_parquet("data.parquet", columns=["id", "金额"]) # Feather 更快的轻量格式 df.to_feather("data.feather") df = pd.read_feather("data.feather") ``` **为什么工程推荐 Parquet**： | 对比 | CSV | Parquet | |------|-----|---------| | 读速度 | 慢（全量解析） | 快（列式 + 谓词下推） | | 大小 | 大 | 压缩后小很多 | | 类型 | 丢失（都变字符串） | 完整保留 dtype | | 分列读取 | 不支持 | 支持 | ```python # 生产常用：分区目录 df.to_parquet("sales.parquet", partition_cols=["月份"]) # 生成 sales.parquet/月份=2025-01/xxx.parquet ``` --- ## 5. 数据库读写优化 ```python from sqlalchemy import create_engine import pandas as pd engine = create_engine("sqlite:///app.db") # 分块写入（大表） df.to_sql("big_table", engine, if_exists="append", index=False, chunksize=10000) # 只读需要的列 df = pd.read_sql("SELECT id, 金额 FROM orders WHERE 金额 > 100", engine) # 参数化查询（防注入） df = pd.read_sql( "SELECT * FROM orders WHERE city = :c AND date >= :d", engine, params={"c": "上海", "d": "2025-01-01"}, ) # 写临时表做复杂计算 df.to_sql("tmp_t", engine, if_exists="replace", index=False) result = pd.read_sql(""" SELECT region, SUM(amount) AS total FROM tmp_t GROUP BY region """, engine) ``` > **安全**：永远用 `params` 传参，不要拼 SQL 字符串，防注入（参考系统安全要求）。 --- ## 6. 网页表格抓取 ```python # 依赖：pip install lxml html5lib beautifulsoup4 # 抓取页面所有表格 tables = pd.read_html("") print(len(tables)) # 按关键词匹配表格 for t in tables: if "销售额" in str(t.columns): df = t break # 抓取时指定匹配 df = pd.read_html("", match="销量")[0] # 从 HTML 字符串读取 html_str = "

a
1

" df = pd.read_html(html_str)[0] print(df) ``` --- ## 7. 编码与异常兜底 ```python import chardet def read_csv_any(path): """自动探测编码读取""" with open(path, "rb") as f: enc = chardet.detect(f.read(10000))["encoding"] or "utf-8" try: return pd.read_csv(path, encoding=enc) except Exception: return pd.read_csv(path, encoding="utf-8", errors="replace") df = read_csv_any("未知编码.csv") ``` ```python # 读取失败兜底 def safe_read(path, **kwargs): for enc in ["utf-8", "gbk", "gb18030", "utf-8-sig", "latin1"]: try: return pd.read_csv(path, encoding=enc, **kwargs) except UnicodeDecodeError: continue raise ValueError(f"无法解析文件编码: {path}") ``` --- ## 8. 读写实战：完整数据管道 ```python import pandas as pd from sqlalchemy import create_engine # 1. 多源汇入 orders = pd.read_csv("orders.csv", encoding="utf-8") users = pd.read_json("users.json", orient="records") # 2. 清洗 orders["金额"] = pd.to_numeric(orders["金额"], errors="coerce") orders = orders.dropna(subset=["金额"]) orders["日期"] = pd.to_datetime(orders["日期"], errors="coerce") # 3. 合并 df = orders.merge(users[["id", "城市"]], left_on="用户ID", right_on="id", how="left") # 4. 中间结果落盘（Parquet 保类型） df.to_parquet("stage/clean_orders.parquet", index=False) # 5. 汇总入库 engine = create_engine("sqlite:///warehouse.db") summary = df.groupby(["城市", "日期"], as_index=False)["金额"].sum() summary.to_sql("daily_summary", engine, if_exists="append", index=False) # 6. 导出交付 with pd.ExcelWriter("日报.xlsx") as writer: summary.head(1000).to_excel(writer, sheet_name="日报", index=False) summary.groupby("城市")["金额"].sum().to_frame("累计").to_excel( writer, sheet_name="城市汇总" ) ``` --- ## 9. 常见坑与注意事项 | 坑 | 现象 | 解决办法 | |----|------|----------| | 追加写入重复表头 | 文件多一行列名 | 第二次起 `header=False` | | json_normalize 路径错 | KeyError/空表 | 先 `type(raw)`、`raw.keys()` 探查结构 | | read_html 缺依赖 | ModuleNotFoundError | `pip install lxml html5lib` | | 大表 to_excel 慢 | 卡顿 | 改用 CSV/Parquet，Excel 只交付小数据 | | SQL 拼接注入 | 安全风险 | 用 `params` 参数化 | | 未知编码 | UnicodeDecodeError | chardet 探测 + 多编码兜底 | --- ## 10. 本章小结与练习 ### 小结 - Excel：`openpyxl` 写样式；pandas 一次写多表； - JSON：`json_normalize(record_path=, meta=)` 展开嵌套； - 大 CSV：`chunksize` 分块、`mode="a"` 追加； - Parquet：列式、保类型、分列读取，工程首选； - 数据库：`chunksize` 批量、`params` 参数化防注入； - 网页：`read_html(match=)` 抓表格。 ### 练习题 1. 用 json_normalize 展开一份含嵌套地址和订单列表的 JSON。 2. 将 10 万行数据分块读取并统计某列总和。 3. 用 openpyxl 给一张表加表头样式和列宽并保存。 4. 写一个 `read_csv_auto(path)`：自动探测编码并读取。 5. 把两份 CSV 合并后写入 SQLite，再用 SQL 查询汇总。 --- > **下一篇预告**：[第 23 章 向量化性能优化大全](23-向量化性能优化大全.md) —— 告别 apply，全面拥抱向量化。