S10-X06 · 数据分析脚本的写法：可复用与可复跑

一个取数脚本第一次写完，通常长这样：路径写死在第二行，时间范围写死在 SQL 里，结果 print 到终端再手工粘到表格。它跑通了一次。三个月后有人问"上个月那个数按什么口径算的"，你打开它，发现得改四处常量才能重跑，而且想不起第四处为什么那么改。

分界线其实很简单：这个脚本会不会跑第二次。会，它就需要三件事——输入由参数给、输出落到确定位置、中间态可丢弃。这篇按这三条拆开写。

1. 一次性脚本与可复用脚本的分界

判断项	一次性脚本	可复用脚本
跑第二次的可能	不会	每月/每次有新数据都会
谁执行	只有作者	同事、调度器、半年后的自己
参数怎么变	改代码	改命令行参数或配置文件
结果去哪	终端 / 剪贴板	固定路径 + 固定文件名规则
出错怎么查	重跑看	日志定位到文件与行号

命中"跑第二次"这一条，上面每一行都会跟着翻面。

2. 三件事：输入、输出、中间态

三条都是同一个目的：让"换一次参数重跑"变成等价操作。

输入：全部来自 argparse + 配置文件，代码里不出现日期、路径、账号
输出：落在 OUT_DIR，文件名带参数指纹，重复执行覆盖而非追加
中间态：写到临时目录，成功后被丢弃；失败时保留便于排查

3. `argparse` 的最小用法与配置外置

# scripts/report.py
from __future__ import annotations

import argparse
from datetime import date
from pathlib import Path

def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description="订单月度汇总")
    p.add_argument("--month", required=True, help="统计月份，格式 YYYY-MM")
    p.add_argument("--config", type=Path, default=Path("conf/report.yaml"))
    p.add_argument("--out-dir", type=Path, default=Path("out"))
    p.add_argument(
        "--mode", choices=("sample", "full"), default="sample",
        help="sample 只取 1% 抽样，full 跑全量",
    )
    p.add_argument("--limit", type=int, default=None, help="调试时限制行数")
    p.add_argument("--dry-run", action="store_true", help="只校验不写文件")
    return p

def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    if args.month > date.today().strftime("%Y-%m"):
        raise SystemExit(f"--month={args.month} 晚于当前月份")
    args.out_dir.mkdir(parents=True, exist_ok=True)
    ...
    return 0

三个取舍：

必填项用 required=True，不要靠 if args.month is None: ... 兜底，那等于把默认值藏起来。
--dry-run 必须有。数据分析脚本最常见的失误是"用错参数跑了一遍，结果文件已被覆盖"。
main(argv) 接受参数列表，测试里可直接 main(["--month", "2025-01"])，不必改 sys.argv。

配置外置放"变化慢但会变"的东西：

# conf/report.yaml
sources:
  orders: "mysql://analytics:***@db:3306/ods/orders"
rules:
  exclude_status: [9, 10]        # 已取消、已退款
  min_amount: 0.01
output:
  columns: [dt, channel, order_cnt, gmv]
  float_format: "%.2f"

放在哪里	适合放什么	不适合放什么
命令行参数	每次运行都变的东西：月份、模式、输出目录	长列表、结构化规则
配置文件	连接串、字段映射、业务口径	密码明文（用环境变量）
环境变量	密钥、集群地址	业务规则（不可版本化）
代码常量	真正不变的东西	一切"以后可能要改"的值

4. 用日志替代 `print`

print 的问题不是难看，而是不可分级、不可定位、关不掉。

import logging, sys

logging.basicConfig(
    level="INFO", stream=sys.stdout,
    format="%(asctime)s %(levelname)-7s %(name)s %(message)s",
)
log = logging.getLogger("report")
log.info("读取订单 month=%s mode=%s", month, mode)      # 参数用 %s 占位
log.warning("用户维度缺失 %d 行，已按空值填充", miss)
log.error("文件解析失败 path=%s line=%d", path, lineno)

级别	用在哪	出现在正常运行日志里
`DEBUG`	每条记录的中间值	否
`INFO`	阶段开始/结束、行数	是
`WARNING`	缺失、截断、类型转换	是
`ERROR`	本次计算不可信	是，且非 0 退出

写日志的两条硬规则：用 %s 占位而非 f-string（级别未启用时不付出格式化成本）；每行日志自带定位信息，否则事后只能猜。

5. 拆成四个函数：取数、清洗、计算、导出

一长串脚本的毛病是无法单独测试第 3 步。拆开后每步都能独立喂数据验证。

from dataclasses import dataclass
import pandas as pd

@dataclass
class Ctx:
    month: str
    mode: str
    cfg: dict

def fetch(ctx: Ctx) -> pd.DataFrame:
    """取数：只拿原始数据，不做加工。"""
    df = read_sql("SELECT * FROM orders WHERE dt LIKE %(m)s",
                  {"m": f"{ctx.month}%"})
    log.info("fetch rows=%d", len(df))
    return df

def clean(df: pd.DataFrame, cfg: dict) -> pd.DataFrame:
    """清洗：类型转换、去重、剔除无效状态。"""
    df = df.drop_duplicates(subset=["order_id"])
    df = df[~df["status"].isin(cfg["rules"]["exclude_status"])]
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
    df = df[df["amount"] >= cfg["rules"]["min_amount"]]
    return df

def compute(df: pd.DataFrame) -> pd.DataFrame:
    """计算：纯函数，输入输出都是 DataFrame，最容易被测试。"""
    return df.groupby(["dt", "channel"], as_index=False).agg(
        order_cnt=("order_id", "nunique"), gmv=("amount", "sum")
    )

def export(df: pd.DataFrame, out: Path) -> Path:
    """导出：唯一允许写文件的一步。"""
    out.parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(out, index=False, float_format="%.2f")
    return out

函数	是否碰 IO	能否单元测试
`fetch`	是（数据库）	需 mock 或测试库
`clean`	否	构造小表即可
`compute`	否	构造小表即可

原则：越靠后的函数越纯。IO 挤到两端，中间的计算就能被小样例覆盖，而小样例正是"口径变了"时最先要跑的东西。

6. 幂等：重复执行不产生重复数据

判据是：同一输入跑 N 次，结果与跑 1 次相同。

场景	非幂等写法	幂等写法
写 CSV	`mode="a"` 追加	覆盖写
写数据库	`INSERT INTO t`	`DELETE ... WHERE dt=?` 再插
写分区表	`INSERT OVERWRITE` 整表	只覆盖本次分区

def write_partition(df: pd.DataFrame, table: str, part: str) -> None:
    with engine.begin() as conn:                 # 同一事务
        conn.execute(f"DELETE FROM {table} WHERE dt = %s", (part,))
        df.to_sql(table, conn, if_exists="append", index=False)

删除与插入必须同事务。分开执行时若删除成功而插入失败，会得到一个"干净地空着"的分区，比重复数据更难发现。

7. 抽样与全量两种模式

def maybe_sample(df: pd.DataFrame, mode: str, seed: int = 42) -> pd.DataFrame:
    if mode == "full":
        return df
    return df.sample(frac=0.01, random_state=seed)

模式	用途	关键要求
`sample`	调试、验证口径	固定 `random_state`
`full`	出数交付	行数与总量校验

抽样模式必须有明确标识：文件名带 _sample 后缀，日志打一行 mode=sample frac=0.01。否则抽样结果被当真实数据发出去，是这类脚本最典型的翻车方式。

8. 出错时的可定位性

数据脚本的错误大多是"某一行数据不对"，不是"代码有 bug"。异常信息要带文件与行号。

import csv

def iter_rows(path: Path):
    with path.open(encoding="utf-8", newline="") as f:
        for lineno, row in enumerate(csv.DictReader(f), start=2):  # 2 = 跳过表头
            yield row   # 调用方按 lineno 拼错误信息

更实用的是分级：少量坏行跳过并计入 WARNING，坏行比例超过阈值（如 1%）整体失败。既不因一行脏数据让整批白跑，也不悄悄把脏数据算下去。

9. 结果校验：交付前的四道检查

检查	具体做法	失败含义
行数	与上一版对比，波动超过阈值告警	数据源或过滤条件变了
空值率	关键列 `isna().mean()`	维度缺失
总量	`gmv.sum()` 与上一版比值	口径或重复计算
唯一键	`df["order_id"].duplicated().sum() == 0`	清洗漏了去重

def validate(cur: pd.DataFrame, prev: pd.DataFrame, tol: float = 0.2) -> None:
    assert cur["order_id"].is_unique, "主键重复"
    ratio = cur["gmv"].sum() / max(prev["gmv"].sum(), 1e-9)
    if not (1 - tol) < ratio < (1 + tol):
        raise ValueError(f"总量偏离: 本期/上期={ratio:.3f}")

阈值宁可先宽后收：首次上线用 ±50%，看清实际波动再收紧，比一开始卡死 ±10% 然后天天误报好。

10. 常见坑对照

现象	原因	处理
换台机器就报 `relative import`	硬编码 `sys.path.insert`	用 `-m scripts.report` 从包根运行
手工改代码切换环境	连接串写在代码里	外置到配置文件 + 环境变量
重跑后数据翻倍	追加写 / 无删除	事务内先删后插
结果无法复现	抽样没固定种子、依赖当前时间	固定 `random_state`，时间从参数进
两个脚本口径不一致	脚本间复制粘贴	抽成共享模块只留一份
出错不知道是哪一行	异常被 `except: pass` 吞掉	记行号 + 计数 + 阈值

11. 边界：什么时候不该写成脚本

场景	更适合的形态
结论要给人看、要带图	Notebook 或报表工具
定时跑、要失败重试	脚本 + 调度器（见 S10-X07）
多步依赖、回填历史	工作流引擎（Airflow 等）
口径需要业务方随时改	配置驱动 + 版本化口径文件

一句话：脚本是"确定输入 → 确定输出"的载体。这件事不再确定，问题就不在写法上。

12. 自测

判断一个脚本该不该工程化的第一标准是什么？
为什么 --dry-run 在数据脚本里几乎是必需项？
fetch/clean/compute/export 里哪个最该写成纯函数，为什么？
幂等里"先删后插"为什么必须放进同一个事务？
少量坏行与大量坏行，处理策略为什么应该不同？
结果校验里为什么先设宽阈值再收紧？