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

爬虫跑得很好但数据爬下来后发现——价格字段混进了面议、日期格式不统一、某一天的数据全为空、商品数量比昨天少了 30%。这篇讲怎么自动监控数据质量让爬虫不光能爬还能自己发现问题。一、数据质量问题的常见类型问题类型表现后果数据缺失某列空值突然增多分析结果偏差数据异常价格出现 999999统计结果失真格式不一致日期有 “2026-01-01” 和 “2026/01/01”入库报错重复数据同一条数据爬了多次统计分析翻倍数据量突变平时每天 1000 条今天只有 100 条可能采集失败二、数据质量检查器1. 基础检查框架importpandasaspdimportnumpyasnpfromdatetimeimportdatetimeimportjsonclassDataQualityChecker:数据质量检查器def__init__(self,df):self.dfdf self.report{check_time:datetime.now().isoformat(),total_rows:len(df),total_columns:len(df.columns),checks:[]}defcheck_null_ratio(self,column,threshold0.1):检查空值率是否超过阈值null_ratioself.df[column].isnull().mean()is_passednull_ratiothreshold self.report[checks].append({type:空值检查,column:column,value:f{null_ratio:.1%},threshold:f≤{threshold:.0%},passed:is_passed,severity:errorifnull_ratiothresholdelseinfo})returnis_passeddefcheck_value_range(self,column,min_valNone,max_valNone):检查数值是否在合理范围内col_datapd.to_numeric(self.df[column],errorscoerce)issues0ifmin_valisnotNone:issues(col_datamin_val).sum()ifmax_valisnotNone:issues(col_datamax_val).sum()is_passedissues0self.report[checks].append({type:范围检查,column:column,value:f{issues}个异常值,threshold:f{min_val}~{max_val},passed:is_passed,severity:errorifissues0elseinfo})returnis_passeddefcheck_duplicates(self,subsetNone,threshold0):检查重复数据duplicatesself.df.duplicated(subsetsubset).sum()is_passedduplicatesthreshold self.report[checks].append({type:重复检查,column:subsetor整行,value:f{duplicates}条重复,threshold:f≤{threshold},passed:is_passed,severity:warningifduplicates0elseinfo})returnis_passeddefcheck_value_distribution(self,column,expected_mean,std_tolerance3):检查数据分布是否异常均值偏移col_datapd.to_numeric(self.df[column],errorscoerce).dropna()iflen(col_data)0:returnFalseactual_meancol_data.mean()actual_stdcol_data.std()# 检查均值是否在 expected_mean ± 3*std 范围内lowerexpected_mean-std_tolerance*actual_std upperexpected_meanstd_tolerance*actual_std is_passedloweractual_meanupper self.report[checks].append({type:分布检查,column:column,value:f均值{actual_mean:.2f},threshold:f{lower:.2f}~{upper:.2f},passed:is_passed,severity:warningifnotis_passedelseinfo})returnis_passeddefcheck_row_count(self,expected_count,tolerance0.2):检查数据量是否在预期范围内actual_countlen(self.df)min_expectedexpected_count*(1-tolerance)max_expectedexpected_count*(1tolerance)is_passedmin_expectedactual_countmax_expected self.report[checks].append({type:数量检查,column:总行数,value:f{actual_count}行,threshold:f{min_expected:.0f}~{max_expected:.0f},passed:is_passed,severity:errorifnotis_passedelseinfo})returnis_passeddefget_summary(self):获取检查汇总totallen(self.report[checks])passedsum(1forcinself.report[checks]ifc[passed])failedtotal-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. 使用示例# 假设这是今天爬取的数据dfpd.read_csv(today_products.csv)# 执行质量检查checkerDataQualityChecker(df)# 检查空值checker.check_null_ratio(price,threshold0.05)checker.check_null_ratio(title,threshold0.01)# 检查价格范围checker.check_value_range(price,min_val0.1,max_val100000)# 检查重复商品checker.check_duplicates(subset[title])# 检查数据量期望每天 1000 条允许 20% 波动checker.check_row_count(expected_count1000,tolerance0.2)# 输出报告checker.print_report()三、历史对比监控1. 记录历史统计importjsonimportosclassHistoryMonitor:历史数据监控器def__init__(self,history_filedata_quality_history.json):self.history_filehistory_file self.historyself._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_asciiFalse,indent2)defdetect_anomaly(self,today_count):检测今天的数据量是否异常iflen(self.history[daily_stats])7:returnFalse# 数据不足无法判断# 用最近 7 天的数据量计算均值和标准差recent_counts[d[count]fordinself.history[daily_stats][-7:]]meannp.mean(recent_counts)stdnp.std(recent_counts)# 如果今天的数据量偏离均值超过 3 个标准差视为异常ifstd0andabs(today_count-mean)3*std:returnTruereturnFalse四、自动修复classDataRepair:数据自动修复staticmethoddeffill_missing_price(df):用同类商品的中位数填充缺失价格forcategoryindf[category].unique():mask(df[category]category)median_pricedf.loc[mask,price].median()df.loc[maskdf[price].isnull(),price]median_pricereturndfstaticmethoddefremove_outliers(df,column,n_std3):去除超出 n 个标准差的异常值meandf[column].mean()stddf[column].std()lowermean-n_std*std uppermeann_std*stdreturndf[(df[column]lower)(df[column]upper)]staticmethoddefstandardize_date(df,column,fmt%Y-%m-%d):统一日期格式df[column]pd.to_datetime(df[column],errorscoerce)df[column]df[column].dt.strftime(fmt)returndf五、完整流水线defdaily_crawl_pipeline():每日爬虫 数据质量监控流水线print(f开始每日采集:{datetime.now()})# 1. 爬取数据crawlerProductCrawler()dfcrawler.run()# 2. 数据清洗dfDataRepair.fill_missing_price(df)dfDataRepair.standardize_date(df,crawl_time)# 3. 数据质量检查checkerDataQualityChecker(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)reportchecker.get_summary()# 4. 历史监控monitorHistoryMonitor()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(fproducts_{datetime.now().strftime(%Y%m%d)}.csv,indexFalse)monitor.record(stats)# 6. 如果有检查不通过的发送告警failed_checks[cforcinreport[checks]ifnotc[passed]]iffailed_checks:send_alert(f数据质量检查发现{len(failed_checks)}个问题f{json.dumps(failed_checks,ensure_asciiFalse)})print(采集流程完成)六、告警通知importsmtplibfromemail.mime.textimportMIMETextdefsend_alert(message):发送告警邮件msgMIMEText(message,plain,utf-8)msg[Subject]f[数据质量告警]{datetime.now().strftime(%Y-%m-%d %H:%M)}msg[From]crawlerexample.commsg[To]adminexample.comwithsmtplib.SMTP(smtp.example.com,587)asserver:server.login(user,password)server.send_message(msg)print(f告警已发送:{message[:100]}...)总结数据质量监控 检查规则 历史对比 自动修复 告警通知 每次爬虫跑完自动执行 ✅ 空值检查 → 是否超过阈值 ✅ 范围检查 → 数值是否合理 ✅ 重复检查 → 数据是否重复 ✅ 数量检查 → 采集是否完整 ✅ 分布检查 → 均值是否偏移 觉得有用的话点赞 关注【张老师技术栈】吧每周更新 Java/Python/爬虫 实战干货不让你白来。