如何从Python Async Iterator中随机选取元素？

因为random.sample()仅支持同步可迭代对象，针对异步迭代器可以用两种方案解决：

方法一：先收集所有元素再随机选取 #

如果数据量不大，直接把异步迭代器的所有元素加载到内存列表中，再用常规随机方法处理：

import random

async def async_iter_to_list(aiter):
    result = []
    async for item in aiter:
        result.append(item)
    return result

# 使用示例
participants = await async_iter_to_list(Participant.find(bot, **your_kwargs))
# 随机选取1个元素
random_participant = random.choice(participants)
# 随机选取n个元素
random_sample = random.sample(participants, k=3)

这种方式简单直观，但如果数据量极大，会占用较多内存，此时可以用下面的方法。

方法二：异步蓄水池抽样（适合大数据场景） #

蓄水池抽样算法可以在不加载全部数据到内存的前提下，从未知长度的序列中随机选取k个元素，以下是适配异步迭代器的版本：

import random

async def async_reservoir_sample(aiter, k):
    reservoir = []
    async for item in aiter:
        if len(reservoir) < k:
            reservoir.append(item)
        else:
            # 生成0到当前已遍历元素总数的随机数
            r = random.randint(0, len(reservoir))
            if r < k:
                reservoir[r] = item
    return reservoir

# 使用示例：随机选取1个元素
random_participant = (await async_reservoir_sample(Participant.find(bot, **your_kwargs), k=1))[0]
# 随机选取3个元素
random_sample = await async_reservoir_sample(Participant.find(bot, **your_kwargs), k=3)

这个算法仅需维护一个长度为k的"蓄水池"，遍历一次异步迭代器即可完成抽样，适合数据量庞大的场景。

注意：你的find方法返回的是AsyncIterable，必须用async for遍历，不能直接使用同步迭代逻辑。

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