爬虫数据质量监控——自动检测数据异常与完整性

爬虫跑得很好，但数据爬下来后发现——价格字段混进了"面议"、日期格式不统一、某一天的数据全为空、商品数量比昨天少了 30%。这篇讲怎么自动监控数据质量，让爬虫不光能爬，还能自己发现问题。

一、数据质量问题的常见类型

二、数据质量检查器

1. 基础检查框架

importpandasaspdimportnumpyasnpfromdatetimeimportdatetimeimportjsonclassDataQualityChecker:"""数据质量检查器"""def__init__(self,df):self.df=df self.report={"check_time":datetime.now().isoformat(),"total_rows":len(df),"total_columns":len(df.columns),"checks":[]}defcheck_null_ratio(self,column,threshold=0.1):"""检查空值率是否超过阈值"""null_ratio=self.df[column].isnull().mean()is_passed=null_ratio<=threshold self.report["checks"].append({"type":"空值检查","column":column,"value":f"{null_ratio:.1%}","threshold":f"≤{threshold:.0%}","passed":is_passed,"severity":"error"ifnull_ratio>thresholdelse"info"})returnis_passeddefcheck_value_range(self,column,min_val=None,max_val=None):"""检查数值是否在合理范围内"""col_data=pd.to_numeric(self.df[column],errors="coerce")issues=0ifmin_valisnotNone:issues+=(col_data<min_val).sum()ifmax_valisnotNone:issues+=(col_data>max_val).sum()is_passed=issues==0self.report["checks"].append({"type":"范围检查","column":column,"value":f"{issues}个异常值","threshold":f"{min_val}~{max_val}","passed":is_passed,"severity":"error"ifissues>0else"info"})returnis_passeddefcheck_duplicates(self,subset=None,threshold=0):"""检查重复数据"""duplicates=self.df.duplicated(subset=subset).sum()is_passed=duplicates<=threshold self.report["checks"].append({"type":"重复检查","column":subsetor"整行","value":f"{duplicates}条重复","threshold":f"≤{threshold}","passed":is_passed,"severity":"warning"ifduplicates>0else"info"})returnis_passeddefcheck_value_distribution(self,column,expected_mean,std_tolerance=3):"""检查数据分布是否异常（均值偏移）"""col_data=pd.to_numeric(self.df[column],errors="coerce").dropna()iflen(col_data)==0:returnFalseactual_mean=col_data.mean()actual_std=col_data.std()# 检查均值是否在 expected_mean ± 3*std 范围内lower=expected_mean-std_tolerance*actual_std upper=expected_mean+std_tolerance*actual_std is_passed=lower<=actual_mean<=upper self.report["checks"].append({"type":"分布检查","column":column,"value":f"均值{actual_mean:.2f}","threshold":f"{lower:.2f}~{upper:.2f}","passed":is_passed,"severity":"warning"ifnotis_passedelse"info"})returnis_passeddefcheck_row_count(self,expected_count,tolerance=0.2):"""检查数据量是否在预期范围内"""actual_count=len(self.df)min_expected=expected_count*(1-tolerance)max_expected=expected_count*(1+tolerance)is_passed=min_expected<=actual_count<=max_expected self.report["checks"].append({"type":"数量检查","column":"总行数","value":f"{actual_count}行","threshold":f"{min_expected:.0f}~{max_expected:.0f}","passed":is_passed,"severity":"error"ifnotis_passedelse"info"})returnis_passeddefget_summary(self):"""获取检查汇总"""total=len(self.report["checks"])passed=sum(1forcinself.report["checks"]ifc["passed"])failed=total-passed self.report["summary"]=f"{passed}/{total}检查通过"returnself.reportdefprint_report(self):"""打印检查报告"""print(f"\n{'='*50}")print(f"数据质量检查报告")print(f"检查时间:{self.report['check_time']}")print(f"数据规模:{self.report['total_rows']}行 ×{self.report['total_columns']}列")print(f"{'='*50}")forcheckinself.report["checks"]:status="✅"ifcheck["passed"]else"❌"print(f"\n{status}[{check['type']}]{check['column']}")print(f" 结果:{check['value']}")ifnotcheck["passed"]:print(f" 阈值:{check['threshold']}")print(f" 严重:{check['severity']}")print(f"\n{'='*50}")print(self.report.get("summary",""))print(f"{'='*50}")

2. 使用示例

# 假设这是今天爬取的数据df=pd.read_csv("today_products.csv")# 执行质量检查checker=DataQualityChecker(df)# 检查空值checker.check_null_ratio("price",threshold=0.05)checker.check_null_ratio("title",threshold=0.01)# 检查价格范围checker.check_value_range("price",min_val=0.1,max_val=100000)# 检查重复商品checker.check_duplicates(subset=["title"])# 检查数据量（期望每天 1000 条，允许 20% 波动）checker.check_row_count(expected_count=1000,tolerance=0.2)# 输出报告checker.print_report()

三、历史对比监控

1. 记录历史统计

importjsonimportosclassHistoryMonitor:"""历史数据监控器"""def__init__(self,history_file="data_quality_history.json"):self.history_file=history_file self.history=self._load_history()def_load_history(self):ifos.path.exists(self.history_file):withopen(self.history_file,"r")asf:returnjson.load(f)return{"daily_stats":[]}defrecord(self,stats):"""记录今天的统计"""stats["date"]=datetime.now().strftime("%Y-%m-%d")self.history["daily_stats"].append(stats)# 只保留最近 30 天iflen(self.history["daily_stats"])>30:self.history["daily_stats"]=self.history["daily_stats"][-30:]withopen(self.history_file,"w")asf:json.dump(self.history,f,ensure_ascii=False,indent=2)defdetect_anomaly(self,today_count):"""检测今天的数据量是否异常"""iflen(self.history["daily_stats"])<7:returnFalse# 数据不足，无法判断# 用最近 7 天的数据量计算均值和标准差recent_counts=[d["count"]fordinself.history["daily_stats"][-7:]]mean=np.mean(recent_counts)std=np.std(recent_counts)# 如果今天的数据量偏离均值超过 3 个标准差，视为异常ifstd>0andabs(today_count-mean)>3*std:returnTruereturnFalse

四、自动修复

classDataRepair:"""数据自动修复"""@staticmethoddeffill_missing_price(df):"""用同类商品的中位数填充缺失价格"""forcategoryindf["category"].unique():mask=(df["category"]==category)median_price=df.loc[mask,"price"].median()df.loc[mask&df["price"].isnull(),"price"]=median_pricereturndf@staticmethoddefremove_outliers(df,column,n_std=3):"""去除超出 n 个标准差的异常值"""mean=df[column].mean()std=df[column].std()lower=mean-n_std*std upper=mean+n_std*stdreturndf[(df[column]>=lower)&(df[column]<=upper)]@staticmethoddefstandardize_date(df,column,fmt="%Y-%m-%d"):"""统一日期格式"""df[column]=pd.to_datetime(df[column],errors="coerce")df[column]=df[column].dt.strftime(fmt)returndf

五、完整流水线

defdaily_crawl_pipeline():"""每日爬虫 + 数据质量监控流水线"""print(f"开始每日采集:{datetime.now()}")# 1. 爬取数据crawler=ProductCrawler()df=crawler.run()# 2. 数据清洗df=DataRepair.fill_missing_price(df)df=DataRepair.standardize_date(df,"crawl_time")# 3. 数据质量检查checker=DataQualityChecker(df)checker.check_null_ratio("title",0.01)checker.check_value_range("price",0,100000)checker.check_duplicates(subset=["product_id"])checker.check_row_count(1000,0.2)report=checker.get_summary()# 4. 历史监控monitor=HistoryMonitor()stats={"count":len(df),"null_ratio":df["price"].isnull().mean(),"avg_price":df["price"].mean(),}ifmonitor.detect_anomaly(len(df)):send_alert(f"数据量异常！今日{len(df)}条，历史均值波动较大")# 5. 保存数据df.to_csv(f"products_{datetime.now().strftime('%Y%m%d')}.csv",index=False)monitor.record(stats)# 6. 如果有检查不通过的，发送告警failed_checks=[cforcinreport["checks"]ifnotc["passed"]]iffailed_checks:send_alert(f"数据质量检查发现{len(failed_checks)}个问题："f"{json.dumps(failed_checks,ensure_ascii=False)}")print("采集流程完成")

六、告警通知

importsmtplibfromemail.mime.textimportMIMETextdefsend_alert(message):"""发送告警邮件"""msg=MIMEText(message,"plain","utf-8")msg["Subject"]=f"[数据质量告警]{datetime.now().strftime('%Y-%m-%d %H:%M')}"msg["From"]="crawler@example.com"msg["To"]="admin@example.com"withsmtplib.SMTP("smtp.example.com",587)asserver:server.login("user","password")server.send_message(msg)print(f"告警已发送:{message[:100]}...")

总结

数据质量监控 = 检查规则 + 历史对比 + 自动修复 + 告警通知 每次爬虫跑完自动执行： ✅ 空值检查 → 是否超过阈值 ✅ 范围检查 → 数值是否合理 ✅ 重复检查 → 数据是否重复 ✅ 数量检查 → 采集是否完整 ✅ 分布检查 → 均值是否偏移

💡 觉得有用的话，点赞 + 关注【张老师技术栈】吧！每周更新 Java/Python/爬虫 实战干货，不让你白来。