一、业务场景
在一站式日淘全品类平台中，商品数据需要持续更新——价格变动、库存变化、新商品上架。如果每次都全量抓取，效率和成本都是问题。
以乐天为例，平台上有数百万商品。全量抓取一次需要几小时甚至几天，而且大部分商品的数据其实没有变化。
二、增量抓取的核心思路
增量抓取的核心是：只抓取发生变化的数据。
python
class IncrementalCrawler: def init(self): self.redis = redis.Redis(decode_responses=True) def crawl(self, platform): # 1. 获取上次抓取的时间戳 last_crawl = self.redis.get(f"last_crawl:{platform}") if not last_crawl: last_crawl = time.time() - 86400 # 默认24小时前 # 2. 只抓取上次之后更新的商品 items = self.fetch_updated_items(platform, last_crawl) # 3. 更新数据库 for item in items: self.update_or_insert(item) # 4. 记录本次抓取时间 self.redis.set(f"last_crawl:{platform}", time.time())
三、基于时间戳的增量策略
如果平台提供"按更新时间查询"的API，增量抓取就很简单：
python
def fetch_updated_items_rakuten(since_timestamp): """乐天：使用API的update_time参数""" url = "https://api.rms.rakuten.co.jp/es/1.0/search" params = { 'update_from': since_timestamp, 'hits': 100, 'page': 1 } # 分页获取所有更新的商品 all_items = [] while True: response = requests.get(url, params=params) data = response.json() all_items.extend(data['hits']) if data['page'] >= data['pageCount']: break params['page'] += 1 return all_items
四、基于版本号的增量策略
如果平台不提供时间戳查询（比如骏河屋），可以用版本号对比：
python
def fetch_updated_items_surugaya(): """骏河屋：通过对比版本号判断是否变化""" # 获取当前所有商品ID和版本号 current = get_all_item_versions() # 从数据库获取上次的版本号 previous = get_previous_versions() # 找出变化的商品 changed_ids = [] for item_id, version in current.items(): if item_id not in previous or previous[item_id] != version: changed_ids.append(item_id) # 只抓取变化的商品详情 items = [] for item_id in changed_ids: items.append(fetch_detail(item_id)) # 更新版本号 save_versions(current) return items
五、基于Hash的增量策略
对于内容变化不确定的场景，可以用内容Hash做对比：
python
import hashlibdef get_content_hash(item): """计算商品内容的Hash值""" content = f"{item['title']}{item['price']}{item['status']}" return hashlib.md5(content.encode()).hexdigest()def incremental_crawl_by_hash(items): """通过Hash对比判断是否需要更新""" for item in items: new_hash = get_content_hash(item) db_item = get_from_db(item['id']) if not db_item or db_item['hash'] != new_hash: # 数据发生了变化，更新数据库 update_db(item) # 触发后续处理（如价格变动告警） if db_item and db_item['price'] != item['price']: trigger_price_alert(item)
六、分级更新策略
不同商品的重要性不同，更新频率也应该不同：
python
class TieredUpdateStrategy: def get_update_priority(self, item): """根据商品属性确定更新优先级""" # 高优先级：热门商品、价格波动大的商品 if item['view_count'] > 10000: return 'high' # 每5分钟更新 elif item['view_count'] > 1000: return 'medium' # 每小时更新 else: return 'low' # 每天更新
七、总结
增量抓取的核心是减少无效请求。时间戳是最简单的方式，版本号是次优选择，Hash是最通用的方案。分级更新进一步优化了资源分配。这套策略让日均抓取量从"全量百万级"降到了"增量万级"，效率提升了90%以上。
