淺談增量式爬蟲

Redis · 發表 2019-02-24 22:23:00

摘要：引入在我們爬取某些網站時會遇到一些問題？某些網站會定時在原有網頁資料的基礎上更新一批資料。例如某電影網站會實時更新一批最近熱門的電影。小說網站會根據作者創作的進度實時更新最新的章節資料等等。那麼遇到類似的場景，我們就可以採用增量式爬蟲了而增量式爬蟲分為兩個步驟： ...

引入

在我們爬取某些網站時會遇到一些問題？某些網站會定時在原有網頁資料的基礎上更新一批資料。

例如某電影網站會實時更新一批最近熱門的電影。小說網站會根據作者創作的進度實時更新最新的章節資料等等。

那麼遇到類似的場景，我們就可以採用增量式爬蟲了

而增量式爬蟲分為兩個步驟：

增量爬取
爬取結果去重

增量爬取

一個站點更新也會出現下面兩種情況：

1，單個頁面資料更新

當出現這種情況的時候，我們對此特定頁面的內容做雜湊，當然要去除動態變化的那一部分，比如有的頁面有驗證碼或者日期，程式定期執行，在執行的最開始檢測此頁面的雜湊值跟上次抓取是否有變化，如果有變化就開始抓取。

2，新增了頁面

如果是新增頁面呢，我們會對頁面入口內容做雜湊，並且儲存分頁面的URL雜湊值，如果頁面入口雜湊值發生變化，獲取新增的頁面url列表，在這裡需要用到url的去重，和資料去重類似，採用redis集合型別處理。

redis集合型別不允許新增重複的資料，當新增重複的時候時，返回0，並且新增失敗。我們將所有的url list存入redis集合，當頁面入口變化時，進行頁面url去重，只抓取新增的頁面。

爬取結果去重

結果去重也有以下兩種常用的方法：

布隆過濾器

其中布隆過濾器是通過寫檔案的方式，多個程序使用需要新增同步和互斥，較為繁瑣，不推薦多執行緒/程序的時候使用，另外寫檔案是磁碟I/O操作，耗費時間長，可以累積到一定數量再一次寫入，或者利用上下文管理器在程式結束或異常退出時一次性寫入。

class Spider(object):
def __init():
# 布容過濾器初始化
self.burongname = 'test.bl'
if not os.path.isfile(self.burongname):
self.bl = BloomFilter(capacity=100000, error_rate=0.000001)
else:
with open(self.burongname, 'rb') as f:
self.bl = BloomFilter.fromfile(f)

def __enter__(self):
u"""
上下文管理器進入入口
"""
return self

def __exit__(self, *args):
u"""
上下文管理器，退出出口
"""
if self.conn is not None:
self.conn.close()

with open(self.burongname, 'wb') as f:
self.fingerprints.tofile(f)
def get_infos(self):
"""
抓取主函式
"""
# 布隆過濾器使用部分, x為抓取到得資料
x = json.dumps(i)
if x not in self.bl:
self.bl.add(x)

if __name__ == '__main__':
with Spider() as MSS:
MSS.get_infos()

上下文管理器，在主函式執行之前執行def enter ，在程式執行結束或異常退出時執行def exit ，上下文管理器還可以用來統計程式執行的時間。

redis集合

使用redis集合去重能夠支援多執行緒多程序.

利用redis集合無重複資料的特點，在redis建立集合，往其中新增資料的sha1值，新增成功返回1，表示無重複，新增失敗返回0，表示集合中已經有重複資料

使用步驟：

建立redis連線池
重複檢查

下面的例子是介面，並提供example。

[Redis]
server=192.168.0.100
pass=123@123

import sys
import hashlib
import os
import codecs
import ConfigParser
import redis


"""
利用redis的集合不允許新增重複元素來進行去重
"""


def example():
pool, r = redis_init()
temp_str = "aaaaaaaaa"
result = check_repeate(r, temp_str, 'test:test')
if result == 0:
print ("重複")
else:
print ("不重複")
redis_close(pool)

def redis_init(parasecname="Redis"):
"""
初始化redis
:return: redis連線池
"""
cur_script_dir = os.path.split(os.path.realpath(__file__))[0]
cfg_path = os.path.join(cur_script_dir, "db.conf")

cfg_reder = ConfigParser.ConfigParser()
secname = parasecname
cfg_reder.readfp(codecs.open(cfg_path, "r", "utf_8"))
redis_host = cfg_reder.get(secname, "server")
redis_pass = cfg_reder.get(secname, "pass")

# redis
pool = redis.ConnectionPool(host=redis_host, port=6379, db=0, password=redis_pass)
r = redis.Redis(connection_pool=pool)

return pool, r

def sha1(x):
sha1obj = hashlib.sha1()
sha1obj.update(x)
hash_value = sha1obj.hexdigest()
return hash_value

def check_repeate(r, check_str, set_name):
"""
向redis集合中新增元素，重複則返回0，不重複則新增成功，並返回1
:param r:redis連線
:param check_str:被新增的字串
:param set_name:專案所使用的集合名稱，建議如下格式：”projectname:task_remove_repeate“
"""
hash_value = sha1(check_str)
result = r.sadd(set_name, hash_value)
return result

def redis_close(pool):
"""
釋放redis連線池
"""
pool.disconnect()

if __name__ == '__main__':
example()