一次性脚本和数据管道脚本的差别不在代码量，而在输入变了会怎样、跑到一半挂了会怎样、再跑一次会怎样。前者写完就扔，后者要在没人盯着的时候重复执行几百次。这篇按"能反复跑"的标准拆四件事：编码、缺失、规模、失败处理。
1. 能跑一次与能反复跑
| 维度 | 一次性脚本 | 可反复跑的脚本 |
|---|---|---|
| 幂等 | 追加写入，跑两次翻倍 | 按主键覆盖（upsert） |
| 断点 | 挂了从头来 | 记进度，跳过已完成分片 |
| 输入漂移 | 列名变了 KeyError |
校验列集合，缺列报可读错误 |
| 部分失败 | 中断整批 | 失败落盘，其余继续 |
| 输出 | 直接写目标 | 临时文件后原子替换 |
| 观测 | 只 print |
日志 + 进度计数 |
自检只需一个问题：把昨天的输出当今天的输入再跑一遍，会崩还是会收敛？
2. 编码：显式永远胜过默认
open() 不传 encoding 时用的是 locale.getpreferredencoding(False)：Linux 容器里一般是 UTF-8，中文 Windows 上常是 cp936。同一份 CSV 在两台机器上读出不同乱码，根因就在这里（CPython 3.11 仍是这个行为）。
| 场景 | 该写什么 | 不写会怎样 |
|---|---|---|
| 读外部 CSV | encoding="utf-8-sig" |
首列名带 \ufeff，df["id"] 找不到列 |
| 写给人看的 CSV | utf-8-sig + newline="" |
Excel 里中文乱码；行尾多空行 |
| 读写内部数据 | encoding="utf-8" |
平台差异 |
| 不确定来源 | errors="replace" + 统计替换数 |
静默丢字符，下游对不上账 |
import pathlib
def read_text(path: str) -> tuple[str, int]:
raw = pathlib.Path(path).read_bytes()
for enc in ("utf-8-sig", "gb18030"):
try:
return raw.decode(enc), 0
except UnicodeDecodeError:
pass
text = raw.decode("utf-8", errors="replace")
return text, text.count("\ufffd") # 替换数必须落盘，问题才不被藏起来
errors="replace" 的代价是把解码失败变成静默错误：数据读进来了，字段却已变成 �，拿它做主键就丢行。用它必须统计替换数。
3. 缺失值的三种形态
| 形态 | 谁产生 | isna() |
== None |
备注 |
|---|---|---|---|---|
None |
Python 层、object 列 |
True | True | 语义最干净 |
NaN |
数值列、read_csv 空单元格 |
True | False | NaN != NaN 恒成立 |
pd.NA |
可空整数/布尔（Int64） |
True | False | 比较返回 pd.NA |
"" 空串 |
CSV、手工清洗 | False | False | 要单独 (s == "").sum() |
import pandas as pd
df = pd.DataFrame({"a": [1.0, None, float("nan")], "s": ["x", "", None]})
print(df.isna().sum().to_dict()) # {'a': 2, 's': 1}
print((df["s"] == "").sum()) # 1：空串不算缺失
print(df["a"].eq(None).sum()) # 0：NaN 与 None 不等
# 清洗：先统一成 None，再落业务默认值
df["s"] = df["s"].replace("", None).fillna("unknown")
print(df["s"].tolist()) # ['x', 'unknown', 'unknown']
规则：判缺失只用 isna()/notna()，别用 == None，也别用 != ""。三种形态混在一列时，先统一成一种。
4. 类型推断的陷阱
| 现象 | 原因 | 处理 |
|---|---|---|
ID 前导零消失（007 → 7） |
列被推断成 int64 |
dtype={"id": "string"} |
整列变 object |
混了数字与字符串 | 显式 dtype=str 后业务层转换 |
空值列变 float64 |
整数列有 NaN 只能用浮点 |
dtype="Int64" |
| 日期变字符串 | 未指定 parse_dates |
parse_dates=["ts"] |
"NA" 被当缺失 |
默认 na_values 含 NA、NULL |
keep_default_na=False, na_values=[""] |
import pandas as pd
df = pd.read_csv(
"orders.csv",
dtype={"order_id": "string", "qty": "Int64", "amount": "float64"},
usecols=["order_id", "qty", "amount", "ts"],
parse_dates=["ts"],
keep_default_na=False, na_values=[""],
)
print(df.dtypes.to_dict())
print(df["order_id"].head(3).tolist()) # 前导零保留
keep_default_na=False 是双刃剑：挡住"字符串 NA 被误判成缺失"，也让真空值变 ""，必须配 na_values=[""] 补回。读入阶段定死 dtype 比事后 astype 安全。
5. 大文件：不要一次读进内存
| 手段 | 适用 | 代价 |
|---|---|---|
chunksize=N |
逐批处理、边读边写 | 跨批聚合要自己维护状态 |
usecols=[...] |
列多但只用几列 | 无 |
dtype=... |
降内存（float64 → float32） |
精度损失 |
csv 模块逐行 |
极简 ETL | 无向量化收益 |
循环写法就是 for i, chunk in enumerate(pd.read_csv("big.csv", chunksize=200_000))，逐块 process 后写出，块内务必 .copy() 再赋值，否则会踩到视图告警。
分块写要盯住 mode="a" 与 header=(i == 0) 的配合，否则要么表头重复要么丢表头。更稳的是每块写独立分片再合并，中途失败好定位。
6. 并行：先分清瓶颈在等还是在算
| 瓶颈 | 选择 | 原因 |
|---|---|---|
| 网络/磁盘等待 | ThreadPoolExecutor |
等的时候 GIL 已释放 |
| 纯 Python 计算 | ProcessPoolExecutor |
3.11 有 GIL，线程不并行执行字节码 |
| 大量小文件解析 | 线程池 + 进程池两级 | 分开调，别混一个池 |
from concurrent.futures import ProcessPoolExecutor, as_completed
import pathlib
def parse_one(path: str) -> tuple[str, int]:
return path, pathlib.Path(path).stat().st_size
files = [str(p) for p in pathlib.Path("data").glob("*.csv")]
with ProcessPoolExecutor(max_workers=4) as pool:
futs = [pool.submit(parse_one, f) for f in files]
for fut in as_completed(futs):
try:
print(fut.result())
except Exception as exc: # 单个失败不拖垮整批
print("failed:", exc)
上面的池要在 if __name__ == "__main__": 里创建（spawn 平台必需）。隐性成本：每次 submit 都要 pickle 参数与返回值，传 DataFrame 等于复制内存；Windows/macOS 默认 spawn，模块顶层不能有副作用；子进程不共享全局状态。单任务只有几十毫秒时，启动与序列化会吃掉全部收益。
7. 失败要落盘，不要中断整批
import csv, datetime
def run_batch(items, handle):
ok, failed = 0, []
with open("failed.csv", "w", newline="", encoding="utf-8") as log:
w = csv.writer(log)
w.writerow(["key", "error", "ts"])
for key in items:
try:
handle(key); ok += 1
except Exception as e:
failed.append(key)
w.writerow([key, f"{type(e).__name__}: {e}",
datetime.datetime.now().isoformat(timespec="seconds")])
w.flush() # 进程被 kill 前的记录也要保住
return ok, failed
三个细节决定日志有没有用：flush() 让进程被 kill 时记录还在；记 key 才能重跑；记时间戳才分得清本次和历史。失败清单要成为下次运行的输入。
8. 日志、进度与原子输出
| 需求 | 做法 | 反例 |
|---|---|---|
| 进度可见 | 每 N 条打印 done/total 与速率 |
只在结束打印 |
| 可被日志系统收集 | logging 到 stderr |
混在数据输出里 |
| 输出不残缺 | 写 .tmp 后 os.replace |
直接覆盖目标 |
| 幂等 | 主键 upsert 或先删同日分区 |
无脑 to_csv(mode="a") |
import os, pathlib, tempfile
def atomic_write(target: str, text: str) -> None:
fd, tmp = tempfile.mkstemp(dir=pathlib.Path(target).parent, suffix=".tmp")
with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
f.write(text)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, target) # 同一文件系统内才是原子替换
os.replace 在同一文件系统内原子，跨盘退化成复制、失去原子性。崩溃留下 .tmp 可接受，留下半个 target 才是事故。
9. 幂等地重跑：按主键覆盖而非追加
| 策略 | 幂等 | 适用 |
|---|---|---|
to_csv(mode="a") |
否 | 一次性导出 |
| 按日期分区整体重写 | 是 | 日批、月批 |
主键 upsert 进库 |
是 | 增量同步 |
drop_duplicates(subset=主键, keep="last") |
是 | 文件型落地 |
| 带运行 ID 的新文件 | 是但不收敛 | 需要留历史 |
import pandas as pd
old = pd.read_csv("clean.csv", dtype={"order_id": "string"})
new = pd.read_csv("batch.csv", dtype={"order_id": "string"})
(pd.concat([old, new], ignore_index=True)
.drop_duplicates(subset=["order_id"], keep="last")
.to_csv("clean.next.csv", index=False, encoding="utf-8"))
keep="last" 表达的是"新批次覆盖旧值"，正是重跑要的效果。主键不唯一时（一单多行明细）要换成"主键 + 行号"复合键，否则误删合法明细。
10. 排错：六种典型现场
| 现象 | 根因 | 处理 |
|---|---|---|
| 首列名带不可见前缀 | 有 BOM，用了 utf-8 |
改 utf-8-sig |
SettingWithCopyWarning |
链式赋值，改的是切片副本 | df.loc[mask, "col"] = v |
| 循环几十万行极慢 | iterrows 每行构造 Series |
向量化，或 itertuples |
fillna(0, inplace=True) 后仍空 |
inplace 返回 None，被赋给变量 |
用返回值赋值 |
| 跑完目标文件是空的 | 中途异常，直接写了目标 | 临时文件 + os.replace |
| 换机器结果不同 | 默认编码/na_values/精度差异 |
全部显式指定 |
SettingWithCopyWarning 不是"可以忽略"：你以为改了原表，其实改的是临时副本。看到就该改成 .loc，别用 filterwarnings("ignore")。
11. 边界：这些事不该用脚本硬扛
| 需求 | 更合适的工具 | 原因 |
|---|---|---|
| 超内存一个数量级 | DuckDB / Polars / Spark | pandas 全量进内存 |
| 定时调度与重试 | Airflow / cron + 幂等脚本 | 自己实现调度是重复造轮子 |
| 事务性写入 | 数据库 upsert |
文件保证不了并发一致 |
| 复杂清洗规则 | dbt / SQL | Python 规则难审计 |
| 回溯历史版本 | 数据湖 + 分区 | 覆盖式落地丢历史 |
一句话：脚本负责"这一批数据"，调度与存储负责"很多批"。
12. 自测
open()不指定encoding时，实际用的是哪个编码？为什么跨平台会不一致？NaN != NaN对"判缺失"的写法有什么直接影响？- 为什么
keep_default_na=False之后还要再写na_values=[""]？ - 什么时候该用线程池而不是进程池？GIL 在其中扮演什么角色？
inplace=True与"把返回值赋回变量"混用会出什么问题？- 为什么输出要先写临时文件再
os.replace？