最糟糕的抓取失败,看起来根本不像失败。任务照常运行,请求成功,数据按计划进入数据仓库。然后财务部门有人问,为什么平均价格一夜之间涨了三倍,或者为什么一半的商品目录突然显示缺货,调查最后追溯到三周前的一次网站改版,而当时没有任何人注意到。
这就是模式漂移(schema drift):数据仍在流动,但它的形状或含义已经悄然改变。这是网页数据的常规失败模式,因为数据源在毫无通知的情况下变化,而提取器仍在照常产出结果。本指南介绍了能够捕捉这种情况的两个层次:拒绝错误值的记录级契约,以及能察觉整批数据整体已不再像原来那样的批次级画像。文中还通过一个小型测试展示了为什么你两者都需要。
关键要点
- 漂移通常是无声的。一个变化了的页面很少会让提取器崩溃;它只会让提取器返回一些看似合理但实际错误的结果。
- 记录级契约按规则检查每条记录:必填字段、类型、取值范围和格式。它们能捕捉明显的破坏。
- 批次级画像将每次运行与基线进行比较:填充率、类型构成和中位数。它们能捕捉契约无法察觉的破坏。
- 在我们对三种真实漂移情形的测试中,契约只捕捉到了一种。另外两种通过了所有记录级检查,只有批次画像才捕捉到了它们。
- 在数据发出之前对漂移发出警报,隔离该批次,并记录是哪个站点和哪个提取器版本产生了它。
漂移实际是如何发生的
| 原因 | 数据会发生什么 |
|---|---|
| 改版改变了标记结构 | 选择器匹配到了不同的元素,或者什么都匹配不到 |
| 价格格式变化 | ”9.99” 变成 “€9.99”、“9,99” 或以最小货币单位表示的 999 |
| 某个字段被移到 JavaScript 后面 | 静态 HTML 中不再包含它,因此返回为空 |
| 本地化或地理变体 | 部分页面出现不同的货币、语言或单位 |
| A/B 测试 | 一部分页面使用新布局,导致一部分记录出错 |
| 拦截或验证页面 | 页面正常加载,提取器正常运行,但返回空值或垃圾数据 |
这些情况的共同点是,它们都不会抛出异常。最后一种情况,页面成功加载但并非你想要的页面,已在静默失败率一文中讨论过。其余的情况都是提取器忠实地处理了一个已在底层发生变化的页面。
第一层:记录级契约
数据契约规定了一条有效记录应有的样子:哪些字段是必需的、它们的类型、允许的取值范围和格式。每条记录在被接受之前都要经过检查,违规记录会被计数并隔离,而不是悄悄地存入库中。
import re
import statistics
from collections import Counter
CONTRACT = {
"name": {"type": str, "required": True},
"price": {"type": float, "required": True, "min": 0.01, "max": 100_000},
"currency": {"type": str, "required": True, "pattern": r"^[A-Z]{3}$"},
"in_stock": {"type": bool, "required": False},
}
def violations(record, contract=CONTRACT):
"""Record-level checks: presence, type, range, format."""
problems = []
for field, rule in contract.items():
value = record.get(field)
if value in (None, ""):
if rule.get("required"):
problems.append(f"{field}: missing")
continue
if rule["type"] is float and isinstance(value, int) and not isinstance(value, bool):
value = float(value)
if not isinstance(value, rule["type"]):
problems.append(f"{field}: expected {rule['type'].__name__}, got {type(value).__name__}")
continue
if "min" in rule and value < rule["min"] or "max" in rule and value > rule["max"]:
problems.append(f"{field}: {value} out of range")
if "pattern" in rule and not re.match(rule["pattern"], value):
problems.append(f"{field}: {value!r} bad format")
return problems像 {"name": "x", "price": "€9.99", "currency": "eur"} 这样的记录会两处失败:price 是字符串,而 currency 不是三字母大写代码。契约成本低、明确、易于推理。它的局限在于每条记录都是被单独判断的,而大量的漂移产生的记录在单独看时是有效的。
第二层:批次级画像
画像对整批数据进行总结:对每个字段,记录它被填充的频率、出现了哪些类型,以及对于数字字段,记录其中位数。将每次运行的画像与最近健康运行的基线进行比较,可以显示数据整体形状何时发生了变化,即便每条记录都通过了自身的契约检查。
def profile(records, fields=CONTRACT):
"""Batch-level shape: how often each field is filled, its types, and numeric medians."""
n = max(1, len(records))
out = {}
for field in fields:
values = [r.get(field) for r in records]
present = [v for v in values if v not in (None, "")]
numbers = [float(v) for v in present if isinstance(v, (int, float)) and not isinstance(v, bool)]
out[field] = {
"fill_rate": len(present) / n,
"types": Counter(type(v).__name__ for v in present),
"median": statistics.median(numbers) if numbers else None,
"distinct": len(set(map(str, present))),
}
return out
def drift(baseline, current, fill_drop=0.1, median_ratio=3.0):
"""Compare two batch profiles and describe what changed shape."""
alerts = []
for field, base in baseline.items():
cur = current[field]
if base["fill_rate"] - cur["fill_rate"] > fill_drop:
alerts.append(f"{field}: filled {base['fill_rate']:.0%} -> {cur['fill_rate']:.0%}")
if set(cur["types"]) - set(base["types"]):
alerts.append(f"{field}: new types {sorted(set(cur['types']) - set(base['types']))}")
if base["median"] and cur["median"]:
ratio = cur["median"] / base["median"]
if ratio > median_ratio or ratio < 1 / median_ratio:
alerts.append(f"{field}: median {base['median']:g} -> {cur['median']:g}")
return alerts这些阈值只是起点。填充率下降 10 个百分点,或者中位数发生三倍变化,很少属于业务正常波动;在积累了几周的历史数据之后,应该针对每个字段分别调整这两个阈值。
为什么两者都需要:一个小型测试
我们生成了一个包含 1,000 条产品记录的健康基线,然后模拟了三种真实的破坏情形,并对每一种运行了这两个层次的检查。
| 漂移 | 发生了什么 | 记录级契约 | 批次画像 |
|---|---|---|---|
| 格式化价格 | 改版使 30% 的页面把货币符号放进了价格里 | 捕捉到:300 条记录被拒绝 | 捕捉到:price 中出现新类型 str |
| 最小货币单位 | 价格开始以 6305 而不是 63.05 的形式出现 | 未捕捉到:所有记录都通过了 | 捕捉到:中位数从 63.05 变为 6305,并出现新类型 int |
| 库存选择器损坏 | 80% 的页面 in-stock 字段返回为空 | 未捕捉到:该字段是可选的 | 捕捉到:填充率从 100% 变为 20% |
契约只捕捉到了那个产生明显畸形值的漂移。另外两种产生的记录单独来看都是有效的,一个是在范围内的正数,一个是留空的可选字段,只有批次视图显示出发生了变化。最小货币单位的情形并非假设:我们上周检查的一个真实产品接口,对一双售价 100.00 美元的鞋子返回了 10000,详见停止解析 HTML一文中的描述。
漂移触发时该怎么做
- 隔离该批次。 在有人检查之前,不要发出带有漂移警报的批次数据。数据晚到总比数据错误要好。
- 确定影响范围。 按站点、页面类型和提取器版本对警报进行细分。漂移通常从某次变更后的某一个站点开始。
- 对比样本。 查看少数受影响的记录,并对照它们来源的页面。原因通常几分钟内就能看清楚。
- 修复并回填。 更新提取器,然后如果你保存了原始页面,就对受影响的时间段重新提取;如果没有保存,就重新采集。
- 有意识地更新基线。 当变化是合理的,比如某个站点确实更换了货币,那就有意地重置基线,绝不要自动重置。
从源头上降低漂移发生的可能性
某些数据源的漂移比其他数据源更少。为搜索引擎嵌入的结构化数据,其变化频率远低于页面布局,这也是为什么优先提取结构化数据能大幅降低契约触发的频率。持续监控页面本身也有帮助:通过大规模变更检测检测到的模板变化,是提取即将出错的早期预警,而某个站点上契约违规率不断上升,则是目标健康评分的一个重要输入。此外,每个任务只从一个市场持续采集,也能消除一整类由地理和货币变体引起的漂移。
总结
网页数据会发生漂移,因为网络在不打招呼的情况下发生变化。危险不在于提取器崩溃,而在于它们继续在已不再是当初构建目标的页面上运作,并逐条产出看似正常的数据。
对每条记录都进行契约检查,对每个批次都与其自身最近的历史进行比较。在我们的测试中,单靠契约在三种漂移中只捕捉到一种;批次画像则捕捉到了全部三种。两者结合,能把”仪表盘看起来有点不对劲,持续了三周”变成在问题刚出现那天早上就发出的警报。
来源与参考资料
- 由 Shifter 于 2026 年 9 月 29 日使用上述代码运行的测试,基于 1,000 条生成的产品记录和三种模拟漂移。