抽样是数据处理里非常高频的需求，比如从千万条日志里抽一万条做分析、从海量样本中挑选测试集。数据量小的时候，用random.sample一行代码就能搞定，但数据规模一旦上亿，直接加载到内存就变得不现实。这篇文章就来详细聊聊Python里处理海量数据的几种抽样思路，从基础的随机抽样到经典的水塘抽样算法，再到工程上更实用的分块处理和数据库抽样方案。
一、为什么普通随机抽样在海量数据场景下会失效
先看最直观的写法：把数据全部读进来，然后用random.sample(data, k)随机抽取k条。这种方式在数据量几万、几十万的时候没有任何问题，逻辑清晰、结果准确。但它的核心瓶颈在于random.sample要求被抽样的对象是一个完整存在于内存中的序列，这意味着你必须先把整个数据集加载完毕。
假设有一份10GB的日志文件，每行一条记录，如果每条记录平均200字节，大约有5000万条。用列表全部装下再抽样，内存占用轻松超过10GB，普通机器根本扛不住。而且如果数据存放在数据库或者分布式文件系统里，先全量导出再抽样的做法既慢又浪费IO。
海量数据抽样的本质挑战是：能否在不完全加载数据的前提下，保证每条记录被抽中的概率相等。这就引出了后面要讲的流式抽样算法，它只需要常数级的内存就能完成对任意规模数据的均匀抽样。
二、水塘抽样：单遍扫描、常数内存的经典算法
水塘抽样（Reservoir Sampling）是海量数据抽样的首选方案。它的思路是：维护一个大小为k的“水塘”，数据流依次流过，前k条直接放进水塘；从第k+1条开始，第i条记录以k/i的概率替换掉水塘中随机一条记录。算法结束后，水塘里的k条记录就是一次均匀抽样，每条记录被选中的概率都是k/n。整个过程只需要遍历数据一次，内存占用只有k条记录的大小，跟数据总量完全无关。
用Python实现非常简单：
import random
def reservoir_sample(iterable, k):
reservoir = []
for i, item in enumerate(iterable):
if i < k:
reservoir.append(item)
else:
j = random.randint(0, i) # 生成 [0, i] 之间的随机整数
if j < k:
reservoir[j] = item # 以 k/(i+1) 的概率替换水塘中的一条
return reservoir
# 配合生成器逐行读取大文件，内存只跟 k 有关
def read_large_file(path):
with open(path, 'r', encoding='utf-8') as f:
for line in f:
yield line.strip()
sample = reservoir_sample(read_large_file('huge_log.txt'), 10000)
print(f'抽取了 {len(sample)} 条样本')
这段代码的关键点在于输入是一个生成器，文件是逐行流式读取的，任何时候内存里只有1万条样本加一行临时数据。哪怕文件有100GB，抽样过程也不会爆内存。需要强调的是randint(0, i)必须是闭区间，这样第i条记录被保留的概率才是k/(i+1)，这是算法正确性的核心，写错边界会导致抽样结果有偏。
水塘抽样的另一个优势是可以应用在未知长度的数据流上，比如实时消息队列，你不需要事先知道数据总量就能持续维护一组均匀样本。如果需要按行号精确控制，也可以先用wc -l统计总行数，再根据行号集合去抽取，但这样需要两次遍历文件，效率明显不如水塘抽样。
三、分层抽样与分块抽样：更贴近实际业务的做法
纯随机抽样在数据分布不均时可能出问题。比如数据里90%是正常请求、10%是异常请求，简单随机抽1万条可能只有1000条异常样本，分析时样本量不够。这时候就需要分层抽样：先按类别把数据分组，再在每组内按比例抽样。用pandas实现分层抽样可以借助groupby完成：
import pandas as pd
df = pd.read_csv('data.csv')
def stratified_sample(group, frac):
return group.sample(frac=frac, random_state=42)
# 按标签列分组，每组内抽取30%
stratified_df = df.groupby('label', group_keys=False).apply(stratified_sample, frac=0.3)
print(stratified_df['label'].value_counts(normalize=True))
如果文件大到pandas读不动，可以把分层抽样和水塘抽样结合：按类别维护多个水塘，数据流过时根据类别字段路由到对应的水塘。这样既保留了流式处理的低内存特性，又保证了各层的样本量充足。
分块抽样是另一种工程上常用的思路，即利用pandas.read_csv的chunksize参数分块读取，对每一块做小比例抽样，最后合并结果。这种方法的抽样比例是近似准确的，只要块数足够多，结果在统计上依然可用。它实现简单、代码改动小，适合快速出结果的探索性分析场景。
import pandas as pd
samples = []
for chunk in pd.read_csv('huge.csv', chunksize=100000):
samples.append(chunk.sample(frac=0.01, random_state=42))
result = pd.concat(samples, ignore_index=True)
print(f'共抽取 {len(result)} 条')
四、基于数据库和NumPy的高性能抽样方案
如果数据本身就在数据库里，最省事的方式是直接让数据库做抽样，避免把数据拉到Python进程。以MySQL为例，ORDER BY RAND()虽然写法直观，但会对全表排序，性能极差，千万级表就可能跑几十秒。更好的写法是利用主键随机生成行号再去匹配：
-- 性能较差的写法，全表排序
SELECT * FROM big_table ORDER BY RAND() LIMIT 10000;
-- 优化写法：随机生成主键范围，走索引
SELECT t.* FROM big_table t
JOIN (
SELECT ROUND(RAND() * (SELECT MAX(id) FROM big_table)) AS id
FROM big_table LIMIT 10000
) AS r ON t.id >= r.id
GROUP BY t.id LIMIT 10000;
如果数据已经是NumPy数组或者可以映射为索引，np.random.choice的效率远高于Python原生循环。需要注意replace=False时不放回抽样要求样本数不能超过总数，而且大数据集下它会生成一份完整的排列，内存开销不小。折中方案是只随机生成k个索引再取值：
import numpy as np n = 100_000_000 # 一亿条数据 k = 10000 idx = np.random.choice(n, size=k, replace=False) sample_indices = np.sort(idx) # 有了索引后，可以按需从文件或数据库中取出对应行
总结一下选型思路：数据在文件里且只需均匀抽样，首选水塘抽样；需要保证各类别比例，用分层抽样；数据在数据库中，优先用SQL层面的抽样；数据已经是内存中的数组，用NumPy的索引抽样最快。掌握这几种方案之后，绝大多数海量数据抽样场景都能找到合适的解法。