大数据下的高级算法：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迪斯尼