流数据中等量分箱的分位数估算方法（Python实现）

需求概述 #

需要处理无法全部加载到内存的超大数值可迭代对象，将其划分为N个等量分箱（每个分箱包含的数据点数量大致相等，分箱宽度无需均匀），在不存储全部数据的前提下，获取每个分箱的上下边界值。

可行解决方案 #

场景1：数值范围已知且离散（如示例中的0-20） #

这种场景下，通过统计每个数值的出现频率，再计算累积频率确定分箱边界，内存占用仅与不同数值的数量相关，效率极高。

实现代码

import random
random.seed(0)

class BinBoundaryTracker:
    def __init__(self, n_bins=5, min_val=None, max_val=None):
        self.n_bins = n_bins
        self.value_counts = {}
        self.total_count = 0
        # 已知数值范围时提前初始化计数，减少动态判断
        if min_val is not None and max_val is not None:
            for val in range(min_val, max_val + 1):
                self.value_counts[val] = 0

    def update(self, cur_number):
        self.total_count += 1
        self.value_counts[cur_number] = self.value_counts.get(cur_number, 0) + 1

    def get_bin_boundaries(self):
        if self.total_count == 0:
            return []
        
        # 按数值排序后计算累积计数
        sorted_values = sorted(self.value_counts.keys())
        cumulative = 0
        boundaries = []
        target_per_bin = self.total_count / self.n_bins
        current_target = target_per_bin

        boundaries.append(sorted_values[0])  # 第一个边界为最小值
        for val in sorted_values:
            cumulative += self.value_counts[val]
            # 累积计数达成分箱目标时记录边界
            while cumulative >= current_target and len(boundaries) < self.n_bins:
                boundaries.append(val)
                current_target += target_per_bin
        boundaries.append(sorted_values[-1])  # 最后一个边界为最大值
        
        # 处理数值重复过多导致的边界数量不足问题
        while len(boundaries) < self.n_bins + 1:
            boundaries.append(sorted_values[-1])
        
        return boundaries

# 初始化跟踪器，指定数值范围0-20
tracker = BinBoundaryTracker(n_bins=10, min_val=0, max_val=20)
count0 = 0
sum0 = 0
running_min0 = None
running_max0 = None

for i in range(100000000):
    cur_number = random.randint(0, 20)
    count0 += 1
    sum0 += cur_number
    running_mean0 = sum0 / count0
    if running_min0 is None or running_min0 > cur_number:
        running_min0 = cur_number
    if running_max0 is None or running_max0 < cur_number:
        running_max0 = cur_number
    
    tracker.update(cur_number)
    # 可按需获取边界，例如每处理1000万条数据后
    if i % 10000000 == 0:
        running_bin_boundaries = tracker.get_bin_boundaries()
        print(f"已处理{i+1}条数据，当前分箱边界：{running_bin_boundaries}")

# 全部处理完成后获取最终边界
final_boundaries = tracker.get_bin_boundaries()
print(f"最终分箱边界：{final_boundaries}")

原理说明

• 用字典统计每个数值的出现次数，内存占用仅为不同数值的数量（示例中仅21个键）
• 每次更新仅增加对应数值的计数，时间复杂度O(1)
• 计算边界时，对数值排序后累加计数，累积数达成分箱目标时记录边界，最终得到每个分箱的上下限

场景2：数值范围未知或连续 #

若数值为连续型或范围未知，需使用流式分位数估计算法，这类算法可在O(log n)的空间复杂度下，近似估计指定分位数的值，进而得到分箱边界。常用算法包括：

• Greenwald-Khanna算法：维护有序样本列表，每个样本附带误差范围，通过合并误差重叠的样本控制内存占用
• t-digest算法：通过聚类相似样本点，用少量聚类中心近似数据分布，适合处理偏态数据

基于t-digest的实现示例

借助第三方库tdigest快速实现：

from tdigest import TDigest
import random

random.seed(0)
digest = TDigest()
n_bins = 10

for i in range(100000000):
    cur_number = random.randint(0, 20)
    digest.update(cur_number)

# 计算分位数得到分箱边界
boundaries = [digest.percentile(p) for p in range(0, 101, 10)]
print(f"最终分箱边界：{boundaries}")

说明

• t-digest算法仅需存储数百个聚类中心，即可在极低内存下高精度估计分位数，尤其适配偏态分布数据
• 若需手动实现流式算法，可参考Greenwald-Khanna论文逻辑，但第三方库已做优化，更适合生产环境

对比传统方法的优势 #

• 无需加载全部数据到内存，仅维护少量统计信息，适配数千万甚至数亿级数据
• 支持实时/滚动计算，可在数据处理过程中随时获取当前分箱边界估计值
• 离散场景下能得到精确边界，连续场景下能得到高精度近似边界

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