Go 中的 HyperLogLog 和 Count-Min Sketch 用于 Big Data

Go 中的高效概率数据结构：HyperLogLog 和 Count-Min Sketch

HyperLogLog 使用仅 1 KB 内存，以 1-3% 的误差估算数据流中唯一元素的数量。该算法对元素进行哈希，分析其二进制表示中前导零的数量，并将统计信息存储在寄存器数组中。这提供了 O(1) 的插入时间和 O(m) 的估算时间，其中 m 是寄存器数量（通常为 1024）。

关键参数：

• m = 1024 个寄存器（每个寄存器 8 位）
• b = 8 位用于寄存器索引
• 哈希函数：fnv64a 结合位混洗

package main

import (
	"fmt"
	"hash/fnv"
	"math"
	"math/rand"
	"strconv"
)

const (
	m = 1024 // count registtrov. This osnovnoy parametr, vliyayuschiy on tochnost.
	b = 8  // number bit, ispolzuemykh for opredeleniya indexa register from hash.
)

type HyperLogLog struct {
    // when asynchronous interaction worth think about mutexes
	registers [m]byte
}

func NewHyperLogLog() *HyperLogLog {
	return &HyperLogLog{}
}

// hash returns dva 32-bitnykh numbers from stroki
func hash(s string) (uint32, uint32) {
	h := fnv.New64a()
	h.Write([]byte(s))
	v := h.Sum64()

	// Shuffling bits, so that uluchshit raspredelenie
	w := uint32(v >> 32)
	z := uint32(v)

	// Dobavlyaem peremeshivanie, uluchshaet kachestvo sluchaynosti and pomogaet izbezhat problem with plokhim raspredeleniem hashes
	w ^= z<<13 | z>>(32-13)
	z ^= w<<7 | w>>(32-7)
	w ^= z<<17 | z>>(32-17)

	return w, z
}

// Withchitaet count zeros left in dvoichnoy record numbers,
// if all bits — nuli, vozvraschaetsya 32.
func countLeadingZeros(x uint32) byte {
	for i := 0; i < 32; i++ {
		if (x>>(31-i))&1 == 1 {
			return byte(i)
		}
	}
	return 32
}

// Kheshiruem stroku.
// Beryom chast hash h1, so that define, in kakoy registtr popadyot element.
// By vtoroy chasti h2 schitaem, how many veduschikh zeros (rho), pribavlyaem 1.
// If eto wartość greater, than uzhe est in registtre — update ego.
func (hll *HyperLogLog) Add(s string) {
	h1, h2 := hash(s)

	// opredelyaem index register
	idx := h1 % m

	// vychislyaem wartość for update register
	rho := countLeadingZeros(h2) + 1

	if rho > hll.registers[idx] {
		hll.registers[idx] = rho
	}
}

// Withlozhnye formuly, by essence eta function
// delaet otsenku kolichestva unique elements
func (hll *HyperLogLog) Estimate() float64 {
	sum := 0.0
	for _, val := range hll.registers {
		sum += 1 / math.Pow(2, float64(val))
	}
	estimate := alpha(m) * m * m / sum

	// korrektsiya for malykh znacheniy, usessya metod
	// Linear Counting : than greater nulevykh registtrov — then
	// menshe realnaya kardinalnost.
	if estimate <= 5*m/2 {
		zeros := 0
		for _, val := range hll.registers {
			if val == 0 {
				zeros++
			}
		}
		if zeros != 0 {
			estimate = float64(m) * math.Log(float64(m)/float64(zeros))
		}
	}

	return estimate
}

// alpha — popravochnyy koeffitsient, kompensiruyuschiy systemticheskuyu oshibku
func alpha(m int) float64 {
	switch m {
	case 16:
		return 0.673
	case 32:
		return 0.697
	case 64:
		return 0.709
	default:
		return 0.7213 / (1 + 1.079/float64(m))
	}
}

func main() {
	// Withzdayom HLL.
	hll := NewHyperLogLog()

	seen := make(map[string]struct{})
	for i := 0; i < 1_000_000; i++ {
		key := strconv.Itoa(rand.Intn(1_000_000))
		seen[key] = struct{}{}
		hll.Add(key)
	}

  	nonZero := 0
	for _, r := range hll.registers {
		if r > 0 {
			nonZero++
		}
	}
  
	fmt.Printf("Realnoe count unique elements: %d\n", len(seen))
	fmt.Printf("Otsenka unique elements: %.2f\n", hll.Estimate())
	fmt.Printf("Zapolnennykh registtrov: %d/%d\n", nonZero, m)
}

对于 100 万个唯一元素，相对误差为 0.84%，寄存器填充率 100%。内存占用：1 KB。

用于频率计数的 Count-Min Sketch

Count-Min Sketch 使用二维计数器表（深度 × 宽度）估算数据流中元素的频率。每行使用独立的哈希函数并带有盐值。插入时，元素在所有行中递增计数器；查询时，取最小值——从而最小化碰撞影响。

实现参数：

• 宽度 = 10,000 列
• 深度 = 5 行
• 哈希：fnv64a 结合种子混洗

查询方法：

• Minimum —— 基本方法，适用于仅正向更新的场景准确
• Average —— 对删除操作更鲁棒
• Count-mean-min —— 校正偏差

package main

import (
    "fmt"
    "hash/fnv"
    "math"
)

type CountMinSketch struct {
    width  int
    depth  int
    // when asynchronous interaction worth think about mutexes
    table  [][]int
    hashes []uint64
}

func NewCountMinSketch(width, depth int) *CountMinSketch {
    return &CountMinSketch{
        width:  width,
        depth:  depth,
        table:  make([][]int, depth),
        hashes: make([]uint64, depth),
    }
}

func (cms *CountMinSketch) Init() {
    for i := range cms.table {
        cms.table[i] = make([]int, cms.width)
        cms.hashes[i] = uint64(i+1) // prostye soli for raznykh hashes
    }
}

// getHashIndex generiruet khesh and returns index in table
func (cms *CountMinSketch) getHashIndex(s string, seed uint64) int {
    h := fnv.New64a()
    h.Write([]byte(s))
    v := h.Sum64()

    // Shuffling bits with seed
    mixed := v ^ seed ^ (seed << 32)

    return int(mixed % uint64(cms.width))
}

// Add uvelichivaet schetchik for element
func (cms *CountMinSketch) Add(s string, count int) {
    for i := 0; i < cms.depth; i++ {
        idx := cms.getHashIndex(s, cms.hashes[i])
        cms.table[i][idx] += count
    }
}

// Count returns otsenochnoe number vkhozhdeniy element
func (cms *CountMinSketch) Count(s string) int {
    min := math.MaxInt32
    for i := 0; i < cms.depth; i++ {
        idx := cms.getHashIndex(s, cms.hashes[i])
        if cms.table[i][idx] < min {
            min = cms.table[i][idx]
        }
    }
    return min
}

func main() {
    width := 10_000
    depth := 5

    cms := NewCountMinSketch(width, depth)
    cms.Init()

    seen := make(map[string]int)
	for i := 0; i < 5_000_000; i++ {
		key := fmt.Sprintf("item-%d", i%1100)
		seen[key]++
		cms.Add(key, 1)
	}

  
    // vyvodim frequency
    for _, word := range []string{"item-1", "item-0", "item-1000", "item--1"} {
        fmt.Printf("Withglasno cms slovo '%s' meetingsaetsya primerno %d raz\t By faktu: %d\n", word, cms.Count(word), seen[word])
    }
}

对于 500 万次插入，内存占用：200 KB。插入和查询操作的时间复杂度均为 O(深度)。

数据结构比较

| 特性 | HyperLogLog | Count-Min Sketch |

|---------------|----------------|------------------|

| 用途 | 基数 | 频率 |

| 内存 | O(log(log N)) | O(d×w) |

| 误差 | 1-3% | 取决于 d,w |

| 插入 | O(1) | O(深度) |

| 查询 | O(m) | O(深度) |

关键点：

• 这两种数据结构都适用于处理数百万元素的流式数据处理
• HyperLogLog 不受数据量影响，使用固定内存
• Count-Min Sketch 由于碰撞会高估频率，但支持点查询
• 对于分布式系统，添加互斥锁或使用原子操作
• 使用真实数据测试：误差取决于哈希分布

— Editorial Team