4、随机降采样：简单随机采样、分层随机采样、蓄水池采样算法

聊到降采样，很多人第一反应就是「每隔N条取一条」。嗯，这确实简单，但有时候会出大问题。我遇到过几次，因为数据本身有周期性，固定间隔采样恰好踩到了波峰或波谷，结果分析出来的结论完全跑偏。

所以，随机降采样才是更靠谱的选择。今天咱们就聊聊三种最常用的随机采样方法：简单随机、分层随机，还有那个面试常考的蓄水池算法。

4.1 简单随机采样

说白了，就是从N条日志里，等概率地抽出M条。每条日志被抽中的概率都是 M/N。

实现起来也简单，Python里一行代码搞定：

import random

# 假设 logs 是全部日志列表，sample_size 是目标采样数
sampled_logs = random.sample(logs, sample_size)

但这里有个坑——如果日志量特别大，比如一天几亿条，random.sample 会把所有数据加载到内存里。我当年在某个大促场景下就吃过这个亏，内存直接爆了。

那怎么办？可以用「概率采样」的思路：遍历每条日志，以 p = M/N 的概率决定是否保留。这样内存占用是 O(1) 的。

import random

def simple_random_sample(stream, sample_size, total_count):
    p = sample_size / total_count
    result = []
    for log in stream:
        if random.random() < p:
            result.append(log)
            if len(result) >= sample_size:
                break
    return result

不过要注意，这种方法得到的样本数量是近似值，不是精确的 M 条。如果要求精确数量，就得用蓄水池算法了，后面会讲。

4.2 分层随机采样

简单随机采样有个问题：如果日志里某些类别的数据很少，可能恰好一条都没抽到。比如错误日志只占 0.1%，随机采样后可能一条错误都没留下。

分层随机采样就是来解决这个问题的。先把日志按某个维度（比如日志级别、服务名、错误码）分成若干层，然后在每层内部独立采样。

举个例子，假设我们有三种日志级别：

日志级别	原始数量	占比	采样数量
ERROR	1,000	0.1%	500
WARN	10,000	1%	500
INFO	989,000	98.9%	500

你看，虽然 ERROR 日志很少，但我们保证它被充分采样。这样后续分析时，错误模式就不会被淹没掉。

def stratified_sample(logs, strata_key, sample_size_per_stratum):
    # 按层级分组
    strata = {}
    for log in logs:
        key = log[strata_key]
        if key not in strata:
            strata[key] = []
        strata[key].append(log)
    
    # 每层独立采样
    result = []
    for key, group in strata.items():
        if len(group) <= sample_size_per_stratum:
            result.extend(group)
        else:
            result.extend(random.sample(group, sample_size_per_stratum))
    return result

4.3 蓄水池采样算法

这个算法有点意思。它的应用场景是：数据流长度未知，但你需要从中均匀抽取 k 个样本。

比如实时日志流，你不知道今天会来多少条日志，但你想保留 1000 条作为今天的采样样本。这时候蓄水池算法就派上用场了。

算法思路其实很巧妙：

1. 先把前 k 条日志放进「蓄水池」（一个大小为 k 的数组）
2. 从第 k+1 条开始，每条日志以 k/i 的概率替换池中的某一条

为什么这样能保证均匀？数学证明我就不展开了，你记住结论就行：最终池中每条日志被选中的概率都是 k/N，其中 N 是总日志数。

import random

def reservoir_sample(stream, k):
    """
    蓄水池采样：从未知长度的流中均匀采样 k 条
    """
    reservoir = []
    
    for i, log in enumerate(stream):
        if i < k:
            # 前 k 条直接放入
            reservoir.append(log)
        else:
            # 第 i 条（i从0开始）以 k/(i+1) 的概率替换
            j = random.randint(0, i)
            if j < k:
                reservoir[j] = log
    
    return reservoir

这段代码看起来简单，但背后有大学问。我当年第一次看到这个算法时，觉得「这不就是个随机替换吗？」后来仔细推敲才发现，它保证了每个元素被选中的概率完全相等。

三种方法怎么选？

我整理了一个对比表，方便你快速决策：

方法	适用场景	内存	样本数量	是否保类别
简单随机采样	数据量小，可全部加载	O(N)	精确	否
分层随机采样	需要保留稀有类别	O(N)	精确	是
蓄水池采样	数据流、未知长度	O(k)	精确	否

你想想看，如果日志量不大，直接简单随机采样就行。如果日志里有重要的「小类别」，分层采样是首选。如果日志是源源不断的流式数据，那就用蓄水池算法。

嗯，这三种方法基本覆盖了日常工作中 90% 的随机降采样需求。下一节咱们聊聊更高级的降采样技术，到时候会用到这些基础方法作为构建块。

---

公众号:蓝海资料掘金营，微信deep3321