python使用布隆过滤器如何实现 python使用布隆过滤器实现代码示例

python使用布隆过滤器如何实现？本篇文章小编给大家分享一下python使用布隆过滤器实现代码示例，小编觉得挺不错的，现在分享给大家供大家参考，有需要的小伙伴们可以来看看。

使用库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())