爬虫数据清洗与存储——从爬下来到用得上的最后一公里

爬虫最难的不是把数据爬下来而是把乱七八糟的原始数据洗干净存好。这一篇讲爬虫后处理的完整流程——清洗、去重、存储。一、脏数据的常见问题爬下来的原始数据基本是这样的data[{title: iPhone 15 ,price: ¥6,999 ,stock:100件},{title:华为 Mate 60,price:null,stock:50},{title:,price:5.99,stock:-1},{title:小米 14,price:3999,stock:abc},None,]问题清单空格、换行符、特殊符号混在字符串里价格带货币符号和逗号“¥6,999”空值、None、null 混杂数据类型不对库存是数字但爬下来是字符串脏数据负数库存、空标题重复数据二、pandas 数据清洗三板斧1. 去除空格和特殊字符importpandasaspd dfpd.DataFrame(data)# 去除字符串两端的空格和换行df[title]df[title].str.strip()# 去除所有空格有些用全角空格df[title]df[title].str.replace(r\s,,regexTrue)# 清洗价格字段去符号、去逗号df[price]df[price].str.replace(r[¥$,],,regexTrue)df[price]pd.to_numeric(df[price],errorscoerce)# errorscoerce遇到不能转的变成 NaN# 清洗库存df[stock]pd.to_numeric(df[stock],errorscoerce)print(df.dtypes)# 确认类型已经转成数字2. 处理空值# 查看哪些列有空值print(df.isnull().sum())# 删除空值过多的行dfdf.dropna(subset[title])# 标题为空就删掉# 填充数值列的空值df[price]df[price].fillna(0)df[stock]df[stock].fillna(0)# 或者用均值填充df[price]df[price].fillna(df[price].mean())# 填充字符串列df[brand]df[brand].fillna(未知)3. 去重# 查看重复行数print(df.duplicated().sum())# 删除完全重复的行dfdf.drop_duplicates()# 指定列去重保留第一次出现的dfdf.drop_duplicates(subset[title])# 查看某列重复值print(df[title].value_counts())# 过滤异常数据dfdf[df[price]0]# 价格必须大于0dfdf[df[stock]0]# 库存不能是负数4. 常用数据转换# 类型转换df[price]df[price].astype(float)df[stock]df[stock].astype(int)# 处理时间df[crawl_time]pd.to_datetime(df[crawl_time])# 提取字段中的信息df[city]df[address].str.extract(r(北京|上海|广州|深圳|郑州))# 统一文本格式如品牌名统一小写df[brand]df[brand].str.lower().str.strip()三、存储到 MySQL1. pandas 直接写入最简单fromsqlalchemyimportcreate_engineimportpandasaspd# 连接 MySQLenginecreate_engine(mysqlpymysql://root:123456localhost:3306/spider_db?charsetutf8mb4)# DataFrame 直接写入 MySQL 表# 如果表已存在replace 替换append 追加df.to_sql(nameproducts,# 表名conengine,if_existsappend,# 追加模式indexFalse,# 不保存 DataFrame 的索引chunksize1000,# 分批写入每批1000条)print(数据已写入 MySQL)2. 逐条写入适合自定义处理importpymysql connpymysql.connect(hostlocalhost,port3306,userroot,password123456,databasespider_db,charsetutf8mb4,)cursorconn.cursor()# 批量插入sqlINSERT INTO products (title, price, brand, stock, crawl_time) VALUES (%s, %s, %s, %s, %s)data_list[(iPhone 15,6999,Apple,100,2026-06-26),(华为 Mate 60,5999,华为,50,2026-06-26),]cursor.executemany(sql,data_list)conn.commit()print(f插入了{cursor.rowcount}条数据)cursor.close()conn.close()3. 更新 vs 插入去重逻辑重复爬取时同一件商品可能出现两次。处理方式# 方案一INSERT IGNORE主键或唯一索引冲突时跳过sqlINSERT IGNORE INTO products (id, title, price) VALUES (%s, %s, %s)# 方案二REPLACE INTO冲突时替换sqlREPLACE INTO products (id, title, price) VALUES (%s, %s, %s)# 方案三ON DUPLICATE KEY UPDATE冲突时更新指定字段sqlINSERT INTO products (id, title, price, crawl_time) VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE price VALUES(price), crawl_time VALUES(crawl_time)推荐方案三既保留旧数据又更新最新价格和爬取时间。四、存储到 MongoDBMongoDB 适合结构不固定的爬虫数据——不同商品字段可能不一样手机有像素冰箱有容积用 MongoDB 不用提前建表。1. 写入frompymongoimportMongoClient# 连接 MongoDBclientMongoClient(localhost,27017)dbclient[spider_db]collectiondb[products]# 插入一条collection.insert_one({title:iPhone 15,price:6999,brand:Apple,crawl_time:2026-06-26,})# 批量插入直接传 DataFramecollection.insert_many(df.to_dict(records))print(f插入了{len(df)}条)2. 去重更新# 以 title 为唯一标识有则更新无则插入for_,rowindf.iterrows():collection.update_one({title:row[title]},# 查询条件{$set:row.to_dict()},# 更新内容upsertTrue,# 不存在就插入)3. 查询# 价格大于5000的商品forproductincollection.find({price:{$gt:5000}}):print(product[title],product[price])# 按品牌统计数量pipeline[{$group:{_id:$brand,count:{$sum:1}}},{$sort:{count:-1}},]forresultincollection.aggregate(pipeline):print(result[_id],result[count])五、MySQL vs MongoDB 怎么选MySQLMongoDB数据格式需要提前建表字段固定JSON 格式字段灵活爬虫场景同一种商品字段稳定不同来源字段不一致查询适合关联查询、统计适合存和查不适合复杂关联上手需要建表写 SQL直接 insert零门槛实际建议爬电商商品字段固定→MySQL爬多种网站数据格式不统一 →MongoDB小数据量几千条→JSON 文件就够了六、完整流水线示例importpandasaspdfromsqlalchemyimportcreate_engineimportrequestsfrombs4importBeautifulSoupimporttimedefcrawl_products():爬虫 → 清洗 → 存储 一条龙# 1. 爬取all_data[]forpageinrange(1,6):resprequests.get(fhttps://example.com/products?page{page})soupBeautifulSoup(resp.text,html.parser)foriteminsoup.select(.product):all_data.append({title:item.select_one(.title).text,price:item.select_one(.price).text,brand:item.select_one(.brand).text,stock:item.select_one(.stock).text,})time.sleep(1)print(f爬取完成共{len(all_data)}条原始数据)# 2. 清洗dfpd.DataFrame(all_data)df[title]df[title].str.strip()df[price]df[price].str.replace(r[¥$,],,regexTrue)df[price]pd.to_numeric(df[price],errorscoerce)df[stock]pd.to_numeric(df[stock],errorscoerce)dfdf.dropna(subset[title])dfdf.drop_duplicates(subset[title])dfdf[df[price]0]df[crawl_time]pd.Timestamp.now()print(f清洗后剩余{len(df)}条有效数据)# 3. 存储enginecreate_engine(mysqlpymysql://root:123456localhost:3306/spider_db?charsetutf8mb4)df.to_sql(products,conengine,if_existsappend,indexFalse)print(数据已保存到 MySQL)returndfif__name____main__:crawl_products()总结爬虫的核心公式爬下来 → 清洗去空格/去重/类型转换/去异常 → 存储MySQL/MongoDB/文件 ↑ ↑ 最耗时的环节 按场景选很多人把精力都放在爬取上结果数据存进去一堆脏数据后面分析根本没法用。洗数据花的功夫往往比写爬虫还多但这步做好了后面的分析才会顺利。 觉得有用的话点赞 关注【张老师技术栈】吧每周更新 Java/Python/爬虫 实战干货不让你白来。