实现“无限循环”数据集

Question

我想实现一个无限循环数据集

class Infinite(Dataset):
    def __len__(self):
        return HPARAMS.batch_size
#         return 1<<30 # This causes huge memory usage.
    def __getitem__(self, idx):
        """Randomly generates one new example."""
        return sample_func_to_be_parallelized()

infinite_loader = DataLoader(
    dataset=Infinite(), 
    batch_size=HPARAMS.batch_size, 
    num_workers=16,
    worker_init_fn=lambda worker_id: np.random.seed(worker_id),  
)

while True:
    for idx, data in enumerate(infinite_loader):
        # forward + backward on "data"

如您所见，这里的主要挑战是\uu len（）\uu方法。如果我在那里放一个足够大的数字，比如1

如果我在那里放一个小数字，比如1或BATCH_SIZE，训练循环中采样的“数据”将定期重复。这不是我想要的，因为我想产生新的数据

我猜过度使用内存的罪魁祸首是堆栈中的某个地方，缓存了一堆东西。随便看看Python的一面，我就不知道在哪里了。

有人能告诉我什么是实现我想要的最好的方法吗？（使用DataLoader的并行加载，同时确保加载的每个批都是全新的。）

Answer 1

尝试使用itertools中的cycle。以下是简单数据集的示例：

代码：

from itertools import cycle

import torch
from torch.utils.data import Dataset, DataLoader


# Create some dummy data.
data = torch.tensor([[0, 0],
                     [1, 1],
                     [2, 2],
                     [3, 3]])


class DataSet(Dataset):
    """Our dataset. Iterates over tensor data"""

    def __init__(self, data):
        self.data = data
        self.n = self.data.shape[0]

    def __len__(self):
        return self.n

    def __getitem__(self, idx):
        return self.data[idx]


bs = 1  # batch size
workers = 1  # number of workers

dataset = DataSet(data)
data_loader = DataLoader(dataset, batch_size=bs, shuffle=False, num_workers=workers)

# Infinite loop.
print(f'batch size: {bs} | number of workers: {workers}')
for i, data in cycle(enumerate(data_loader)):
    print(i, data)

输出：

batch size: 1 | number of workers: 1
0 tensor([[0, 0]])
1 tensor([[1, 1]])
2 tensor([[2, 2]])
3 tensor([[3, 3]])
0 tensor([[0, 0]])
1 tensor([[1, 1]])
2 tensor([[2, 2]])
3 tensor([[3, 3]])
...

batch size: 2 | number of workers: 2
0 tensor([[0, 0],
        [1, 1]])
1 tensor([[2, 2],
        [3, 3]])
0 tensor([[0, 0],
        [1, 1]])
1 tensor([[2, 2],
...

Answer 2

DataLoader在不替换数据集的情况下对数据集进行采样。为此，它生成一个介于0和len（数据集）之间的随机索引排列。我猜这种排列方式消耗了你大部分的记忆。我认为PyTorch API不支持无限集合，但您可以尝试在DataLoader中分叉代码，然后自己动手。您可以使用batch_sampler参数，并传入一个基于RandomSampler实现的自定义变量。这将允许您保留DataLoader的并行加载部分。

也就是说，基于\uu len\uu和\uu getitem\uu的迭代协议不适合无限集合。您最好重新实现数据集__len_uu只返回您的数据集1__getitem__;始终返回一个新样本，而不管索引如何，然后从此数据集中使用替换项对n次进行采样。从技术上讲，它将为第0个样本询问n次，但由于您重写\uuu getitem\uuuu以返回不同的样本，这将有效地满足您的要求。

Answer 3

这似乎在不定期复制数据的情况下起作用：

import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader

BATCH_SIZE = 2

class Infinite(Dataset):

    def __len__(self):
        return BATCH_SIZE

    def __getitem__(self, idx):
        return torch.randint(0, 10, (3,))


data_loader = DataLoader(Infinite(), batch_size=BATCH_SIZE, num_workers=16)

batch_count = 0
while True:
    batch_count += 1
    print(f'Batch {batch_count}:')

    data = next(iter(data_loader))
    print(data)
    # forward + backward on "data"  

    if batch_count == 5:
        break

结果:

Batch 1:
tensor([[4, 7, 7],
        [0, 8, 0]])
Batch 2:
tensor([[6, 8, 6],
        [2, 6, 7]])
Batch 3:
tensor([[6, 6, 2],
        [8, 7, 0]])
Batch 4:
tensor([[9, 4, 8],
        [2, 4, 1]])
Batch 5:
tensor([[9, 6, 1],
        [2, 7, 5]])

所以我认为问题出在你的函数sample_func_to_be_parallelized（）。

编辑：If而不是火炬。randint（0，10，（3，）I使用np。随机的randint（10，size=3）在\uuuuGetItem\uuuuuuuuuuuuuuuuuuuuuuuuuuuuuuuu中（作为sample\u func\u to\u be\u parallellelized（）的一个示例），那么每个批次的数据确实是重复的。看看这个问题。

因此，如果您在示例\u func\u to\u be\u parallelized（）中的某个地方使用numpy的RGN，那么解决方法是使用

worker_init_fn=lambda worker_id: np.random.seed(np.random.get_state()[1][0] + worker_id) 

并通过np重置种子。随机的seed（）在每次调用data=next（iter（数据加载器））之前。