要做什么：每天凌晨自动跑一遍，跑完就退
数据管道是最常见也最容易被低估的实战题：每天把若干数据源的榜单抓回来，清洗成统一结构，落进 SQLite，供站内页面和分析脚本使用。它是一个跑批脚本，不是常驻服务——每天跑一次，跑完就退。”每日”这个约束会让每个设计决策都变重：数据源不可靠（超时、改版、限速），失败要隔夜才能被发现，而重跑是常态而不是事故，所以第一职责不是跑得快，是重跑不出错。
| 指标 | 目标值 | 怎么测 |
|---|---|---|
| 重跑幂等 | 0 重复行（同一天同一源同一主键只有一行） | 同一批数据连跑两次，第二次行数与第一次相同 |
| 失败隔离 | 单个源挂掉不影响其他源正常入库 | 故意把一个源的 URL 改错，看其余源是否照常 |
| 失败可见 | 跑批异常 5 分钟内收到 webhook 告警 | 断网或改错 URL 触发一次真实失败 |
| 可追溯 | 每行数据带来源、批次号（run_id） | 抽查任意一行，能回答”谁、何时、从哪抓的” |
| 跑批耗时 | 全量控制在分钟级（估算口径，未实测） | 日志首尾时间戳之差 |
目录结构：四层各管一段
pipeline/
├── config.py # 源列表、阈值、webhook 地址，只此一处
├── fetch.py # 抓取层：URL → 原始文本，只管取回
├── clean.py # 清洗层：原始文本 → 结构化记录（脏数据抛异常）
├── store.py # 落库层：建表与 UPSERT，唯一约束在这里
├── alert.py # 告警层：按批次汇总 + 静默窗口，防风暴
└── run_daily.py # 入口：串联四层，APScheduler 定时分层理由：抓取和清洗分开，是因为清洗逻辑的改动频率远高于抓取——数据源改版时只动 clean.py；清洗层是纯函数（不碰网络不碰库，输入输出都是确定值），单测只测它，跑得飞快。抓取和落库分开，是为了让”取回来的原始数据”与”入库的最终状态”之间有明确的界线，出问题时能分辨是源变了还是处理错了。
难点一：抓取——超时、重试、礼貌三件套
import httpx
def fetch_one(client: httpx.Client, source: dict) -> str:
1. 显式超时：哪怕库给了默认值也写出来，这是给读代码的人看的文档
resp = client.get(source["url"], timeout=10.0)
resp.raise_for_status()
return resp.text
transport = httpx.HTTPTransport(retries=3) # 连接建立失败自动重试 3 次
with httpx.Client(transport=transport) as client:
for source in SOURCES:
raw = fetch_one(client, source)诚实边界：HTTPTransport(retries=...) 只重试”连接建立失败”；请求发出去之后的读超时、5xx 不会自动重试——这些要在业务层自己判断，想对 5xx 也重试就写显式循环或引入 tenacity 这类重试库。
反模式一：不设超时。requests 的默认超时是永不超时，一个挂起的数据源能把整条管道卡到天亮，这是数据管道最常见的死法。httpx 有默认超时（5 秒），但仍然显式写出来。限速同理：对同一个域名的连续请求之间加 sleep 加随机抖动，别把数据源当压测目标；几十个源以内用串行加限速，比并发池更简单也更礼貌。
难点二：清洗——脏数据宁可丢弃，不可静默入库
from dataclasses import dataclass
@dataclass(frozen=True)
class Item:
source: str
day: str # YYYY-MM-DD，主键的一部分
rank: int
title: str
url: str
def parse_item(source: str, day: str, raw: dict) -> Item:
title = (raw.get("title") or "").strip()
if not title:
raise ValueError("empty title") # 抛异常由调用方计数，绝不静默放行
rank = int(raw["rank"]) # 解析失败同样抛异常
url = (raw.get("url") or "").strip()
return Item(source=source, day=day, rank=rank, title=title, url=url)清洗层的纪律：ValueError 抛出来、计数、记进当次跑批的汇总，比一条字段残缺的数据混进库里好一百倍——脏数据一旦入库，污染的是下游所有查询和分析。函数与类的基础写法见《Python 函数与面向对象》。
难点三：落库——去重靠唯一约束，不靠”先查再插”
conn.execute("""
CREATE TABLE IF NOT EXISTS items (
source TEXT NOT NULL,
day TEXT NOT NULL,
rank INTEGER NOT NULL,
title TEXT NOT NULL,
url TEXT NOT NULL,
run_id TEXT NOT NULL,
PRIMARY KEY (source, day, rank) -- 每源每天每名次一行
)
""")反模式二：清洗时”先查有没有再插”。查与插之间存在窗口期，重跑或并发时判重必失效。正确做法是把主键建在表上，用 upsert 让 INSERT 的冲突语义替你判断：
conn.execute("""
INSERT INTO items (source, day, rank, title, url, run_id)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(source, day, rank) DO UPDATE SET
title = excluded.title,
url = excluded.url,
run_id = excluded.run_id
""", (item.source, item.day, item.rank, item.title, item.url, run_id))upsert 还是 ignore 取决于业务：榜单名次会变，这里选”后到覆盖”，保证库内最终状态与数据源一致；如果是”只留首见”的日志型数据，换成 DO NOTHING。这一条主键同时是幂等的全部实现——重跑 N 次和跑一次结果相同。
难点四：调度——APScheduler 的三个参数对应三种真实事故
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
@sched.scheduled_job("cron", hour=2, minute=30,
misfire_grace_time=3600, # 错过触发点后 1 小时内仍补跑
coalesce=True, # 补跑只补一次，不堆积
max_instances=1) # 上一轮没跑完时不并发起跑
def daily_pipeline():
run_all()
sched.start()三个参数各对应一种事故：机器凌晨重启错过 2:30（misfire_grace_time）；积压的补跑被排了三次、同一批数据跑三遍（coalesce）；上一轮还没跑完下一轮又起跑，两个实例同时写库（max_instances）。默认值在这三处都不符合跑批的预期，必须显式设。
另一个诚实的提醒：如果部署环境已有 crontab 或 systemd timer，APScheduler 不是必需品——30 2 * * * python -m pipeline.run_daily 同样成立，而且更不容易坏。APScheduler 的价值在进程内调度与动态增删任务；纯每日跑批，cron 是更笨也更可靠的选项。
难点五：告警——汇总、截断、静默，别把自己吵麻木
import time
import httpx
_last_sent = 0.0 # 简化示例用模块级变量；多进程部署时换成共享存储
def notify(webhook: str, lines: list) -> None:
global _last_sent
if not lines:
return
now = time.time()
if now - _last_sent < 1800: # 30 分钟静默窗口：风暴时只发第一条
return
body = {"text": "pipeline 异常\n" + "\n".join(lines[:20])} # 截断防刷屏
try:
httpx.post(webhook, json=body, timeout=5.0)
_last_sent = now
except httpx.HTTPError:
pass # 告警失败不能反过来弄死跑批，但要落日志反模式三：每个失败项单独推一条消息。一个源改版能产生几百条解析失败，群通知几分钟刷到上限，之后的真故障反而没人看得见。按 run 汇总、加静默窗口，是告警系统的自我保护。但要分清另一个方向：except: pass 吞掉一切不叫静默窗口——窗口抑制的是”重复通知”，不是”通知本身”，notify 失败要落日志，跑批结束的汇总里必须写明哪些源失败。
入口串联：失败隔离的结构
def run_all() -> None:
run_id = f"{date.today().isoformat()}-{uuid4().hex[:6]}"
failures, ok = [], 0
transport = httpx.HTTPTransport(retries=3)
with httpx.Client(transport=transport) as client:
for source in SOURCES:
try:
raw = fetch_one(client, source)
items = [parse_item(source["name"], run_day(), r) for r in parse_raw(raw)]
store.upsert(items, run_id) # 每源独立事务
ok += 1
except Exception as e: # 单源失败只记账，不打断全局
failures.append(f'{source["name"]}: {e!r}')
alert.notify(WEBHOOK, failures) # 结束后统一汇报
print(f"run_id={run_id} ok={ok} failed={len(failures)}")关键是每个源独立 try/except、独立入库：一个源炸了，其他源的本次数据照常落库，失败名单进告警——”部分成功”在数据管道里是正常态，全部回滚反而是灾难。
验证
python -m pipeline.run_daily # 手动跑一次，看汇总日志
sqlite3 data/pipeline.db "SELECT COUNT(*) FROM items"
sqlite3 data/pipeline.db "SELECT source, COUNT(*) FROM items
WHERE day = date('now') GROUP BY source"
python -m pytest tests/test_clean.py # 清洗层纯函数单测| 检查项 | 操作 | 期望 |
|---|---|---|
| 重跑幂等 | 连跑两次 | 第二次行数不变（upsert 生效） |
| 失败隔离 | config 里改坏一个 URL | 该源记 0 行，其余源正常，webhook 收到汇总 |
| 补跑 | 把系统时间拨过触发点再启动 | misfire_grace_time 内补跑一次且只跑一次 |
| 可追溯 | 抽查任意一行 | source/day/run_id 三元组完整 |
数量级估算与升级时机
本站未实测，给估算口径：单线程串行、每源抓取加清洗约 2 秒、20 个源，一次跑批约 1 分钟，瓶颈在网络等待而非计算；SQLite 对”每日几千行”的写入量级绰绰有余。什么时候该升级：数据源上百、行数到百万级、或查询要跨机器——那时才考虑换 PostgreSQL、加并发抓取、把清洗搬到消息队列后面。在那之前，这套结构的简单本身就是可靠性。
总结
- 管道的第一职责是重跑不出错：唯一约束加 upsert，让幂等成为表结构的一部分而不是代码里的判断
- 失败隔离与失败可见同样重要：单源挂掉不许拖垮全局，但必须在天亮前有人知道
- 不设超时是抓取最常见的死法；重试要分清”连接失败”与”请求失败”，别指望传输层替你包办
- misfire_grace_time、coalesce、max_instances 三个参数对应三种真实事故，默认值不可信
- 告警要做自我保护：按批次汇总、截断、静默窗口——否则告警本身会先于系统失明
管道跑出结果之后，想给同事一个能看的页面，Web 框架从轻到重可以从《Flask 入门》起步，要类型校验与自动文档则看《FastAPI 框架入门》；如果团队已押注”全家桶”，《Django 入门》里的管理后台能省掉一半展示层工作。
版本核实（以官方文档为准）：写作时点（2025 年 5 月）Python 当前稳定版为 3.13，本文代码兼容 3.10+；httpx 当前稳定版为 0.28.1（2024 年 11 月发布）；APScheduler 3.x 为稳定线（3.11.0，2024 年 11 月发布），4.x 尚未正式发布，用法以官方文档为准；SQLite 经标准库 sqlite3 使用，随 Python 自带。安装命令均未钉版本号。