← 中文资源

简体中文版本

用固定直方图估算分位数

把观测值落入固定桶并估算 p50、p90、p99,无需保存全部样本。

阅读英文原文

功能说明

计算精确分位数通常需要保存所有样本并排序,内存随样本数线性增长。这段代码改用固定范围的直方图:把每个观测值映射到若干等宽桶,只保留每个桶的计数,再用桶内线性插值估算分位数。

它适合延迟指标、请求大小、评分分布等流式监控场景。内存只取决于桶数量,与样本量无关,因此可以长时间运行而不会膨胀。

关键代码

NewHistogram 校验桶数和上下界,并分配计数切片。Observe 先把超出范围的值截断到边界,再按 (value-min)/width 计算桶下标;当值恰好等于最大值时,下标会等于桶数,需要减一。

Percentile 用 ceil(p/100*total) 计算目标排名,然后累加桶计数,找到包含该排名的桶。桶内假设均匀分布,用 (rank-seen)/count 估算位置,再换算回数值。

使用注意

估算误差最大可达一个桶宽,桶越少误差越大;对长尾分布,极端值会被压缩到最后一个桶,p99 可能明显失真。上下界之外的观测会被截断,不能用于判断真实最大值。

直方图不是并发安全的,多 goroutine 写入需要加锁或使用原子操作。桶边界应覆盖业务真实范围,并随数据变化定期调整。需要精确分位数或高精度尾延迟时,应使用 t-digest、DDSketch 或保存原始样本。

验证方式

示例把 10 到 90 以及 1000 放入 0 到 1000 的十个桶,输出约 p50=56、p90=100、p99=1000。代码断言三个分位数严格递增且 p99 等于 1000。可以减少桶数到 2,观察估算误差变大,从而理解精度与内存的权衡。

代码示例

package main

import (
	"fmt"
	"math"
)

type Histogram struct {
	min    float64
	max    float64
	counts []int
	total  int
}

func NewHistogram(minimum, maximum float64, buckets int) *Histogram {
	if buckets <= 0 || maximum <= minimum {
		panic("invalid histogram bounds")
	}
	return &Histogram{
		min:    minimum,
		max:    maximum,
		counts: make([]int, buckets),
	}
}

func (histogram *Histogram) Observe(value float64) {
	if value < histogram.min {
		value = histogram.min
	}
	if value > histogram.max {
		value = histogram.max
	}
	width := (histogram.max - histogram.min) / float64(len(histogram.counts))
	index := int((value - histogram.min) / width)
	if index == len(histogram.counts) {
		index--
	}
	histogram.counts[index]++
	histogram.total++
}

func (histogram *Histogram) Percentile(percentile float64) float64 {
	if histogram.total == 0 {
		return math.NaN()
	}
	rank := int(math.Ceil(percentile / 100 * float64(histogram.total)))
	if rank < 1 {
		rank = 1
	}
	seen := 0
	width := (histogram.max - histogram.min) / float64(len(histogram.counts))
	for index, count := range histogram.counts {
		if seen+count >= rank {
			within := float64(rank-seen) / float64(count)
			return histogram.min + (float64(index)+within)*width
		}
		seen += count
	}
	return histogram.max
}

func main() {
	histogram := NewHistogram(0, 1000, 10)
	for _, value := range []float64{10, 20, 30, 40, 50, 60, 70, 80, 90, 1000} {
		histogram.Observe(value)
	}
	p50 := histogram.Percentile(50)
	p90 := histogram.Percentile(90)
	p99 := histogram.Percentile(99)
	fmt.Printf("p50=%.0f p90=%.0f p99=%.0f\n", p50, p90, p99)
	if !(p50 < p90 && p90 < p99) || p99 != 1000 {
		panic("unexpected percentile estimates")
	}
}