核心要点
场景：数据流总量 n 未知或无法一次载入内存，需等概率抽取 k 个样本。
算法：前 k 个直接入池；第 i 个元素（i>k）以 k/i 的概率替换池中随机一个位置。
正确性：任一元素最终留在池中的概率恒为 k/n，与位置无关。
复杂度：单遍扫描 O(n) 时间、O(k) 空间。
标准回答
一、先把结论讲清楚
可以先说使用场景：数据流很长、长度 n 事先不知道，或者没法全部放进内存，但还想等概率抽样。
一、k=1 时怎么做？
维护一个结果 res。遍历到第 i 个元素时，以 1/i 的概率把 res 替换成当前元素。这样第 1 个元素虽然一开始必选，但后面每一步都有可能被替换；
二、拆开机制和判断点
第 i 个元素虽然来得晚，但它入选概率就是 1/i。最后可以证明每个元素留下来的概率都是 1/n。
二、k 个样本怎么做？
前 k 个元素先直接放进池子。之后遍历到第 i 个元素时，以 k/i 的概率决定是否让它入池；如果入池，就随机替换池子里的一个位置。
三、补上例子、边界和取舍
最终每个元素被选中的概率都是 k/n。
三、复杂度是什么？
它只扫一遍数据，时间 O(n)，池子大小固定是 k，所以空间 O(k)。需要强调：这题的价值不在代码长，而在“流式、未知长度、等概率”这三个关键词。
import random
def reservoir_sample_one(stream):
"""k=1: 从数据流中等概率抽取 1 个元素"""
res = None
for i, x in enumerate(stream, start=1): # i 从 1 计数
if random.randint(1, i) == 1: # 以 1/i 概率替换
res = x
return res
def reservoir_sample_k(stream, k):
"""一般 k: 等概率抽取 k 个元素，每个被选概率 k/n"""
pool = []
for i, x in enumerate(stream, start=1):
if i <= k:
pool.append(x) # 前 k 个直接入池
else:
j = random.randint(1, i) # j 在 [1, i]
if j <= k: # 以 k/i 概率入池
pool[j - 1] = x # 替换池中随机位置
return pool
if __name__ == '__main__':
# 频率验证：每个元素被选概率应约为 k/n
from collections import Counter
n, k, trials = 10, 3, 200000
cnt = Counter()
for _ in range(trials):
for v in reservoir_sample_k(range(n), k):
cnt[v] += 1
for v in range(n):
print(v, round(cnt[v] / trials, 3)) # 应都 ≈ k/n = 0.3
常见误区
⚠️ 常见踩坑
误区一：用固定概率替换。 比如一直用 1/2，会让后来的元素更容易留下，破坏等概率。
误区二：只会背 k/i，但说不出场景。 面试官通常想听到“数据流、长度未知、内存有限”。
追问
追问 1：为什么 k=1 时第 1 个元素和第 n 个元素被选概率相同？
第 1 个元素一开始必选，但它要一路不被替换，概率会乘成 1/n；第 n 个元素最后一步以 1/n 概率入选，而且没有后续替换。所以两者最终都是 1/n。
追问 2：如果要按权重加权采样怎么办？
可以用 A-Res 一类加权蓄水池方法：给每个元素按权重生成随机 key，然后保留 key 最大的 k 个。口头重点是：普通蓄水池是等概率，加权版本要让高权重元素更容易进入样本。
🔗 相似问题
同一考点的不同问法，换着练更稳
延伸学习
按主题分类的相关资源，便于系统复习