5.0 KiB
5.0 KiB
id, title, type, domain, stack, tags, level, prerequisites, est_hours, related, week, order
| id | title | type | domain | stack | tags | level | prerequisites | est_hours | related | week | order | ||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| backend.data-pipeline.reliability | 数据管道基础：批处理与流处理、增量同步、限流重试、幂等、数据质量和告警 | lesson | backend |
|
|
intermediate |
|
3 |
|
|
1 |
数据管道基础：批处理与流处理、增量同步、限流重试、幂等、数据质量和告警
BOSS直聘 2026-10-23 的调研中，部分 AI 应用岗位要求“数据处理 / 数据管道”能力：知识库文档的采集清洗、第三方平台数据对接、报表数据准备。这里讲后端工程师需要掌握的基础，不涉及大数据平台。
解释说明
批处理和流处理
| 批处理 | 流处理 | |
|---|---|---|
| 触发 | 定时（每小时 / 每天） | 事件到达即处理 |
| 延迟 | 分钟到小时 | 秒级 |
| 实现 | 定时任务 + SQL / Python | 消息队列（Kafka / RabbitMQ）+ 消费者，或 CDC |
| 适合 | 报表、对账、全量重建 | 搜索同步、实时告警、实时推荐 |
很多系统两者结合：流处理保证实时，批处理定期对账兜底。
增量同步的关键
- 水位线：记录上次同步到哪里（时间戳 + 主键打破平局），失败后从水位线继续。水位线只在本批数据成功落库后推进。
- 限流：第三方接口有调用频率限制，用令牌桶控制；收到 429 时遵守
Retry-After。 - 重试：区分可重试错误（超时、5xx、429）和不可重试错误（401、参数错误）；指数退避 + 抖动（避免大量客户端同时重试），有上限。
- 幂等：重复运行结果一致——按主键 upsert，并且比较版本号 / 更新时间，旧数据不能覆盖新数据。
- 坏数据：单条坏数据进入死信表，不阻塞整批；坏数据比例超过阈值说明上游出问题了，应停止并告警。
- 数据质量检查：行数波动、空值率、取值范围、唯一性、与源头的对账。
- 可观测：每次运行记录拉取数、写入数、死信数、耗时；失败和质量异常告警。
常用工具（了解）
定时调度：cron、Celery Beat、APScheduler；编排：Airflow、Dagster、Prefect；CDC：Debezium；数据质量：Great Expectations 一类工具。小团队先用“定时任务 + 本课的这些原则”就够了。
案例
每 10 分钟从电商平台开放接口同步订单到自己的数据库，供 AI 报表 Agent 查询：接口每秒限 50 次、偶尔超时；有时返回状态值异常的数据；同步中断后要能续跑；不能产生重复订单。
代码示例
examples/incremental-sync（只用标准库，python pipeline.py 运行，test_pipeline.py 6 个测试通过）：
第一次：拉取 251，写入 250，死信 1，接口调用 5 次（含 2 次失败重试）
第二次（没有新数据）：拉取 0
orders 表 250 行，水位线 ('2026-10-23T11:00:00', 'bad1')
幂等且防止旧版本覆盖：
INSERT INTO orders (id, amount_cents, status, updated_at) VALUES (:id, :amount_cents, :status, :updated_at)
ON CONFLICT(id) DO UPDATE SET amount_cents = excluded.amount_cents, status = excluded.status, updated_at = excluded.updated_at
WHERE excluded.updated_at > orders.updated_at
重试（指数退避 + 全抖动，遵守 Retry-After）：
def with_retry(fn, *, attempts=5, base=0.2, cap=5.0, sleep=time.sleep, rng=None):
rng = rng or random.Random()
for i in range(attempts):
try:
return fn()
except RetryableError as e:
if i == attempts - 1:
raise
delay = e.retry_after if e.retry_after is not None else rng.uniform(0, min(cap, base * 2 ** i))
sleep(delay)
质量检查不通过时整批回滚、水位线不前进并告警（见 test_quality_gate_rolls_back_and_alerts）。
易错点
- 先推进水位线再写数据：中途失败就丢数据。
- 只用时间戳做水位线：同一秒内有多条数据时会漏或重复；要加主键打破平局。
- 迟到数据：来源系统的事务晚提交，
updated_at比水位线还早，被永远跳过。办法：每次往回多拉一个重叠窗口（例如 5 分钟），依赖幂等写入去重；再加定期对账。 - 对所有错误都重试，包括 401，结果被对方封禁。
- 没有抖动的重试，故障恢复瞬间所有任务一起打过去。
- 坏数据直接让整个任务失败，或者相反——静默丢弃没有任何记录。
面试常问
- 怎么设计一个可以断点续传的增量同步任务？
- 怎么保证同步任务幂等？
- 第三方接口限流和不稳定怎么处理？
- 怎么发现数据质量问题？