使用庫pybloom_live

from pybloom_live import ScalableBloomFilter,BloomFilter

# 可自動伸縮的布隆過濾器
bloom = ScalableBloomFilter(initial_capacity=100,error_rate=0.001)

# 新增內容
bloom.add('daqi')
print('daqi'in bloom)

# 定長的布隆過濾器
bloom1 = BloomFilter(capacity=10000)
bloom1.add('daqi')
print('daqi'in bloom1)

手動實現一個簡單的布隆過濾器

使用bitarray實現，將初始陣列置為0，根據hash計算出節點置為1，同時寫了一個生成隨機碼的函式用於測試。

import random
import mmh3
from bitarray import bitarray
import os.path
import re


# bitarray長度
BIT_SIZE = 50000

class BloomFilter():

  def __init__(self):
    bit_array = bitarray(BIT_SIZE)
    bit_array.setall(0)
    self.bit_array = bit_array
    self.bit_size = self.length()


  def get_points(self,url):
    """
    生成需要插入的位置
    :param url:
    :return:節點的列表
    """
    point_list = []
    for i in range(7):
      point = mmh3.hash(url,30+i) % self.bit_size
      point_list.append(point)
    return point_list

  def add(self,url):
    """
    新增url到bitarray中
    :param url:
    :return:
    """
    res = self.bitarray_expand()
    points = self.get_points(url)
    try:
      for point in points:
        self.bit_array[point] = 1
      return '註冊完成！'
    except Exception as e:
      return e

  def contains(self,url):
    """
    驗證url是否存在
    :param url:
    :return:True or False
    """
    points = self.get_points(url)
    # 在bitarray中查詢對應的點，如果有一個點值為0就說明該url不存在
    for p in points:
      if self.bit_array[p] == 0:
        return False
    return True


  def count(self):
    """
    獲取bitarrray中使用的節點數
    :return: bitarray長度
    """
    return self.bit_array.count()


  def length(self):
    """
    獲取bitarray的長度
    :return:bitarray的長度
    """
    return len(self.bit_array)


  def bitarray_expand(self):
    """
    擴充bitarray長度
    :return:bitarray的長度或使用率，布隆過濾器的bitarray的使用最好不要超過50%,這樣誤判率低一些
    """
    isusespace = round(int(self.count()) / int(self.length()),4)
    if 0.50 < isusespace:
      # 新建bitarray
      expand_bitarray = bitarray(BIT_SIZE)
      expand_bitarray.setall(0)
      # 增加新建的bitarray
      self.bit_array = self.bit_array + expand_bitarray
      self.bit_size = self.length()
      return self.bit_size
    else:
      return f'長度尚可,{round(isusespace * 100,2)}%'



def get_captcha():
  """
  生成用於測試的隨機碼
  :return:
  """
  seed = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
  captcha = ""
  for i in range(10):
    captcha += random.choice(seed)
  print(captcha)
  return captcha


if __name__ == '__main__':
  bloom = BloomFilter()
  for i in range(100000):
    bloom.add(f'www.{get_captcha()}.com')
    print(bloom.length())
    print(bloom.count())
  print(bloom.count())

到此這篇關於python使用布隆過濾器的實現示例的文章就介紹到這了,更多相關python 布隆過濾器內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！