?

From：https://cloud.tencent.com/developer/article/1084962

Python分布式爬蟲打造搜索引擎Scrapy精講—將bloomfilter(布隆過濾器)集成到scrapy-redis中
https://www.cnblogs.com/adc8868/p/7442306.html

?

scrapy redis + bloomfilter ：https://github.com/pujinxiao/project_pjx/tree/master/s0vkaq/ScrapyRedisTest/ScrapyRedisTest/utils

?

具體的 bloomfilter 概念和原理應該查看這篇文章：傳送，還有《海量數據處理算法》以及《大規模數據處理利器》

BloomFilter Based on py3(基于py3的布隆過濾器)：https://github.com/Sssmeb/BloomFilter

?

首先回顧一下 Scrapy-Redis 的去重機制。Scrapy-Redis 將 Request 的指紋存儲到了 Redis 集合中，每個指紋的長度為40，例如27adcc2e8979cdee0c9cecbbe8bf8ff51edefb61 就是一個指紋，它的每一位都是16 進制數。

我們計算一下用這種方式耗費的存儲空間。每個十六進制數占用 4 b （bit），1個指紋用40個十六進制數表示，占用空間為20 B (Byte)，1萬個指紋即占用空間200 KB，1億個指紋占用 2 GB。當爬取數量達到上億級別時，Redis的占用的內存就會變得很大，而且這僅僅是指紋的存儲。Redis還存儲了爬取隊列，內存占用會進一步提高，更別說有多個Scrapy項目同時爬取的情況了。當爬取達到億級別規模時，Scrapy-Redis提供的集合去重已經不能滿足我們的要求。所以我們需要使用一個更加節省內存的去重算法Bloom Filter。

?

?

?

Bloom Filter

?

?

1. 了解Bloom Filter

?

Bloom Filter，中文名稱叫作布隆過濾器，是1970年由Bloom提出的，它可以被用來檢測一個元素是否在一個集合中。Bloom Filter的空間利用效率很高，使用它可以大大節省存儲空間。Bloom Filter使用位數組表示一個待檢測集合，并可以快速地通過概率算法判斷一個元素是否存在于這個集合中。利用這個算法我們可以實現去重效果。

本節我們來了解Bloom Filter的基本算法，以及Scrapy-Redis中對接Bloom Filter的方法。

?

?

2. Bloom Filter的算法

?

在Bloom Filter中使用位數組來輔助實現檢測判斷。在初始狀態下，我們聲明一個包含m位的位數組，它的所有位都是0，如下圖所示。

現在我們有了一個待檢測集合，其表示為S={x1, x2, …, xn}。接下來需要做的就是檢測一個x是否已經存在于集合S中。在Bloom Filter算法中，首先使用 k 個相互獨立、隨機的散列函數 來 將 集合S 中的每個元素 x1, x2, …, xn 映射到長度為 m 的位數組上，散列函數得到的結果記作位置索引，然后將位數組該位置索引的位置 1。例如，我們取 k為 3，表示有三個散列函數，x1 經過三個散列函數映射得到的結果分別為 1、4、8，x2 經過三個散列函數映射得到的結果分別為 4、6、10，那么位數組的1、4、6、8、10 這五位就會置為1，如下圖所示。

如果有一個新的元素x，我們要判斷x是否屬于S集合，我們仍然用k個散列函數對x求映射結果。如果所有結果對應的位數組位置均為1，那么x屬于S這個集合；如果有一個不為1，則x不屬于S集合。

例如，新元素x經過三個散列函數映射的結果為4、6、8，對應的位置均為1，則x屬于S集合。如果結果為4、6、7，而7對應的位置為0，則x不屬于S集合。

注意，這里m、n、k滿足的關系是m>nk，也就是說位數組的長度m要比集合元素n和散列函數k的乘積還要大。

這樣的判定方法很高效，但是也是有代價的，它可能把不屬于這個集合的元素誤認為屬于這個集合。我們來估計一下這種方法的錯誤率。當集合S={x1, x2,…, xn} 的所有元素都被k個散列函數映射到m位的位數組中時，這個位數組中某一位還是0的概率是：

散列函數是隨機的，則任意一個散列函數選中這一位的概率為1/m，那么1-1/m就代表散列函數從未沒有選中這一位的概率，要把S完全映射到m位數組中，需要做kn次散列運算，最后的概率就是1-1/m的kn次方。

一個不屬于S的元素x如果誤判定為在S中，那么這個概率就是k次散列運算得到的結果對應的位數組位置都為1，則誤判概率為：

根據：

可以將誤判概率轉化為：

在給定m、n時，可以求出使得f最小化的k值為：

這里將誤判概率歸納如下：

表中第一列為m/n的值，第二列為最優k值，其后列為不同k值的誤判概率。當k值確定時，隨著m/n的增大，誤判概率逐漸變小。當m/n的值確定時，當k越靠近最優K值，誤判概率越小。誤判概率總體來看都是極小的，在容忍此誤判概率的情況下，大幅減小存儲空間和判定速度是完全值得的。

簡單點說：Bloom Filter算法 就是有幾個seeds，現在申請一段內存空間，一個seed可以和字符串哈希映射到這段內存上的一個位，幾個位都為1即表示該字符串已經存在。插入的時候也是，將映射出的幾個位都置為1

接下來，我們將Bloom Filter算法應用到Scrapy-Redis分布式爬蟲的去重過程中，以解決Redis內存不足的問題。

?

布隆 優點

?

相比于其它的數據結構，布隆過濾器在空間和時間方面都有巨大的優勢。布隆過濾器存儲空間和插入/查詢時間都是常數。另外, Hash 函數相互之間沒有關系，方便由硬件并行實現。布隆過濾器不需要存儲元素本身，在某些對保密要求非常嚴格的場合有優勢。

布隆過濾器可以表示全集，其它任何數據結構都不能；

k 和 m 相同，使用同一組 Hash 函數的兩個布隆過濾器的交并差運算可以使用位操作進行。

?

布隆 缺點

?

但是布隆過濾器的缺點和優點一樣明顯。誤算率（False Positive）是其中之一。隨著存入的元素數量增加，誤算率隨之增加。但是如果元素數量太少，則使用散列表足矣。

另外，一般情況下不能從布隆過濾器中刪除元素. 我們很容易想到把位列陣變成整數數組，每插入一個元素相應的計數器加1, 這樣刪除元素時將計數器減掉就可以了。然而要保證安全的刪除元素并非如此簡單。首先我們必須保證刪除的元素的確在布隆過濾器里面. 這一點單憑這個過濾器是無法保證的。另外計數器回繞也會造成問題。

總的來說，布隆很適合來處理海量的數據，而且速度優勢很強。

?

?

redis 與 bloom

?

去重”是日常工作中會經常用到的一項技能，在爬蟲領域更是常用，并且規模一般都比較大。參考文章《基于Redis的Bloomfilter去重》，作者【九茶】還有另一篇文章可以參考《scrapy_redis去重優化，已有7億條數據》

?

去重需要考慮兩個點：去重的數據量、去重速度。為了保持較快的去重速度，一般選擇在內存中進行去重。

• 數據量不大時，可以直接放在內存里面進行去重，例如python可以使用set()進行去重。
• 當去重數據需要持久化時可以使用redis的set數據結構。
• 當數據量再大一點時，可以用不同的加密算法先將長字符串壓縮成 16/32/40 個字符，再使用上面兩種方法去重；
• 當數據量達到億（甚至十億、百億）數量級時，內存有限，必須用“位”來去重，才能夠滿足需求。Bloomfilter就是將去重對象映射到幾個內存“位”，通過幾個位的 0/1值來判斷一個對象是否已經存在。
• 然而Bloomfilter運行在一臺機器的內存上，不方便持久化（機器down掉就什么都沒啦），也不方便分布式爬蟲的統一去重。如果可以在Redis上申請內存進行Bloomfilter，以上兩個問題就都能解決了。

?

?

?

對接Scrapy-Redis

?

實現 Bloom Filter 時，首先要保證不能破壞 Scrapy-Redis 分布式爬取的運行架構。我們需要修改 Scrapy-Redis 的源碼，將它的去重類替換掉。同時，Bloom Filter 的實現需要借助于一個位數組，既然當前架構還是依賴于Redis，那么位數組的維護直接使用Redi s就好了。

首先實現一個基本的散列算法，將一個值經過散列運算后映射到一個m位數組的某一位上，代碼如下：

BLOOMFILTER_HASH_NUMBER = 6 BLOOMFILTER_BIT = 30class HashMap(object):def __init__(self, m, seed):self.m = mself.seed = seeddef hash(self, value):"""Hash Algorithm:param value: Value:return: Hash Value"""ret = 0for i in range(len(value)):ret += self.seed * ret + ord(value[i])return (self.m - 1) & retclass BloomFilter(object):def __init__(self, server, key, bit=BLOOMFILTER_BIT, hash_number=BLOOMFILTER_HASH_NUMBER):"""Initialize BloomFilter:param server: Redis Server:param key: BloomFilter Key:param bit: m = 2 ^ bit:param hash_number: the number of hash function"""# default to 1 << 30 = 10,7374,1824 = 2^30 = 128MB, max filter 2^30/hash_number = 1,7895,6970 fingerprintsself.m = 1 << bitself.seeds = range(hash_number)self.maps = [HashMap(self.m, seed) for seed in self.seeds]self.server = serverself.key = keydef exists(self, value):"""if value exists:param value::return:"""if not value:return Falseexist = 1for map in self.maps:offset = map.hash(value)exist = exist & self.server.getbit(self.key, offset)return existdef insert(self, value):"""add value to bloom:param value::return:"""for f in self.maps:offset = f.hash(value)self.server.setbit(self.key, offset, 1)import redis conn = redis.StrictRedis(host='localhost', port=6379) bf = BloomFilter(conn, 'testbf', 5, 6) bf.insert('Hello') bf.insert('World') result = bf.exists('Hello') print(bool(result)) result = bf.exists('Python') print(bool(result))

這里新建了一個 HashMap 類。構造函數傳入兩個值，一個是 m 位數組的位數，另一個是種子值 seed。不同的散列函數需要有不同的 seed，這樣可以保證不同的散列函數的結果不會碰撞。

在hash()方法的實現中，value是要被處理的內容。這里遍歷了value的每一位，并利用 ord() 方法取到每一位的ASCII碼值，然后混淆 seed進行迭代求和運算，最終得到一個數值。這個數值的結果就由 value 和seed 唯一確定。我們再將這個數值和 m進行按位與運算，即可獲取到m位數組的映射結果，這樣就實現了一個由 字符串 和 seed 來確定的散列函數。當 m 固定時，只要seed值相同，散列函數就是相同的，相同的value必然會映射到相同的位置。所以如果想要構造幾個不同的散列函數，只需要改變其seed就好了。以上內容便是一個簡易的散列函數的實現。

接下來就是實現 Bloom Filter。Bloom Filter里面需要用到k個散列函數，這里要對這幾個散列函數指定相同的m值和不同的seed值，

由于我們需要億級別的數據的去重，即前文介紹的算法中的n為1億以上，散列函數的個數k大約取10左右的量級。而m>kn，這里m值大約保底在10億，由于這個數值比較大，所以這里用移位操作來實現，傳入位數bit，將其定義為30，然后做一個移位操作1<<30，相當于2的30次方，等于1073741824，量級也是恰好在10億左右，由于是位數組，所以這個位數組占用的大小就是2^30 b=128 MB。開頭我們計算過Scrapy-Redis集合去重的占用空間大約在2 GB左右，可見Bloom Filter的空間利用效率極高。

隨后我們再傳入散列函數的個數，用它來生成幾個不同的seed。用不同的seed來定義不同的散列函數，這樣我們就可以構造一個散列函數列表。遍歷seed，構造帶有不同seed值的HashMap對象，然后將HashMap對象保存成變量maps供后續使用。

另外，server就是Redis連接對象，key就是這個m位數組的名稱。

接下來，我們要實現比較關鍵的兩個方法：一個是判定元素是否重復的方法exists()，另一個是添加元素到集合中的方法insert()

首先看下insert()方法。Bloom Filter算法會逐個調用散列函數對放入集合中的元素進行運算，得到在m位位數組中的映射位置，然后將位數組對應的位置置1。這里代碼中我們遍歷了初始化好的散列函數，然后調用其hash()方法算出映射位置offset，再利用Redis的setbit()方法將該位置1。

在exists()方法中，我們要實現判定是否重復的邏輯，方法參數value為待判斷的元素。我們首先定義一個變量exist，遍歷所有散列函數對value進行散列運算，得到映射位置，用getbit()方法取得該映射位置的結果，循環進行與運算。這樣只有每次getbit()得到的結果都為1時，最后的exist才為True，即代表value屬于這個集合。如果其中只要有一次getbit()得到的結果為0，即m位數組中有對應的0位，那么最終的結果exist就為False，即代表value不屬于這個集合。

Bloom Filter的實現就已經完成了， 下面 就是 用一個實例來測試一下。

這里首先定義了一個Redis連接對象，然后傳遞給Bloom Filter。為了避免內存占用過大，這里傳的位數bit比較小，設置為5，散列函數的個數設置為6。

調用insert()方法插入Hello和World兩個字符串，隨后判斷Hello和Python這兩個字符串是否存在，最后輸出它的結果，運行結果如下：

True False

很明顯，結果完全沒有問題。這樣我們就借助Redis成功實現了Bloom Filter的算法。

接下來繼續修改Scrapy-Redis的源碼，將它的dupefilter邏輯替換為Bloom Filter的邏輯。這里主要是修改RFPDupeFilter類的request_seen()方法，實現如下：

def request_seen(self, request):fp = self.request_fingerprint(request) if self.bf.exists(fp): return Trueself.bf.insert(fp) return False

利用 request_fingerprint()方法獲取Request的指紋，調用Bloom Filter的exists()方法判定該指紋是否存在。如果存在，則說明該Request是重復的，返回True，否則調用Bloom Filter的insert()方法將該指紋添加并返回False。這樣就成功利用Bloom Filter替換了Scrapy-Redis的集合去重。

對于Bloom Filter的初始化定義，我們可以將__init__()方法修改為如下內容：

def __init__(self, server, key, debug, bit, hash_number):self.server = serverself.key = keyself.debug = debugself.bit = bitself.hash_number = hash_numberself.logdupes = Trueself.bf = BloomFilter(server, self.key, bit, hash_number)

其中bit和hash_number需要使用from_settings()方法傳遞，修改如下：

@classmethod def from_settings(cls, settings):server = get_redis_from_settings(settings)key = defaults.DUPEFILTER_KEY % {'timestamp': int(time.time())}debug = settings.getbool('DUPEFILTER_DEBUG', DUPEFILTER_DEBUG)bit = settings.getint('BLOOMFILTER_BIT', BLOOMFILTER_BIT)hash_number = settings.getint('BLOOMFILTER_HASH_NUMBER', BLOOMFILTER_HASH_NUMBER) return cls(server, key=key, debug=debug, bit=bit, hash_number=hash_number)

其中，常量DUPEFILTER_DEBUG和BLOOMFILTER_BIT統一定義在defaults.py中，默認如下：

BLOOMFILTER_HASH_NUMBER = 6 BLOOMFILTER_BIT = 30

現在，我們成功實現了Bloom Filter和Scrapy-Redis的對接。

?

代碼地址為：https://github.com/Python3WebSpider/ScrapyRedisBloomFilter

?

使用的方法和 Scrapy-Redis基本相似，在這里說明幾個關鍵配置。

# 去重類，要使用Bloom Filter請替換DUPEFILTER_CLASS DUPEFILTER_CLASS = "scrapy_redis_bloomfilter.dupefilter.RFPDupeFilter" # 散列函數的個數，默認為6，可以自行修改 BLOOMFILTER_HASH_NUMBER = 6 # Bloom Filter的bit參數，默認30，占用128MB空間，去重量級1億 BLOOMFILTER_BIT = 30

DUPEFILTER_CLASS是去重類，如果要使用Bloom Filter，則DUPEFILTER_CLASS需要修改為該包的去重類。

BLOOMFILTER_HASH_NUMBER是Bloom Filter使用的散列函數的個數，默認為6，可以根據去重量級自行修改。

BLOOMFILTER_BIT即前文所介紹的BloomFilter類的bit參數，它決定了位數組的位數。如果BLOOMFILTER_BIT為30，那么位數組位數為2的30次方，這將占用Redis 128 MB的存儲空間，去重量級在1億左右，即對應爬取量級1億左右。如果爬取量級在10億、20億甚至100億，請務必將此參數對應調高。

?

?

測試

?

源代碼附有一個測試項目，放在tests文件夾，該項目使用了ScrapyRedisBloomFilter來去重，Spider的實現如下：

from scrapy import Request, Spiderclass TestSpider(Spider):name = 'test'base_url = 'https://www.baidu.com/s?wd='def start_requests(self):for i in range(10):url = self.base_url + str(i) yield Request(url, callback=self.parse)# Here contains 10 duplicated Requests for i in range(100): url = self.base_url + str(i) yield Request(url, callback=self.parse) def parse(self, response):self.logger.debug('Response of ' + response.url)

start_requests()方法首先循環10次，構造參數為0~9的URL，然后重新循環了100次，構造了參數為0~99的URL。那么這里就會包含10個重復的Request，我們運行項目測試一下：

scrapy crawl test

最后的輸出結果如下：

{'bloomfilter/filtered': 10, 'downloader/request_bytes': 34021, 'downloader/request_count': 100, 'downloader/request_method_count/GET': 100, 'downloader/response_bytes': 72943, 'downloader/response_count': 100, 'downloader/response_status_count/200': 100, 'finish_reason': 'finished', 'finish_time': datetime.datetime(2017, 8, 11, 9, 34, 30, 419597), 'log_count/DEBUG': 202, 'log_count/INFO': 7, 'memusage/max': 54153216, 'memusage/startup': 54153216, 'response_received_count': 100, 'scheduler/dequeued/redis': 100, 'scheduler/enqueued/redis': 100, 'start_time': datetime.datetime(2017, 8, 11, 9, 34, 26, 495018)}

最后統計的第一行的結果：

'bloomfilter/filtered': 10,

這就是Bloom Filter過濾后的統計結果，它的過濾個數為10個，也就是它成功將重復的10個Reqeust識別出來了，測試通過。
?

以上內容便是Bloom Filter的原理及對接實現，Bloom Filter的使用可以大大節省Redis內存。在數據量大的情況下推薦此方案。

?

?

?

總結

以上是生活随笔為你收集整理的Scrapy 爬虫去重效率优化之 Bloom Filter的算法的对接的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。