如何随机播放存储在 Python 文件中的非常大的列表?

How can I shuffle a very large list stored in a file in Python?

我需要确定性地生成一个随机列表,其中包含从 0 到 2^32-1 的数字。

这将是一种天真的(并且完全没有功能)的做法,只是为了清楚我想要什么。

import random
numbers = range(2**32)
random.seed(0)
random.shuffle(numbers)

我试过用 numpy.arange() 制作列表并使用 pycrypto 的 random.shuffle() 来打乱它。制作列表消耗了大约 8gb 的 ram,然后洗牌将其增加到大约 25gb。我只有 32gb 可以提供。但这并不重要,因为...

我试过将列表切成 1024 片并尝试上面的方法,但即使是其中一片也太长了。我将这些切片中的一个切成 128 个更小的切片,that 每个切片大约需要 620 毫秒。如果它线性增长,那么这意味着整个过程大约需要 22 个半小时才能完成。听起来不错,但它不是线性增长的。

我尝试过的另一件事是为每个条目生成随机数并将其用作新位置的索引。然后我沿着列表往下看,并尝试将数字放在新索引处。如果该索引已在使用中,则该索引会递增,直到找到空闲索引为止。这在理论上是可行的,它可以完成大约一半,但接近尾声时它总是不得不搜索新的位置,在列表中循环多次。

有什么办法可以解决这个问题吗?这是一个可行的目标吗?

因此,一种方法是跟踪您已经给出了哪些数字,并继续一次一个地分发新的随机数,考虑

import random
random.seed(0)

class RandomDeck:
      def __init__(self):
           self.usedNumbers = set()

      def draw(self):
          number = random.randint(0,2**32)
          while number in self.usedNumbers:
                 number = random.randint(0,2**32)
          self.usedNumbers.append(number)
          return number

      def shuffle(self):
          self.usedNumbers = set()

如您所见,我们基本上有一组 0 到 2^32 之间的随机数,但我们只存储我们给出的数字以确保我们没有重复。然后你可以通过忘记你已经给出的所有数字来重新洗牌。

这在大多数用例中应该是有效的,只要您不在没有重新洗牌的情况下绘制大约 100 万个数字。

如果你有一个连续范围的数字,你根本不需要存储它们。在打乱后的列表中的值与其在该列表中的位置之间设计双向映射很容易。这个想法是使用伪随机排列,这正是 block ciphers 提供的。

诀窍是找到与您对 32 位整数的要求完全匹配的分组密码。很少有这样的块密码,但是 SimonSpeck 密码(由 NSA 发布)是可参数化的,并且支持 32- 的块大小位(通常块大小要大得多)。

This library 似乎提供了一个实现。我们可以设计以下功能:

def get_value_from_index(key, i):
    cipher = SpeckCipher(key, mode='ECB', key_size=64, block_size=32)
    return cipher.encrypt(i)

def get_index_from_value(key, val):
    cipher = SpeckCipher(key, mode='ECB', key_size=64, block_size=32)
    return cipher.decrypt(val)

该库使用 Python 的大整数,因此您甚至可能不需要对它们进行编码。

一个 64 位密钥(例如 0x123456789ABCDEF0)并不多。您可以使用类似的结构将 DES 中的密钥大小增加到 Triple DES。请记住,密钥应该随机选择,如果你想要确定性,它们必须是恒定的。


如果您不想为此使用 NSA 的算法,我会理解。还有其他的,但我现在找不到了。 Hasty Pudding 密码更加灵活,但我不知道是否有 Python.

的实现

我创建的 class 使用一个位数组来跟踪哪些数字已经被使用过。有了评论,我认为代码很容易解释。

import bitarray
import random


class UniqueRandom:
    def __init__(self):
        """ Init boolean array of used numbers and set all to False
        """
        self.used = bitarray.bitarray(2**32)
        self.used.setall(False)

    def draw(self):
        """ Draw a previously unused number
         Return False if no free numbers are left
        """

        # Check if there are numbers left to use; return False if none are left
        if self._free() == 0:
            return False

        # Draw a random index
        i = random.randint(0, 2**32-1)

        # Skip ahead from the random index to a undrawn number
        while self.used[i]:
            i = (i+1) % 2**32

        # Update used array
        self.used[i] = True

        # return the selected number
        return i

    def _free(self):
        """ Check how many places are unused
        """
        return self.used.count(False)


def main():
    r = UniqueRandom()
    for _ in range(20):
        print r.draw()


if __name__ == '__main__':
    main()

设计注意事项
虽然 Garrigan Stafford 的回答非常好,但该解决方案的内存占用要小得多(略多于 4 GB)。我们的答案之间的另一个区别是,当生成的数字数量增加时,Garrigan 的算法会花费更多时间来生成随机数(因为他一直迭代直到找到未使用的数字)。如果某个号码已被使用,该算法将查找下一个未使用的号码。这使得每次抽取数字所花费的时间几乎相同,无论免费数字池用完多远。

计算所有值似乎是不可能的,因为Crypto计算一个随机整数大约需要一毫秒,所以整个工作需要几天时间。

这是一个作为生成器的 Knuth 算法实现:

from Crypto.Random.random import randint  
import numpy as np

def onthefly(n):
    numbers=np.arange(n,dtype=np.uint32)
    for i in range(n):
        j=randint(i,n-1)
        numbers[i],numbers[j]=numbers[j],numbers[i]
        yield numbers[i]

对于n=10

gen=onthefly(10)
print([next(gen) for i in range(9)])
print(next(gen))
#[9, 0, 2, 6, 4, 8, 7, 3, 1]
#5

对于 n=2**32,生成器需要一分钟时间进行初始化,但调用的时间复杂度为 O(1)。

这是我写的一个排列 RNG,它使用了一个事实,即对一个数 mod 一个素数(加上一些复杂的)进行平方得到一个伪随机排列。

https://github.com/pytorch/pytorch/blob/09b4f4f2ff88088306ecedf1bbe23d8aac2d3f75/torch/utils/data/_utils/index_utils.py

简短版本:

def _is_prime(n):
    if n == 2:
        return True
    if n == 1 or n % 2 == 0:
        return False

    for d in range(3, floor(sqrt(n)) + 1, 2):  # can use isqrt in Python 3.8
        if n % d == 0:
            return False

    return True


class Permutation(Range):
    """
    Generates a random permutation of integers from 0 up to size.
    Inspired by https://preshing.com/20121224/how-to-generate-a-sequence-of-unique-random-integers/
    """

    size: int
    prime: int
    seed: int

    def __init__(self, size: int, seed: int):
        self.size = size
        self.prime = self._get_prime(size)
        self.seed = seed % self.prime

    def __getitem__(self, index):
        x = self._map(index)

        while x >= self.size:
            # If we map to a number greater than size, then the cycle of successive mappings must eventually result
            # in a number less than size. Proof: The cycle of successive mappings traces a path
            # that either always stays in the set n>=size or it enters and leaves it,
            # else the 1:1 mapping would be violated (two numbers would map to the same number).
            # Moreover, `set(range(size)) - set(map(n) for n in range(size) if map(n) < size)`
            # equals the `set(map(n) for n in range(size, prime) if map(n) < size)`
            # because the total mapping is exhaustive.
            # Which means we'll arrive at a number that wasn't mapped to by any other valid index.
            # This will take at most `prime-size` steps, and `prime-size` is on the order of log(size), so fast.
            # But usually we just need to remap once.
            x = self._map(x)

        return x

    @staticmethod
    def _get_prime(size):
        """
        Returns the prime number >= size which has the form (4n-1)
        """
        n = size + (3 - size % 4)
        while not _is_prime(n):
            # We expect to find a prime after O(log(size)) iterations
            # Using a brute-force primehood test, total complexity is O(log(size)*sqrt(size)), which is pretty good.
            n = n + 4
        return n

    def _map(self, index):
        a = self._permute_qpr(index)
        b = (a + self.seed) % self.prime
        c = self._permute_qpr(b)
        return c

    def _permute_qpr(self, x):
        residue = pow(x, 2, self.prime)

        if x * 2 < self.prime:
            return residue
        else:
            return self.prime - residue