对可迭代的大型笛卡尔积进行随机采样

Question

我有多个可迭代对象，我需要创建这些可迭代对象的笛卡尔积，然后从生成的元组池中随机采样。问题是这些可迭代的组合总数约为 1e19 左右，所以我不可能将所有这些加载到内存中。

我的想法是使用 itertools.product 与随机数生成器结合来跳过随机数量的项目，然后一旦我到达随机选择的项目，我就会执行计算并继续，直到用完发电机。计划是做类似的事情：

from itertools import product
from random import randint

iterables = () # tuple of 18 iterables
versions = product(iterables)

def do_stuff():
    # do stuff

STEP_SIZE = int(1e6)

# start both counts from 0. 
# First value to be taken is start + step
# after that increment start to be equal to count and repeat
start = 0
count = 0

while True:
    try:
        step = randint(1, 100) * STEP_SIZE

        for v in versions:
            # if the count is less than required skip values while incrementing count
            if count < start + step:
                versions.next()
                count += 1
            else:
                do_stuff(*v)
                start = count             
    except StopIteration:
        break

不幸的是，itertools.product对象没有next()方法，所以这不起作用。还有什么其他方法可以遍历如此大量的元组并随机采样或直接对值进行计算？

Answer 1

不要尝试生成笛卡尔积。使用random.choice()一次从一个迭代中采样以生成结果。所有可迭代对象的元素数量都很小，因此您可以将所有元素直接存储在内存中。

下面是一个使用 18 个迭代器的示例，每个迭代器有 10 个元素（如注释中指定）：

import random

iterables = [list(range(i, i + 10)) for i in range(0, 180, 10)]
result = [random.choice(iterable) for iterable in iterables]
print(result)

Answer 2

您使用的是哪个版本的 Python？在此过程中，.next() 方法被弃用，取而代之的是新的 next() 内置函数。这对所有迭代器都适用。例如，在当前发布的 3.10.1 下：

>>> import itertools
>>> itp = itertools.product(range(5), repeat=6)
>>> next(itp)
(0, 0, 0, 0, 0, 0)
>>> next(itp)
(0, 0, 0, 0, 0, 1)
>>> next(itp)
(0, 0, 0, 0, 0, 2)
>>> next(itp)
(0, 0, 0, 0, 0, 3)
>>> for ignore in range(50):
...     ignore = next(itp)
>>> next(itp)
(0, 0, 0, 2, 0, 4)

除此之外，您没有向我们展示代码中最重要的部分：如何构建产品。

如果没有看到这一点，我只能猜测，从传递给 product() 的第一个序列中进行随机选择，然后从第二个序列中进行随机选择，效率会远，等等。一次从一个组件构建产品的随机元素。

有效地选择随机乘积元组

也许有些过分了，但是这个类展示了一种特别有效的方法来做到这一点。 .index() 方法将整数 i 映射到乘积中的第 i 个元组（从 0 开始）。然后从产品中选择一个随机元组只需将 .index() 应用于范围（产品中元素的总数） 中的随机整数。

from math import prod
from random import randrange

class RanProduct:
    def __init__(self, iterables):
        self.its = list(map(list, iterables))
        self.n = prod(map(len, self.its))

    def index(self, i):
        if i not in range(self.n):
            raise ValueError(f"index {i} not in range({self.n})")
        result = []
        for it in reversed(self.its):
            i, r = divmod(i, len(it))
            result.append(it[r])
        return tuple(reversed(result))

    def pickran(self):
        return self.index(randrange(self.n))

进而

>>> r = RanProduct(["abc", range(2)])
>>> for i in range(6):
...     print(i, '->', r.index(i))
0 -> ('a', 0)
1 -> ('a', 1)
2 -> ('b', 0)
3 -> ('b', 1)
4 -> ('c', 0)
5 -> ('c', 1)
>>> r = RanProduct([range(10)] * 19)
>>> r.pickran()
(3, 5, 8, 8, 3, 6, 7, 6, 8, 6, 2, 0, 5, 6, 1, 0, 0, 8, 2)
>>> r.pickran()
(4, 5, 0, 5, 7, 1, 7, 2, 7, 4, 8, 4, 2, 0, 2, 9, 3, 6, 2)
>>> r.pickran()
(8, 7, 4, 1, 3, 0, 4, 6, 4, 3, 9, 8, 5, 8, 9, 9, 7, 1, 8)
>>> r.pickran()
(8, 6, 6, 0, 6, 7, 1, 3, 9, 5, 1, 4, 5, 8, 6, 8, 4, 9, 9)
>>> r.pickran()
(4, 9, 4, 7, 1, 5, 5, 1, 6, 7, 1, 8, 9, 0, 7, 9, 1, 7, 0)
>>> r.pickran()
(3, 0, 3, 9, 8, 6, 3, 0, 3, 0, 9, 9, 3, 5, 2, 3, 7, 8, 8)