水库抽样算法（大数据算法作业）

时隔一个多月，终于想起来写大数据算法基础的实验报告，主要是快截止了，hh

这两天加急把这个报告写完了~

接下来，写一写证明过程（参考书籍：高等教育出版社《数据科学与工程算法基础》）主要代码以及总结体会o(*￣▽￣*)ブ

本次实验主要设计三块内容，分别是水库抽样算法（当水库大小为1时），水库抽样算法（当水库大小为k>1时）以及分布式水库抽样算法

水库抽样算法

主要证明过程

主要Python代码

水库抽样算法（返回一个）

import randomdef sampling_single(stream):reservoir = stream[0]i = 1for i, item in enumerate(stream):j = random.randint(0, i)if j < 1:reservoir = itemreturn reservoir F = [i for i in range(100)]H = sampling_single(F)
print(f"Randomly sampled element: {H}")

水库抽样算法（返回多个）

import randomdef reservoir_sampling(stream, k):reservoir = []for i, item in enumerate(stream):if i < k:reservoir.append(item)else:j = random.randint(0, i)if j < k:reservoir[j] = itemreturn reservoirdata_stream = [i for i in range(100)]sampled_data = reservoir_sampling(data_stream, 10)

分布式水库抽样算法

主要证明过程

一个Hadoop任务Sample由 n 个 Map 组成，其中每个 Map 都接受到一个数据流 Substream，当这些数据无法完全保存在内存时，如何随机地抽取一个含有 k 条记录的样本（每条记录被抽中的概率相同），于是，这就引出了分布式水库抽样算法（分层水库抽样 + 重抽样 = 分布式水库抽样算法）

先在每个 Map 上独立运行水库抽样算法，之后对 n 个子样本就行重抽样，获得满足要求的最终结果。

主要 Python 代码

import randomdef reservoir_sampling(stream, k):reservoir = []for i, item in enumerate(stream):if i < k:reservoir.append(item)else:j = random.randint(0, i)if j < k:reservoir[j] = itemreturn reservoirdef distributed_sampling(n, k, stream):N = []F = []H = []for i in range(n):F.append(reservoir_sampling(stream, k))N.append(len(F[i]))total_N = sum(N)for j in range(k):p = random.random()m = 0cumulative_N = 0while cumulative_N < p * total_N :cumulative_N += N[m]m += 1H.append(random.choice(F[m-1]))return Hn = 15
k = 10
data_stream = [i for i in range(100)]
H = distributed_sampling(n, k, data_stream)
print("Final Sample H:", H)