一个凌晨跑的 Python 脚本，半年没出事。某天上游延迟两小时，脚本照样启动，算出半天的数据，把结果表写成一锅粥——而没人知道它跑过、跑了多久、写坏了哪一段。脚本和管道的差别不在代码量，在四个性质：能不能重跑、产物在哪、依赖是谁、出事怎么知道。

---

1. 脚本与管道的分界

性质	脚本	管道
幂等	跑两次多半出事（重复追加）	同一分区跑 N 次结果一致
可重跑	要手工改日期、清表	指定区间就能重放
产物	打印或写死一张表	按分区/时间片落在可定位的位置
依赖	靠注释和记忆	显式声明，失败可定位到上游
观测	只有 stdout	每次运行都有结构化记录
失败语义	挂了就挂了	部分成功可解释、可续跑

一句话判据："三点钟补跑昨天的数据"能不能只敲一条命令，且不污染已经正确的部分。

2. 四个组成与最小实现

组成	职责	最小实现
任务	可独立重放的逻辑	入参 ds，出参行数/产物路径
依赖	谁先谁后	DAG 或 depends_on
调度	什么时候跑、跑几个	cron / APScheduler / 编排器
观测	跑成什么样	一行 run_log

# pipeline.py —— 能重跑的最小管道
import datetime as dt, json, pathlib

BASE = pathlib.Path("data")

def run(ds: str) -> dict:
    out = BASE / "events" / f"ds={ds}"        # 分区路径：天然幂等
    if out.exists():
        return {"ds": ds, "rows": -1, "skipped": True}
    rows = extract(ds)                        # 上游拉数
    out.mkdir(parents=True)
    (out / "part-0.jsonl").write_text("\n".join(json.dumps(r) for r in rows))
    return {"ds": ds, "rows": len(rows), "skipped": False}

这段代码里有三个决定：入参只接日期、产物按分区落盘（覆盖写而非追加）、已存在就跳过。

3. 调度方案对照

方案	表达力	回填	观测	适合
cron	只有时间，无依赖	手工改日期	靠日志与退出码	单机、少量、无依赖
APScheduler	进程内、可动态增删	不支持	回调里自己记	常驻服务里的周期任务
Airflow	DAG + 时间片语义完整	backfill 原生	UI、日志、重试都有	多依赖、需要回填与追数
Dagster	资产（asset）为中心	按分区重物化	资产图与元数据	数据质量与血缘优先
Prefect	Python 原生、装饰器式	支持	有 UI，轻量部署	中小团队、想少写配置

判据：任务少且无依赖 → cron 够；要补跑过去 30 天 → 必须有回填语义；要知道"这张表由谁产出" → 资产模型。

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

sched = BlockingScheduler(timezone="Asia/Shanghai")
sched.add_job(run, CronTrigger(hour=2, minute=0),
              args=["2024-05-01"],
              misfire_grace_time=3600,   # 迟到 1 小时内仍执行
              coalesce=True,             # 堆积的多次触发合并成一次
              max_instances=1)           # 同一任务不并发
sched.start()

misfire_grace_time、coalesce、max_instances 是 cron 给不了的三件事：cron 错过就是错过，堆积就是堆积，重叠就是重叠。

4. DAG 的核心概念

概念	含义	踩错的后果
任务	最小调度单元	过细 → 调度开销超过计算
依赖	上下游关系	隐式依赖随并发度崩掉
时间片（ds）	这次运行代表哪段数据	用 now() → 无法回填
回填（backfill）	按区间重放历史	不支持 → 追数只能手工
并发限制	max_active_runs、池容量	不限 → 打满上游

# 追 30 天，Airflow 2.x 语义，按版本核对子命令
airflow dags backfill my_dag -s 2024-04-01 -e 2024-05-01 --reset-dependents

回填的前提是任务不读当前时间。任何 datetime.now() 出现在任务体里，等于给这个 DAG 判了不可回填。

5. 幂等与增量

做法	机制	坑
水位线	记录处理到的最大键/时间	时钟漂移与迟到数据会漏
分区覆盖写	每次重写整个 ds=... 分区	分区键选错（用写入时间而非事件时间）就永远对不上
哈希对账	比对源与目标的 count/sum	只对行数会漏值被改写

-- 水位线推进：带 overlap，避免边界数据被跳过
INSERT INTO etl.watermark (job, last_key, updated_at)
SELECT 'orders', MAX(id), now() FROM src.orders WHERE id > %(last_key)s
ON CONFLICT (job) DO UPDATE SET last_key = EXCLUDED.last_key;

-- 幂等写入：同主键重跑覆盖而不是追加
INSERT INTO dw.orders (id, amount, ds)
SELECT id, amount, %(ds)s FROM stg.orders WHERE ds = %(ds)s
ON CONFLICT (id) DO UPDATE SET amount = EXCLUDED.amount, ds = EXCLUDED.ds;

水位线要留重叠（回退 10 分钟或 1000 个 id），迟到数据才不会漏。重叠加幂等写，才不既漏又重。

6. 失败处理

失败类型	策略	注意
网络抖动、429、超时	指数退避 + 抖动，有限次数	次数写死，别写"无限"
格式错误、约束冲突	不重试，失败并隔离该批	重试只会再撞一次
部分成功	断点续跑：记录已完成分片	无断点就整段重来
上游未就绪	等待 + 超时	立刻失败会放大成"下游全红"

import random, time

def with_retry(fn, tries=3, base=2.0):
    for i in range(tries):
        try:
            return fn()
        except TransientError:
            if i == tries - 1:
                raise
            time.sleep(base ** i + random.uniform(0, 0.3))   # 抖动防同步重试

脏数据不丢也不进主表：写到 quarantine/ds=.../reason=...，附原始行与失败原因，次日人工看。隔离区是可运维管道的标志——把"数据错了"从事故变成待办。

7. 数据质量检查点

检查	表达式示意	放哪
行数基线	在近 7 日均值 ±30% 内	写主表前
空值率	null_ratio < 0.01	同上
主键唯一	count(*) = count(distinct id)	staging 层
跨表对账	订单金额合计 = 支付金额合计	日终，跨 DAG
枚举合法	状态列取值在允许集合内	入 staging 时

CHECKS = [
    ("row_count", lambda df, base: 0.7 * base <= len(df) <= 1.3 * base),
    ("null_ratio", lambda df, base: df["user_id"].isna().mean() < 0.01),
    ("pk_unique", lambda df, base: df["id"].is_unique),
]
def gate(df, base_rows):
    bad = [n for n, f in CHECKS if not f(df, base_rows)]
    if bad:
        raise QualityError(f"checks failed: {bad}")   # 阻断下游，不写主表

放在写主表之前，失败就不产物，比"先写再告警"干净，下游也不必判断这份数据能不能用。

8. 观测：每次运行记什么

字段	用途
run_id/dag_id/task_id	定位一次执行
ds（逻辑日期）	与分区对齐，追数时可查
started_at/duration	趋势与 SLA
status	success / failed / skipped / quarantined
rows_in/rows_out	突变是第一信号
retry_count/error_class	区分系统故障与偶发

INSERT INTO ops.run_log (run_id, dag_id, ds, started_at, finished_at, status, rows_out)
VALUES (%(run_id)s, %(dag)s, %(ds)s, %(t0)s, now(), %(st)s, %(n)s);

-- SLA：超过 40 分钟未完成即告警（供外部巡检查询）
SELECT dag_id, ds FROM ops.run_log
WHERE status = 'running' AND started_at < now() - interval '40 minutes';

SLA 按"数据什么时候可用"定，不是按"跑了多久"定；混用会把上游延迟算成自己的问题。

9. 本地与生产的一致性

项	不锁定的后果	做法
运行环境	本地 3.11、线上 3.9 直接报错	镜像固定版本，本地跑同一镜像
依赖	小版本升级改变默认行为	lock 文件 + 哈希校验
配置	连接串散在代码里	配置外置，镜像不含密钥
时区	本地 +08、容器 UTC 差一天	全链路显式 TZ

FROM python:3.11-slim
ENV TZ=Asia/Shanghai PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirements.lock .
RUN pip install --no-cache-dir --require-hashes -r requirements.lock
COPY . .
ENTRYPOINT ["python", "-m", "pipeline.cli"]

时区那条最容易被忽略：容器默认 UTC，本地开发 +08，同一个 date 会差一天，而报错往往出现在第二天的分区上。

10. 成本与调度粒度

粒度	成本特征	适合
每 5 分钟	启动开销占比高，连接频繁	低延迟、数据量小
每小时	折中，最常用	大多数业务指标
每天	峰值高，重跑代价大	对账、报表、重算

依据是单次固定开销 ÷ 单次计算时间：超过 10% 就该调粗粒度或合并批次。反之，延迟要求直接定死粒度，成本只能在约束内优化。

11. 排错

现象	原因	处理
重跑后数据翻倍	追加重跑，无幂等	改分区覆盖或 ON CONFLICT，清污染分区
分区混进两天数据	用写入时间当分区键	改事件时间 + 迟到窗口
常有"修好了"的记录	重试被当补偿	区分瞬时/永久错误，永久进隔离区
补跑历史全失败	任务写死当天日期	逻辑日期从入参取，禁 now()
失败后整段重来	无断点续跑	记分片进度，按分片跳过
上游晚到，下游全红	缺"等待上游就绪"的语义	加 wait_for/传感器，而不是立刻失败

12. 边界

场景	该用什么	理由
一次性的数据迁移	幂等脚本 + 手工确认	上编排器是过度工程
每分钟一条的定时任务	系统 cron / systemd timer	编排器的调度精度与开销都不划算
有跨天依赖与追数需求	Airflow 一类编排器	回填语义是核心价值
流式、秒级延迟	消息队列 / Flink 一类	批调度模型本身不适合
只关心"这张表新不新"	资产模型 + 新鲜度检查	比任务成功与否更贴近使用方

13. 自测

1. 用四条性质判断：一个脚本算不算管道？
2. cron 与 APScheduler 在"错过的触发"这件事上差在哪三个参数？
3. 为什么任务体里出现 datetime.now() 等于放弃回填？
4. 水位线为什么要留重叠窗口？重叠之后靠什么保证不重复？
5. 数据质量检查放在写主表之前还是之后，各有什么代价？
6. 容器里 TZ 没设，最可能在哪一天的分区上出问题？