爬虫监控怎么做？用运行报告实现定时任务的可复现与异常告警

如果你手上有三五个定时爬虫每天都在服务器上安静地跑那你大概率经历过这种时刻某个早晨想看看昨天的数据有没有更新打开后台发现采集量是0日志却停在前天。于是你开始瞎猜——IP被封了目标网站改版了服务器半夜重启把定时任务弄丢了其实大部分爬虫不是跑不到而是跑完了你不知道它跑得好不好。你只知道进程退出码是0但抓下来的数据是否完整、字段有没有缺失、有没有被反爬系统盯上这些信息全都埋在日志深处没人理。今天想聊的这套东西就是给爬虫加一层体检机制每跑一次生成一份结构化的运行报告记录采集量、失败率、响应时间、数据质量、异常样本并把代码版本、依赖版本、配置快照一起钉进报告里。这就是标题里说的可复现——报告不是事后看一眼的进度条而是可以回放、可以对比、可以触发告警的运行证据。这套方案适合谁手上有定时任务或批量采集脚本、但不想专门搭建监控平台的同学或者你的爬虫已经在向团队、向下游交付数据需要向别人证明这批数据是完整的、可信的。接下来从核心思路到代码实现从部署到排坑我都按自己实际搭过的流程写你可以改一改直接拿去用。1. 费了三天才想明白的事爬虫监控其实是在监控确定性1.1 爬虫监控和Web服务监控不是一回事先说一个容易踩的思维误区。很多人一听监控系统第一反应是搬服务器监控那套进程活着吗、端口通不通、CPU高不高、5xx多不多。这套思路搬过来会水土不服因为爬虫和Web服务的运行模式完全不同。Web服务是被动响应只要进程活着、延迟正常基本就没什么大事。爬虫是主动出击它面对的是网络波动、目标网站改版、反爬策略调整、下游数据校验失败……这些意外几乎不可能用一个存活探针测出来。举个真实例子。我的一个爬虫当时一直返回200进程也没崩看起来一切正常。但目标网站悄悄改了新版把列表页改成懒加载我原来的解析逻辑抓到的全是空数组。如果只看存活状态这个bug可以挂一周没人发现。但如果有运行报告机制就会发现请求数正常、状态码正常而解析成功条数从几百掉到0当场就该拉响警报。1.2 可复现到底指什么很多人以为可复现就是代码在版本库里随时能跑。那离可复现还差得远。我理解的可复现是三个层面同时成立环境可复现报告里记录了git commit、Python版本、关键依赖版本、宿主平台三个月后回看也能定位到当时跑的是哪一版代码。行为可复现调度顺序、随机因子、时间基准都是确定性的。同样一份代码、同样一份配置在目标网站没变的前提下跑出来的统计口径一致。结果可复现报告本身就是稳定的结构化数据字段定义固定、枚举值固定、schema有版本号。这样历史报告才能被程序批量解析和横向对比。层面回答的问题对应报告字段环境可复现当时跑的哪版代码、什么环境commit_id, python_version, deps行为可复现上次和这次的执行口径一致吗seed, config_hash, timezone结果可复现不同时间点能横向比较吗schema_version, metrics字典1.3 这套系统的适用边界也别把问题想得太复杂。如果你只是偶尔写个一次性脚本抓点数据跑完就不管了那确实不需要这套东西——为一次性需求搭监控维护成本大于收益。但一旦爬虫满足下面任意一条就很值得补上运行报告定时任务每天或每周固定执行数据会被下游拿去分析、建模或展示需要向别人说明这批数据是好的你已经吃过一次数据静默断更的亏。我自己的判断标准很简单只要这个爬虫还会跑第二次就值得在第一天顺手把报告框架搭进去。后面所有加告警、加趋势对比的步骤都是在给这个框架添砖加瓦。2. 一次跑完留证运行报告的数据结构应该长什么样写第一行代码之前最好先把报告的数据结构定下来。这一步省不得因为报告就是你排查问题的案发现场字段少一个复盘时可能就要多翻半天日志。我的习惯是把报告拆成四块。2.1 运行元数据报告自己的身份证run_id是整个系统的天然主键。我的生成规则是任务名加UTC时间戳例如product_list_20250107_023015。只要报告目录里出现这个文件名任何同事都可以拿着它去查对应的日志和代码版本。字段含义示例schema_version报告结构版本号字段变更时11task_name任务名和调度任务一一对应product_listrun_id主键任务名UTC时间戳product_list_20250107_023015started_at / finished_at起止时间ISO8601带时区2025-01-07T02:30:15Zcommit_id代码git commit前12位3fa8c1b2d9e4python_versionPython版本3.11.4deps关键依赖版本列表[requests2.31.0]config_hash配置JSON的SHA256前12位7c2f9a11d3e8为什么要存这些因为过两天回去看报告发现数据不对但不知道是哪版代码跑的这件事太常见了。有了commit_id直接git checkout 3fa8c1b2d9e4就能还原代码现场有了config_hash就能确认参数没被偷偷改过。2.2 采集统计这次实际发生了什么这一块记录HTTP层面的运行情况相当于运动手表的配速和心率。requests_total / requests_success / requests_failed请求的总数、成功数、失败数requests_retried发生重试的次数。注意这个数字必须单独记。如果不记重试会把失败率搞得很假status_codes状态码分布字典例如{200: 482, 403: 7, 504: 3}response_time_avg / response_time_p95平均响应时间以及P95。P95比平均值更能反映尾延迟我通常是保留每次请求的耗时数组跑完再算分位值。不要小看状态码分布这一项。有一次我的爬虫被某个节点拦了7个请求如果不看分布光看失败率3%可能觉得还好但看到403突然冒出来就会立刻意识到该去检查代理池了。2.3 数据质量指标HTTP正常不代表数据正常这是我最看重、也是很多教程不讲的一块。爬虫的最终产物是解析出的结构化数据而不是请求成功多少次。所以必须把数据质量单独计量parsed_items解析成功的记录条数field_missing_count关键字段缺失的记录数也可以细化成每个字段的缺失数量duplicate_count抽样去重后发现的重要字段重复条数coverage_ratio计划抓取URL数除以实际成功解析数。这个指标建议在爬虫启动时就算出预期列表页数量结束时再算覆盖率。注意parsed_items 为 0和requests_total 为 0是两个不同量级的故障。前者说明目标网站可能改版了或者解析逻辑失效后者说明任务可能压根没被调度。在告警里要分开处理权重也不同。2.4 异常快照给未来复盘留线索错误信息不能在报告里大海捞针。我会把异常按类型聚合只保留前50条样本每条样本带上URL、异常类型、摘要信息anomalies: { TimeoutError: { count: 12, samples: [ { url: https://example.com/list?page3, message: Connection timed out after 10s, time: 2025-01-07T02:33:41Z } ] } }这样一份报告任何一个人拿到手不需要翻原始日志就能回答这次跑了多少、遇到什么问题、采到多少数据、当时的代码是什么。3. 落地到代码上下文采集、指标埋点和报告生成数据结构定了代码就很好写。我尽量把代码控制在能放进一个项目里直接跑的规模不依赖重型框架。3.1 RunContext跑一次就建一个账本import hashlib import importlib.metadata import json import platform import socket import sys from datetime import datetime, timezone def _get_commit_id(): try: import git repo git.Repo(search_parent_directoriesTrue) return repo.head.commit.hexsha[:12] except Exception: return unknown def _get_key_deps(): key_deps [] for name in (requests, scrapy, bs4, lxml, pandas): try: version importlib.metadata.version(name) key_deps.append(f{name}{version}) except Exception: key_deps.append(f{name}unknown) return key_deps class RunContext: def __init__(self, task_name, config): self.task_name task_name self.config config self.run_id f{task_name}_{datetime.now(timezone.utc).strftime(%Y%m%d_%H%M%S)} self.started_at datetime.now(timezone.utc).isoformat() self.finished_at None self.metrics { requests_total: 0, requests_success: 0, requests_failed: 0, requests_retried: 0, response_times: [], parsed_items: 0, field_missing_count: 0, duplicate_count: 0, } self.status_codes {} self.anomalies {} def _bump_status(self, code): self.status_codes[str(code)] self.status_codes.get(str(code), 0) 1 def _record_anomaly(self, key, url, message): entry self.anomalies.setdefault(key, {count: 0, samples: []}) entry[count] 1 if len(entry[samples]) 50: entry[samples].append({ url: url, message: message[:200], time: datetime.now(timezone.utc).isoformat(), }) def finish(self): self.finished_at datetime.now(timezone.utc).isoformat() def snapshot(self): times self.metrics[response_times] avg_rt round(sum(times) / len(times), 3) if times else 0 sorted_rt sorted(times) p95_rt round(sorted_rt[int(len(sorted_rt) * 0.95)], 3) if sorted_rt else 0 metrics {k: v for k, v in self.metrics.items() if k ! response_times} metrics.update({ response_time_avg: avg_rt, response_time_p95: p95_rt, }) return { schema_version: 1, run_id: self.run_id, task_name: self.task_name, started_at: self.started_at, finished_at: self.finished_at, env: { python: sys.version.split()[0], platform: platform.platform(), hostname: socket.gethostname(), commit_id: _get_commit_id(), config_hash: hashlib.sha256( json.dumps(self.config, sort_keysTrue, ensure_asciiFalse).encode() ).hexdigest()[:12], deps: _get_key_deps(), }, metrics: metrics, status_codes: self.status_codes, anomalies: self.anomalies, }这段代码解决三个问题统一起止时间、把埋点集中到一个对象、所有公有字段在快照时一次性落盘。注意response_times这个数组只在内部使用生成报告时换算成avg和p95避免JSON体积被时间序列撑大。3.2 用装饰器统一埋点很多人埋点是每写一个请求就手打一行计数器稍微漏掉一个分支统计就不准了。我的做法是用装饰器包住唯一请求入口import time from functools import wraps def track_request(ctx): def decorator(func): wraps(func) def wrapper(*args, **kwargs): url args[0] if args else kwargs.get(url, unknown) ctx.metrics[requests_total] 1 start time.perf_counter() try: resp func(*args, **kwargs) ctx.metrics[requests_success] 1 ctx._bump_status(resp.status_code) return resp except Exception as exc: ctx.metrics[requests_failed] 1 ctx._record_anomaly(type(exc).__name__, url, str(exc)) raise finally: elapsed time.perf_counter() - start ctx.metrics[response_times].append(round(elapsed, 4)) return wrapper return decorator这里有两个关键决策。第一把response_time收集成数组而不是只记总和这样才能在快照时算P95。第二重试逻辑要单独封装重试时只增加requests_retried不要混进失败计数器里否则重试风暴会把失败率污染得失真。3.3 报告落地文件目录当数据库用报告存储我用非常朴素的目录不引入数据库。JSON文件本身就可以当数据库用按任务名分目录、按时间排序遍历reports/ ├── product_list/ │ ├── product_list_20250107_023015.json │ ├── product_list_20250106_021500.json │ └── ... └── user_profile/ └── ...import json from pathlib import Path def save_report(ctx, report_dirreports): report ctx.snapshot() task_dir Path(report_dir) / ctx.task_name task_dir.mkdir(parentsTrue, exist_okTrue) out_path task_dir / f{ctx.run_id}.json out_path.write_text( json.dumps(report, ensure_asciiFalse, indent2), encodingutf-8 ) return out_path3.4 一个完整的运行闭环把所有零件串起来主程序长这样import requests def main(): config {base_url: https://example.com/list, max_pages: 20} ctx RunContext(product_list, config) track_request(ctx) def fetch(url, session): return session.get(url, timeout10) try: session requests.Session() for page in range(1, config[max_pages] 1): url f{config[base_url]}?page{page} resp fetch(url, session) items parse_items(resp.text) ctx.metrics[parsed_items] len(items) finally: ctx.finish() save_report(ctx)核心思路只有一个所有请求都走同一个入口统计才不会漏。你可以在track_request之上再加代理切换、限速、重试只要保证入口统一监控数据就能自动聚齐。4. 可复现性的三个螺丝版本、配置快照和重放能力数据结构里已经有了env字段但字段存在不等于真的可复现。要让可复现从口号变成能力下面三个螺丝必须拧紧。4.1 钉死代码版本与依赖我见过很多团队用镜像tag表示版本比如my-crawler:v3.2。听起来没问题但镜像tag可以被覆盖重打一次tag历史版本就丢了。相对靠谱的锚点有两个git commit id不可变、全局唯一。报告里记了commit_id任何时候都能git checkout还原镜像digest如果已经容器化python:3.11-slimsha256:...的摘要比tag可靠得多。依赖层面不需要把整个pip freeze全塞进报告那太占空间。挑requests、scrapy、bs4、lxml、pandas这类会实质影响行为的依赖记录版本即可已经足够定位是不是某个版本跑出来结果不一样。4.2 配置快照存哈希不存明文配置里常有API key、数据库连接串、Cookie之类的敏感信息直接写进报告是给自己埋雷。正确做法是把配置序列化后算SHA256只把哈希写进报告。哈希有两个作用一是确认两次运行用的配置是否一致二是在排查问题时如果哈希值变了就去版本库里看对应配置文件的变更记录。这里有个细节必须提醒config_hash的生成规则要严格统一必须用sort_keysTrue加ensure_asciiFalse保证同一个配置在任何机器上算出来的哈希完全一致。不要用Python内置的hash()内置哈希在进程间不保证一致。4.3 时间和随机性的对齐昨天00:00到今天00:00的数据究竟按哪个时区切割这个问题不提前统一报告对比就会出鬼。我的习惯是所有时间字段一律UTCISO8601带时区后缀如2025-01-07T02:30:1500:00。展示层需要转当地时间再转存储层不做任何时区换算。如果爬虫里用到随机数比如随机User-Agent、随机延时每次运行都会引入不确定性。可复现不要求随机数固定但必须在报告里记录随机种子seed这样才能保证同一个seed加同一版代码能得到同一份调度顺序。实现起来很简单import random seed 42 random.seed(seed) # 把seed写进report的env字段4.4 拿着报告能重放现场的用法可复现不是理论概念它是排查故障时的实际操作。我之前遇到过一次数据异常怀疑配置文件被临时改过但没人记得。按复盘流程我拿到当天报告里的config_hash先看配置文件的变更记录再用报告里的commit_id切到当时代码用同一个seed重跑一遍发现解析结果和报告完全一致。那一刻我才真正理解了什么叫可复现——它不是代码能跑而是你随时可以回到出事那天再看一次事故是怎么发生的。5. 单份报告只是开始用历史对比发现爬虫的慢性病告警适合抓急性问题比如解析量为0、失败率突然飙升。但爬虫还有一类更难防的问题——慢性恶化。响应时间一天比一天慢、失败率从2%悄悄爬到10%、每天采集到的条数逐渐缩水。这些问题单看一天的报告根本发现不了必须把历史报告拉出来对比。5.1 文件系统就是最省事的报告仓库reports/{task_name}/{run_id}.json这个布局的好处是文件名本身就是时间戳sorted()一下就天然有序不需要额外建数据库用pathlib几行代码就能遍历。等以后量大了再换数据库也不迟因为JSON文件的schema已经固定迁移成本很低。5.2 一个能直接抄的历史对比函数import json from pathlib import Path def load_reports(report_dir, task_name, limit30): task_dir Path(report_dir) / task_name if not task_dir.exists(): return [] files sorted(task_dir.glob(*.json)) reports [] for f in files[-limit:]: reports.append(json.loads(f.read_text(encodingutf-8))) return reports def build_trend(reports): trend [] for r in reports: m r[metrics] trend.append({ run_id: r[run_id], started_at: r[started_at], requests_total: m[requests_total], requests_failed: m[requests_failed], parsed_items: m[parsed_items], response_time_avg: m[response_time_avg], failure_rate: round(m[requests_failed] / m[requests_total], 4) if m[requests_total] else 0, }) return trend这份trend列表可以直接打印成表格也可以喂给可视化模块。就算什么都不画在命令行里按列对齐打出来也能很直观看出一周的采集量变化。5.3 用matplotlib画三条关键曲线我通常看三张图采集量parsed_items、失败率failure_rate、平均响应时间response_time_avg。画图代码很朴素核心就是plt.plot(dates, values)加plt.xticks(rotation45)这里不贴完整示例了。真正让我印象深刻的是有一次失败率曲线从2%到2.5%到3.5%连续一周都在缓步上升像极了慢性病。单看任何一天都在阈值之内但趋势一眼就暴露了某个代理节点在批量失效。没有历史对比这个问题大概率要等到失败率到40%才开始触发告警。除了折线图还可以在HTML报告里附带一个与上次运行对比的表格比如采集量变化百分比、耗时变化百分比。这样每次跑完扫一眼就能知道这次是变好了还是变差了。6. 告警设计让报告自己开口说话报告写出来了如果没人看等于白写。下一步是让系统在报告不健康的时候主动通知人。这里我不建议一开始就上太重的平台对大部分爬虫项目来说轻量规则检查就够了。6.1 一个朴素的健康检查器class HealthChecker: def __init__(self, failure_rate_threshold0.2, min_parsed_items1): self.failure_rate_threshold failure_rate_threshold self.min_parsed_items min_parsed_items def check(self, report): m report[metrics] issues [] if m[requests_total] 0: issues.append((FATAL, no requests were made)) elif m[requests_failed] / m[requests_total] self.failure_rate_threshold: issues.append((WARN, failure_rate over threshold)) if m[parsed_items] self.min_parsed_items: issues.append((FATAL, parsed_items too low)) return issues这套检查器的核心思想是分权重requests_total 0说明任务可能压根没被调度属于FATALparsed_items 0说明可能网站改版也是FATAL失败率略高可能在网络波动范围内给WARN就够了。6.2 连续N次触发避免狼来了爬虫跑在公网上偶尔一次失败率超标太常见了。如果每次波动都告警一周后所有人都会把告警消息当垃圾信息。我的经验是加一个连续N次的规则只有当同一个任务连续N份报告都触发同类问题时才告警。实现时扫描最近N份报告如果每一份都命中同一检查项再发通知。这个连续N次极大减少了告警疲劳。我的默认值是连续3次。比如某个晚上目标网站短暂挂了那只会留下2份有问题的报告系统不会每份都轰炸一遍。6.3 最少但有效的告警通道告警通道我推荐从最简单的开始退出码加cron日志报告里带着FATAL问题时脚本以sys.exit(1)退出cron自己就能把失败的任务标出来邮件通知检查脚本周期性扫描一次报告发现问题发一封标题带run_id和任务名的邮件Webhook如果有企业微信、钉钉之类的群机器人把检查结果POST到webhook。告警消息再快也不如报告详细。无论你用哪种通道消息里至少要有任务名、run_id、触发等级、关键指标值。人接到告警后的第一个动作应该是打开对应JSON报告而不是继续翻聊天记录。7. 上线之后踩的坑那些不会出现在教程里的细节最后把这些年实际跑这套系统踩过的坑集中写一下也算补全一些教程不会提的边角。7.1 状态码200不代表解析成功这个坑最大也最隐蔽。HTTP层和数据层必须分开监控。我在第2节已经强调过这里说个真实案例某个列表页改版后服务端为了兼容旧客户端返回了200和一段空列表JSON。我的爬虫正常请求、正常解析结果parsed_items直接归零。如果当时只监控请求失败率这个故障能躺一周。对策很简单解析成功条数必须进入FATAL级别的告警维度一旦触底立即发警报。7.2 重试风暴会把监控数据污染掉重试逻辑写得不严谨会带来一个隐蔽问题假设某个节点持续超时每个URL都被重试3次那么requests_failed会被放大3倍。这倒不是坏消息坏的是失败率这个指标开始变得不可信。我的对策是重试次数单独计数不算进requests_failed给每个请求设置重试上限比如最多2次防止无限重试拖垮整个任务报告里单独加一个retry_storm_count字段记录重试过于密集的URL数量方便判断是否出现了系统性故障。7.3 报告本身也会膨胀报告虽然只有几十KB一个任务一天跑一次一年也就几十份听起来不多。但如果任务多、频率高每小时一次一年就能攒几千份JSON再加上异常样本里的URL和message体积会越写越大。我的处理方式告警级的报告全量保存正常级的报告只保留最近30天超过30天的旧报告压缩成gzip或者归档到冷存储anomalies.samples限制在50条以内防止某个异常类型把报告撑爆炸。7.4 时区不统一导致昨天的报告对不上这个坑我在第4节提过原则但上线后还是踩了一回。当时有个任务在服务器上按本地时区跑报告目录按UTC命名结果每天凌晨调度的文件经常落在前一天的分区里。排查了好久才发现是datetime.now()和datetime.now(timezone.utc)混用了。现在我的代码里所有时间生成点都统一走一个工具函数杜绝手工写时间。可复现系统的每个细节都要有唯一约定。时间、编码、目录格式、字段命名任何一处不一致都可能让历史对比工具读出来的数据前后对不上。这套监控系统我维护了大半年最大的体会不是有了告警再也不出事而是出事之后半小时内就能定位到是哪一版代码、哪一份配置、哪一次请求开始变坏的。爬虫这个行业真正稀有的不是写爬虫的能力而是对数据生产过程的掌控力。你不需要一开始就搭很重的平台先把RunContext类抄过去把请求入口埋点统一起来跑一次看一眼JSON报告后面加告警、加趋势对比都是顺水推舟的事。希望这篇分享能帮你少踩几个我之前踩过的坑。