0


大数据下的高级算法:hyperloglog,统计海量数据下不同元素的个数

如果你被面试到redis,通常对方会问你用过什么数据结构,如果你说使用过hyperloglog那绝对是个加分项,因为对方知道你正在处理基于海量数据和高并发下的问题。上一节我们使用min-count-sketch 算法统计了海量数据下给定元素的重复次数,而hyperloglog正好反过来,它统计整个数据集中不同元素的个数。

在传统应用场景下,实现这个目标的常用方法是使用哈希表,我们遍历一次所有元素,然后看看哈希表是否已经有了对应元素,最后再遍历一次哈希表就能得到不同元素的个数。这种做法存在问题是,在海量数据情况下,哈希表很可能要存储大量数据,特别是重复元素比较少时,哈希表要占用的内存就很大,而且数据元素是复杂结构体的情况下,占用的内存将会进一步加大。

跟上一节类似,大数据场景下算法都遵循一个套路,那就是拿准确度换取内存节省,内存省的越多,准确度就会相应下降,通常情况下算法会把原来用几十个G的内存降到几M,同时准确度控制在99%左右,在海量数据情形下,这种准确率是完全能接受的。下面我们看看具体的算法流程。

我们先思考一个问题,假设有一个公平的硬币,也就是出现人头和字的概率各是0.5,假设我们抛硬币N次,结果发现人头出现了100次,那么请问N的值是多少?由于出现人头的概率是0.5,因此我们可以估算N的取值是100 / 0.5 = 200。HyperLogLog第一步设计就出于这个想法。假设我们有一个含有n个元素的数据集,其中包含k个不同元素,我们要想实现前面提到的“抛硬币”效果,那么就可以用一个哈希函数,其输出结果是一个长度为L的二进制字符串,也就是字符串包含L个字符,字符为0或者1。如果L的值足够大,那么我们就能把不同的输入数据哈希到不同的输出结果,如果数据集中包含k个不同数据,那么输出结果就会有k个不同的值。

上面这种做法问题在于,我们还是要把所有结果存储下来,如果L的值比集合中元素所需存储空间还要大的话,那么算法反而需要更大空间,因此我们需要对其进行优化,下面我们介绍一种优化方法叫概率计数,它的原理为:在获得哈希结果后,我们从右往左计算0的个数,然后把结果加1,我们用p来标记这个结果。例如"1100",从右往左数有2个0,因此p值就是2+1=3,对于“0111”,从第一个位置开始就是1,因此对应p值就是0+1=1,对于“0000”,从右往左数有4个0,因此对应p值就是4+1=5.我们把所有元素都计算出哈希值,然后计算出最大的那个p值,我们记作p(max),那么不同元素个数的估算为2^p(max),需要预先提醒的是这个方法所得结果并不准确,我们后面会不断改进它的精确度。

我们看看算法的基本逻辑。假设哈希函数的计算结果足够随机,如果一个包含n个元素的数组,其中有k个不同元素,那么我们预计在这k个不同元素中,有一半其哈希结果最右边的元素取值0,另外一半最右边元素取值1.下面我们针对取值为0的那一部分(包含k/2个元素)进行下一步处理。在这部分元素中,其哈希结果的倒数第二个元素取值为0和取值为1的各占一半,也就是每部分元素个数为k/4,也就说哈希结果最右边两个元素都取值为0的元素数量为k/4,以此类推哈希结果最右边i个元素都取值为0的元素个数为k/(2^i),由此根据我们前面预测抛硬币次数的逻辑一样,如果从右边起0最多的个数为p_max,那么我们就可以估算不同元素的个数为2 ^ p_max个。这里我们需要再次强调这只是一个概率估算,其结果并不足够准确,我们看看其代码实现;

import hashlib
import random

defconvert_hash_to_binary(hash_hex):# 将哈希字符串转换为包含0和1的字符串returnbin(int(hash_hex,16))defnum_trailing_zeros(hash_bin):# 从右到左统计0的个数一直到遇见1停止
    reverse = hash_bin[::-1]
    count =0for i inrange(len(reverse)):if reverse[i]=='0':
            count +=1else:breakreturn count +1defprobability_counting(array):# 估算给定数组中不同元素的个数
    p_max =0for a in array:
        hash_str = hashlib.sha256(str(a).encode()).hexdigest()
        bin_str = convert_hash_to_binary(hash_str)
        p = num_trailing_zeros(bin_str)if p > p_max:
            p_max = p
            # print(f"str: {bin_str} with trailing zeros: {p_max}")return2** p_max

ELEMENT_COUNT =100000# 随机创建给定个取值位于(0, 10000)之间的整数defgenerate_random_array():
    count =0
    array =[]
    diff_count =0
    diff_map ={}
    same_count =0while count < ELEMENT_COUNT:
        num = random.randint(0,10000)
        array.append(num)if num notin diff_map:
            diff_map[num]=True
            diff_count +=1else:
            same_count +=1
        count +=1# print(f"diff count: {diff_count}, same count : {same_count}")return array, diff_count

array, differ_count = generate_random_array()

probability_count = probability_counting(array)print(f"different elements cont: {differ_count}")print(f"probability count: {probability_count}")

在上面代码中我们生成100000个元素,每个元素取值在(0,10000)之间,然后记录不同元素的个数,最后使用概率估算来预测一下不同元素的个数,代码运行结果如下:

different elements cont:10000
probability count:8192

可以看到估算的结果并不准确,但这种想法却是hyperloglog算法的起始。下面我们对上面描述的算法进行改进,这次改进后的算法叫随机平均,它的做法是取出哈希结果最右边的b个比特位,根据其结果分别放置到m=2^b个“桶”中,然后对每个“桶”中的元素,依次计算他们对应的p_max,然后把这些p_max加总后再除以m, 也就是A = p_1_max + p_2_max +… + p_m_max,其中p_i_max表示第i个桶对应的p_max,最后估算值就是m * (2 ^ A),我们看看代码实现:

defget_bits_val(bin_str, b):# 例如 1001011, 取前边4个比特位,1001,其对应数值就是9
    b_bit_str = bin_str[0:b]
    b_bit_val =int(b_bit_str,2)return b_bit_val

deffirst_bits(h, b):# h 将h对应的哈希值转换为只包含0,1的二进制字符串,然后去最右边b个比特位,并计算他们形成的数值
    bin_str = convert_hash_to_binary(h)
    bit_val = get_bits_val(bin_str, b)return bit_val

# 下面代码打印结果为9#print("fist 4  bits of 1001111: ", get_bits_val("1001011", 4))

b =14# 桶的个数为2 ^ 14 
bucket_map ={}defstochastic_average(array):
    bucket_count =0for a in array:
        hash_str = hashlib.sha256(str(a).encode()).hexdigest()
        bin_str = convert_hash_to_binary(hash_str)
        p = num_trailing_zeros(bin_str)# 将哈希结果转换为二进制,取最左边b个比特值计算当前元素哈希值所在的桶
        bucket = first_bits(hash_str, b)if bucket in bucket_map:#记录每个桶元素从右边算起0做多的个数if p > bucket_map[bucket]:
                bucket_map[bucket]= p
        else:
            bucket_map[bucket]= p
            bucket_count +=1

    p_max_sum =0
    m =2** b

    for i inrange(m):if i in bucket_map:
            p_max_sum += bucket_map[i]

    avg = p_max_sum / bucket_count 
    return bucket_count *(2** avg)

stochastic_count = stochastic_average(array)print(f"result of stochastic_average is {stochastic_count}")

上面代码运行后结果如下:

different elements cont:10001
result of stochastic_average is25158.04266464522

说实话看起来也没有改进到那里去。因此我们再做进一步改进,完成这一步算法就叫做loglog,这次改进主要是在上一步结果的基础上乘以一个参数,这个参数的值跟上面算法中的m,也就是桶的数量有关,这个参数为:
a(m) = 0.39701 - (2*(pi^2) + (ln(2)) ^ 2) / 28m

如果m的值大于64, 那么a(m)就可以直接取值为0.39701,我们把前面的结果乘以这个数值所得结果为9887.99约等于9888,可以看到已经非常接近了。从数理统计上看,在乘以参数a(m)后,错误率在1/sqrt(m) ,当b=14时,这个值在1%左右。从当前算法看内存占据主要在“桶”上,如果我们设置一个桶的大小在8字节,那么桶的数量设置在2 ^ 14 时,内存需要130kb左右,而且算法不管你要处理的数据量是多大,错误率都可以保持不变。

此外如果我们能确认数据集中不同元素个数最多不超过k-max, 那么我们只需要哈希函数给出的结果只要log(k-max)个比特位即可(例如32对应二进制只需要5个比特位来表示),此外由于每个桶用于存储哈希结果转换为二进制后从右到左数起0的个数,因此一个桶需要的内存大小就是log(log(k-max))个比特位,这里可能有点绕,具体来说,假设假设哈希结果转换为二进制形式后最多不超过64个比特位,这意味着从右到左数起0的个数不超过64个,因此计数使用6个比特位即可,因为2 ^ 6 = 64,这就是loglog算法名称的由来。由此如果我们确定数据中不同元素的个数不超过 k-max = 2 ^ 64, 如果桶的数量设置为2 ^ 14, 那么总共需要的内存就是(2 ^ 14) * log(log(2 ^ 64)) 约等于12kb,

最后我们再次改进上面的LogLog算法得到SuperLogLog算法,改进的地方在于前面随机平均是将每个桶的最大值加总后求平均,这里我们使用所谓的调和平均来计算,其公式为:
E_bucket = m / (2 ^ -p_1_max + 2 ^ -p_2_max +… + 2 ^ -p_m_max)
然后用上面的计算结果乘以参数a(m)以及桶的数量m,不过这里参数a(m)跟上面有所不同,它会根据桶的数量进行不同取值,基本情况如下:
a(16)=0.673, a(32)=0.697, a(64) = 0.709, a(m) = 0.7213/(1+1.079/m) m >= 128
也就是如果桶的数量不超过16,那么a(m)取值0.673, 大于16但小于32则取值0.697,以此类推,我们看看代码实现:


bucket_map ={}defget_alpha(m):if m <=16:return0.673elif16< m <=32:return0.697elif32< m <=64:return0.709else:return0.7213/(1+1.079/ m)

b =11#不同取值对结果影响较大,原因在于我们的实验数据没能达到"海量"标准defhyper_log_log(array):
    bucket_count =0for a in array:
        hash_str = hashlib.sha256(str(a).encode()).hexdigest()
        bin_str = convert_hash_to_binary(hash_str)
        p = num_trailing_zeros(bin_str)# 将哈希结果转换为二进制,取最左边b个比特值计算当前元素哈希值所在的桶
        bucket = first_bits(hash_str, b)if bucket in bucket_map:# 记录每个桶元素从右边算起0做多的个数if p > bucket_map[bucket]:
                bucket_map[bucket]= p
        else:
            bucket_map[bucket]= p
            bucket_count +=1

    compute_sum =0# 计算调和平均数for key in bucket_map:
        compute_sum +=(2**(-1* bucket_map[key]))

    harmonic_avg = bucket_count / compute_sum
    result = get_alpha(bucket_count)* bucket_count * harmonic_avg
    '''
    最后还需要根据结果做一些调整,这些调整主要基于比较复杂的数理统计推导,我们暂时忽略
    '''if result <5* bucket_count /2:print(f"small correction")if result >2**32/30:
        result =-2**32* math.log(1- result /2**32)return result

print(f"result of hyperloglog {hyper_log_log(array)}")

上面代码运行后所得结果如下:

different elements cont:9999
alpha :0.7182725932495458
result of hyperloglog 9945.058986524531

从代码实验中我发现,b的取值不同对结果影响较大,我个人认为原因在于代码实验所使用的数据量达不到"海量“的要求,毕竟个人电脑的内存和算力非常有限。更多内容请在b站搜索coding迪斯尼

标签: 大数据 算法

本文转载自: https://blog.csdn.net/tyler_download/article/details/128652032
版权归原作者 tyler_download 所有, 如有侵权,请联系我们删除。

“大数据下的高级算法:hyperloglog,统计海量数据下不同元素的个数”的评论:

还没有评论