在Golang中合并Redis序列化的HyperLogLog集合（不依赖Redis命令）

我在Redis中存储了两个HyperLogLog集合，希望在Golang应用中完成合并操作，而非使用PFMERGE等Redis命令。目标是获取Redis中序列化的HyperLogLog数据，反序列化后合并集合，再使用合并后的结果。目前使用github.com/axiomhq/hyperloglog库进行HyperLogLog操作，但不清楚反序列化与合并这两个集合的最优实现方式，现有代码如下：

package main

import (
    "bytes"
    "context"
    "encoding/gob"
    "fmt"
    "log"

    "github.com/axiomhq/hyperloglog"
    "github.com/redis/go-redis/v9"
)

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })

    // Create HyperLogLog in Redis
    rdb.PFAdd(ctx, "hll1", "foo", "bar")
    rdb.PFAdd(ctx, "hll2", "baz", "qux")

    // Get serialized HyperLogLog sets from Redis
    data1, err := rdb.Get(ctx, "hll1").Bytes()
    if err != nil {
        log.Fatal("Error fetching hll1:", err)
    }
    data2, err := rdb.Get(ctx, "hll2").Bytes()
    if err != nil {
        log.Fatal("Error fetching hll2:", err)
    }

    /*  想要实现类似下面的逻辑，但不知道怎么让它工作
     Deserialize HyperLogLog sets
    hll1, err := deserializeHLL(data1)
    if err != nil {
        log.Fatal("Error deserializing hll1:", err)
    }
    hll2, err := deserializeHLL(data2)
    if err != nil {
        log.Fatal("Error deserializing hll2:", err)
    }

    // Merge the two sketches
    err = hll1.Merge(hll2)
    if err != nil {
        log.Fatal("Error merging HyperLogLog sets:", err)
    }

    // Get the estimated count
    fmt.Println("Estimated count after merge:", hll1.Estimate())
   */
}

核心问题是Redis的HyperLogLog二进制格式与axiomhq/hyperloglog库的序列化格式不兼容，需要手动解析Redis的HLL结构，转换成库对应的实例后再进行合并。

1. 解析Redis的HyperLogLog二进制格式 #

Redis的HLL二进制结构如下：

• 前3字节：固定标识HLL（0x48、0x4C、0x4C）
• 第4字节：版本号（当前为0）
• 第5字节：寄存器数量的对数（log2(m)，默认Redis创建的HLL为14，对应m=16384个寄存器）
• 后续字节：紧凑存储的寄存器数据，每个寄存器占6位，每8个寄存器占用6字节

下面是解析函数，将Redis返回的二进制数据转换为axiomhq/hyperloglog的实例：

import (
    "errors"
    "math/bits"

    "github.com/axiomhq/hyperloglog"
)

// parseRedisHLL 解析Redis序列化的HLL数据，转换为axiomhq的HyperLogLog实例
func parseRedisHLL(data []byte) (*hyperloglog.Sketch, error) {
    // 检查头部标识
    if len(data) < 5 || data[0] != 'H' || data[1] != 'L' || data[2] != 'L' || data[3] != 0 {
        return nil, errors.New("invalid Redis HLL format")
    }

    log2m := data[4]
    m := 1 << log2m // 寄存器数量
    expectedLen := 5 + (m*6+7)/8 // 计算预期的总字节数
    if len(data) != expectedLen {
        return nil, errors.New("Redis HLL data length mismatch")
    }

    // 初始化axiomhq的Sketch，精度对应log2m
    sketch := hyperloglog.New(uint8(log2m))
    registers := sketch.Registers() // 获取寄存器数组

    // 解析紧凑存储的6位寄存器
    var bitOffset int
    for i := 0; i < m; i++ {
        byteIdx := bitOffset / 8
        bitPos := bitOffset % 8

        // 读取6位数据
        var val uint8
        if bitPos <= 2 {
            // 6位都在当前字节或跨1个字节
            val = (data[byteIdx] >> (2 - bitPos)) & 0x3F
        } else {
            // 跨两个字节
            val = (data[byteIdx] << (bitPos - 2)) & 0x3F
            val |= data[byteIdx+1] >> (10 - bitPos)
        }

        registers[i] = val
        bitOffset += 6
    }

    return sketch, nil
}

2. 合并HyperLogLog集合 #

使用axiomhq/hyperloglog库的Merge方法即可合并两个Sketch实例，注意两个实例的精度必须一致（Redis默认创建的HLL精度都是14，所以无需额外处理）。

3. 完整实现代码 #

将解析和合并逻辑整合到原有代码中：

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "math/bits"

    "github.com/axiomhq/hyperloglog"
    "github.com/redis/go-redis/v9"
)

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })

    // 创建Redis中的HLL集合
    if err := rdb.PFAdd(ctx, "hll1", "foo", "bar").Err(); err != nil {
        log.Fatal("PFAdd hll1 failed:", err)
    }
    if err := rdb.PFAdd(ctx, "hll2", "baz", "qux").Err(); err != nil {
        log.Fatal("PFAdd hll2 failed:", err)
    }

    // 获取Redis中的HLL序列化数据
    data1, err := rdb.Get(ctx, "hll1").Bytes()
    if err != nil {
        log.Fatal("Error fetching hll1:", err)
    }
    data2, err := rdb.Get(ctx, "hll2").Bytes()
    if err != nil {
        log.Fatal("Error fetching hll2:", err)
    }

    // 反序列化Redis的HLL数据
    hll1, err := parseRedisHLL(data1)
    if err != nil {
        log.Fatal("Error deserializing hll1:", err)
    }
    hll2, err := parseRedisHLL(data2)
    if err != nil {
        log.Fatal("Error deserializing hll2:", err)
    }

    // 合并两个HLL集合
    if err := hll1.Merge(hll2); err != nil {
        log.Fatal("Error merging HyperLogLog sets:", err)
    }

    // 输出合并后的估算计数
    fmt.Println("Estimated count after merge:", hll1.Estimate())
}

// parseRedisHLL 解析Redis序列化的HLL数据，转换为axiomhq的HyperLogLog实例
func parseRedisHLL(data []byte) (*hyperloglog.Sketch, error) {
    // 检查头部标识
    if len(data) < 5 || data[0] != 'H' || data[1] != 'L' || data[2] != 'L' || data[3] != 0 {
        return nil, errors.New("invalid Redis HLL format")
    }

    log2m := data[4]
    m := 1 << log2m // 寄存器数量
    expectedLen := 5 + (m*6+7)/8 // 计算预期的总字节数
    if len(data) != expectedLen {
        return nil, errors.New("Redis HLL data length mismatch")
    }

    // 初始化axiomhq的Sketch，精度对应log2m
    sketch := hyperloglog.New(uint8(log2m))
    registers := sketch.Registers() // 获取寄存器数组

    // 解析紧凑存储的6位寄存器
    var bitOffset int
    for i := 0; i < m; i++ {
        byteIdx := bitOffset / 8
        bitPos := bitOffset % 8

        // 读取6位数据
        var val uint8
        if bitPos <= 2 {
            // 6位都在当前字节或跨1个字节
            val = (data[byteIdx] >> (2 - bitPos)) & 0x3F
        } else {
            // 跨两个字节
            val = (data[byteIdx] << (bitPos - 2)) & 0x3F
            val |= data[byteIdx+1] >> (10 - bitPos)
        }

        registers[i] = val
        bitOffset += 6
    }

    return sketch, nil
}

注意事项 #

• 确保两个HLL集合的精度一致，Redis默认创建的HLL精度为14，如果手动修改过Redis的HLL精度，需要调整parseRedisHLL中的处理逻辑。
• 如果需要将合并后的HLL重新存入Redis，可以将axiomhq的Sketch转换回Redis的二进制格式（需要实现对应的序列化函数），再用SET命令存入。

内容的提问来源于stack exchange，提问作者mohit nagpal