把布隆過濾器用起來
本文偏應用和程式碼實踐，理論請參考本文末尾參考文章
簡介
一句話簡介：過濾器，判斷這個元素在與不在，不在則100%不在；在則去查詢，b確認在不在。
詳細簡介：BloomFilter，中文名稱叫做布隆過濾器，是1970年由 Bloom 提出的，它可以被用來檢測一個元素是否在一個集合中，它的空間利用效率很高，使用它可以大大節省儲存空間。BloomFilter 使用位陣列表示一個待檢測集合，並可以快速地通過概率演演算法判斷一個元素是否存在於這個集合中，所以利用這個演演算法我們可以實現去重效果。
它的優點是空間效率和查詢時間都遠遠超過一般演演算法，缺點是有一定的誤識別率和刪除困難。
場景
1、大量爬蟲資料去重
2、保護資料安全：廣告精確投放 ：廣告主通過裝置id，計算hash演演算法，在資料包（資料提供方）中去查詢，如果在存在，則證明該裝置id屬於目標人群，進行投放廣告，同時保證裝置id不洩露。資料提供方和廣告主都沒有暴露自己擁有的裝置id。間接使用者畫像且不違資料安全法。詳見：https://zhuanlan.zhihu.com/p/37847480
3、比特幣網路轉賬確認
SPV節點:SPV是“Simplified Payment Verification”（簡單支付驗證）的縮寫。中本聰論文簡要地提及了這一概念，指出：不執行完全節點也可驗證支付，使用者只需要儲存所有的block header就可以了。使用者雖然不能自己驗證交易，但如果能夠從區塊鏈的某處找到相符的交易，他就可以知道網路已經認可了這筆交易，而且得到了網路的多少個確認。
4、分散式系統（Map-Reduce）把大任務切分成塊，分配和驗證一個子任務是否在一個子系統上。
必要性
省空間，提升效率
我們首先來回顧一下 ScrapyRedis 的去重機制，它將 Request 的指紋儲存到了 Redis 集合中，每個指紋的長度為 40，例如 27adcc2e8979cdee0c9cecbbe8bf8ff51edefb61 就是一個指紋，它的每一位都是 16 進位制數。
讓我們來計算一下用這種方式耗費的儲存空間，每個 16 進位制數佔用 4b，1 個指紋用 40 個 16 進位制數表示，佔用空間為 20B，所以 1 萬個指紋即佔用空間 200KB，1 億個指紋即佔用 2G，所以當我們的爬取數量達到上億級別時，Redis 的佔用的記憶體就會變得很高，而且這僅僅是指紋的儲存，另外 Redis 還儲存了爬取佇列，記憶體佔用會進一步提高，更別說有多個 Scrapy 專案同時爬取的情況了。所以當爬取達到億級別規模時 ScrapyRedis 提供的集合去重已經不能滿足我們的要求，所以在這裡我們需要使用一個更加節省記憶體的去重演演算法，它叫做 BloomFilter。
(記憶體版)Python實現的記憶體版布隆過濾器pybloom
https://github.com/jaybaird/python-bloomfilter安裝:
pip install pybloom複製程式碼該模組包含兩個類實現布隆過濾器功能。BloomFilter 是定容。ScalableBloomFilter 可以自動擴容
使用:
>>> from pybloom import BloomFilter
>>> f = BloomFilter(capacity=1000,error_rate=0.001) # capacity是容量,error_rate 是能容忍的誤報率
>>> f.add('Traim304') # 當不存在該元素,返回False
False
>>> f.add('Traim304') # 若存在,返回 True
True
>>> 'Traim304' in f # 值得注意的是若返回 True。該元素可能存在,也可能不存在。過濾器能容許存在一定的錯誤
True
>>> 'Jacob' in f # 但是 False。則必定不存在
False
>>> len(f) # 當前存在的元素
1
>>> f = BloomFilter(capacity=1000,error_rate=0.001)
>>> from pybloom import ScalableBloomFilter
>>> sbf = ScalableBloomFilter(mode=ScalableBloomFilter.SMALL_SET_GROWTH)
>>> # sbf.add() 與 BloomFilter 同複製程式碼超過誤報率時丟擲異常
>>> f = BloomFilter(capacity=1000,error_rate=0.0000001)
>>> for a in range(1000):
... _ = f.add(a)
...
>>> len(a)
Traceback (most recent call last):
File "<stdin>",line 1,in <module>
TypeError: object of type 'int' has no len()
>>> len(f)
1000
>>> f.add(1000)
False
>>> f.add(1001) # 當誤報率超過 error_rate 會報錯
Traceback (most recent call last):
File "<stdin>",in <module>
File "/usr/local/lib/python2.7/site-packages/pybloom/pybloom.py",line 182,in add
raise IndexError("BloomFilter is at capacity")
IndexError: BloomFilter is at capacity複製程式碼(持久化)手動實現的redis版布隆過濾器
大資料量，多用Redis持久化版本的布隆過濾器
# 布隆過濾器 redis版本實現
import hashlib
import redis
import six
# 1. 多個hash函式的實現和求值
# 2. hash表實現和實現對應的對映和判斷
class MultipleHash(object):
'''根據提供的原始資料，和預定義的多個salt，生成多個hash函式值'''
def __init__(self,salts,hash_func_name="md5"):
self.hash_func = getattr(hashlib,hash_func_name)
if len(salts) < 3:
raise Exception("請至少提供3個salt")
self.salts = salts
def get_hash_values(self,data):
'''根據提供的原始資料,返回多個hash函式值'''
hash_values = []
for i in self.salts:
hash_obj = self.hash_func()
hash_obj.update(self._safe_data(data))
hash_obj.update(self._safe_data(i))
ret = hash_obj.hexdigest()
hash_values.append(int(ret,16))
return hash_values
def _safe_data(self,data):
'''
python2 str === python3 bytes
python2 uniocde === python3 str
:param data: 給定的原始資料
:return: 二進位制型別的字串資料
'''
if six.PY3:
if isinstance(data,bytes):
return data
elif isinstance(data,str):
return data.encode()
else:
raise Exception("請提供一個字串") # 建議使用英文來描述
else:
if isinstance(data,str):
return data
elif isinstance(data,unicode):
return data.encode()
else:
raise Exception("請提供一個字串") # 建議使用英文來描述
class BloomFilter(object):
''''''
def __init__(self,redis_host="localhost",redis_port=6379,redis_db=0,redis_key="bloomfilter"):
self.redis_host = redis_host
self.redis_port = redis_port
self.redis_db = redis_db
self.redis_key = redis_key
self.client = self._get_redis_client()
self.multiple_hash = MultipleHash(salts)
def _get_redis_client(self):
'''返回一個redis連線物件'''
pool = redis.ConnectionPool(host=self.redis_host,port=self.redis_port,db=self.redis_db)
client = redis.StrictRedis(connection_pool=pool)
return client
def save(self,data):
''''''
hash_values = self.multiple_hash.get_hash_values(data)
for hash_value in hash_values:
offset = self._get_offset(hash_value)
self.client.setbit(self.redis_key,offset,1)
return True
def is_exists(self,data):
hash_values = self.multiple_hash.get_hash_values(data)
for hash_value in hash_values:
offset = self._get_offset(hash_value)
v = self.client.getbit(self.redis_key,offset)
if v == 0:
return False
return True
def _get_offset(self,hash_value):
# 512M長度雜湊表
# 2**8 = 256
# 2**20 = 1024 * 1024
# (2**8 * 2**20 * 2*3) 代表hash表的長度 如果同一專案中不能更改
return hash_value % (2**8 * 2**20 * 2*3)
if __name__ == '__main__':
data = ["asdfasdf","123","456","asf","asf"]
bm = BloomFilter(salts=["1","2","3","4"],redis_host="172.17.0.2")
for d in data:
if not bm.is_exists(d):
bm.save(d)
print("對映資料成功： ",d)
else:
print("發現重複資料：",d)
複製程式碼應用在scrapy-redis中
程式碼已經打包成了一個 Python 包併發布到了 PyPi，連結為：https://pypi.python.org/pypi/scrapy-redis-bloomfilter，因此我們以後如果想使用 ScrapyRedisBloomFilter 直接使用就好了，不需要再自己實現一遍。
我們可以直接使用Pip來安裝，命令如下：
pip3 install scrapy-redis-bloomfilter複製程式碼使用的方法和 ScrapyRedis 基本相似，在這裡說明幾個關鍵配置：
# 去重類，要使用BloomFilter請替換DUPEFILTER_CLASS
DUPEFILTER_CLASS = "scrapy_redis_bloomfilter.dupefilter.RFPDupeFilter"
# 雜湊函式的個數，預設為6，可以自行修改
BLOOMFILTER_HASH_NUMBER = 6
# BloomFilter的bit引數，預設30，佔用128MB空間，去重量級1億
BLOOMFILTER_BIT = 30複製程式碼DUPEFILTERCLASS 是去重類，如果要使用 BloomFilter需要將 DUPEFILTERCLASS 修改為該包的去重類。
BLOOMFILTERHASHNUMBER 是 BloomFilter 使用的雜湊函式的個數，預設為 6，可以根據去重量級自行修改。
BLOOMFILTERBIT 即前文所介紹的 BloomFilter 類的 bit 引數，它決定了位陣列的位數，如果 BLOOMFILTERBIT 為 30，那麼位陣列位數為 2 的 30次方，將佔用 Redis 128MB 的儲存空間，去重量級在 1 億左右，即對應爬取量級 1 億左右。如果爬取量級在 10億、20 億甚至 100 億，請務必將此引數對應調高。
測試
Spider 檔案:
from scrapy import Request,Spider
class 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,26,495018)}複製程式碼可以看到最後統計的第一行的結果：
'bloomfilter/filtered': 10,複製程式碼這就是 BloomFilter 過濾後的統計結果，可以看到它的過濾個數為 10 個，也就是它成功將重複的 10 個 Reqeust 識別出來了，測試通過。
原理
本文偏應用，難以描述的原理，最後說。一個很長的二進位制向量和一個對映函式。
參考資料
1、https://zhuanlan.zhihu.com/p/378474802、https://www.youtube.com/watch?v=uC6Q5m0SSQ03、《python3網路爬蟲開發實戰》崔慶才4、https://www.jianshu.com/p/f57187e2b5b9