当前位置:   article > 正文

Pytorch中数据读取-Dataset、Dataloader 、TensorDataset 和 Sampler 的使用_from torch.utils.data import dataloader

from torch.utils.data import dataloader

0.引言

Pytorch 创建用以输入到模型的数据的一般流程如下:

  1. 创建一个 Dataset 对象,实现__getitem__()__len__()这两个方法

  2. 创建一个 DataLoader 对象,该对象可以对上述Dataset对象进行迭代

  3. 遍历DataLoader对象,将样本和标签加载到模型中进行训练

在上述流程中会涉及 Dataset 、 Dataloader 、Sampler 和 TensorDataset,以下将逐一介绍。

1. Dataset

Dataset 是一个抽象类,所有自定义的 datasets 都需要继承该类,并且重载__getitem()__方法和__len__()方法 。__getitem()__方法的作用是接收一个索引,返回索引对应的样本和标签,这需要根据真实数据具体实现的逻辑。__len__()方法是返回所有样本的数量。

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

Data = np.array([[1, 2], [3, 4],[5, 6], [7, 8]])
Label = np.array([[0], [1], [0], [2]])
# 创建子类
class CustomDataset(Dataset):
    # 初始化,定义数据内容和标签
    def __init__(self, Data, Label):
        self.Data = Data
        self.Label = Label

    # 返回数据集大小
    def __len__(self):
        return len(self.Data)

    # 得到数据内容和标签
    def __getitem__(self, index):
        data = torch.Tensor(self.Data[index])
        label = torch.IntTensor(self.Label[index])
        return data, label


dataset = CustomDataset(Data, Label)
print(dataset)
print('dataset大小为:', dataset.__len__())
print(dataset.__getitem__(0))
print(dataset[0])
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

运行结果如下:

<__main__.test_datasets.<locals>.CustomDataset object at 0x7f4bf21d1128>
dataset大小为: 4
(tensor([1., 2.]), tensor([0], dtype=torch.int32))
(tensor([1., 2.]), tensor([0], dtype=torch.int32))
  • 1
  • 2
  • 3
  • 4

1.2 延伸

其实有2种类型的 Dataset,一种就是上述这种,名为map-style datasets;另一种是iterable-style datasets。一个iterable-style的dataset实例需要继承IterableDataset类并实现__iter__()方法。这种类型的datasets 特别适用于随机读取代价大甚至不可能的情况,以及batch size取决于获取的数据。例如,读取数据库,远程服务器或者实时日志等数据的时候,可使用该样式,一般时序数据不使用这种样式。

2. DataLoader

torch.utils.data.DataLoader是PyTorch中加载数据集的核心。DataLoader 返回的是可迭代的数据装载器(DataLoader),其初始化的参数设置如下。

DataLoader(dataset, batch_size=1, shuffle=False, sampler=None,
           batch_sampler=None, num_workers=0, collate_fn=None,
           pin_memory=False, drop_last=False, timeout=0,
           worker_init_fn=None, *, prefetch_factor=2,
           persistent_workers=False)
  • 1
  • 2
  • 3
  • 4
  • 5

在上述定义的CustomDataset基础上使用DataLoader对其进行遍历:

# 创建DataLoader迭代器
dataloader = DataLoader(dataset, batch_size=2, shuffle=False, num_workers=4)
for i, item in enumerate(dataloader):
    print('i:', i)
    data, label = item
    print('data:', data)
    print('label:', label)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

运行结果:

i: 0
data: tensor([[1., 2.],
        [3., 4.]])
label: tensor([[0],
        [1]], dtype=torch.int32)
i: 1
data: tensor([[5., 6.],
        [7., 8.]])
label: tensor([[0],
        [2]], dtype=torch.int32)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

3.Sampler

DataLoader的参数初始化中有两种sampler:samplerbatch_sampler,都默认为None。前者的作用是生成一系列的index,而batch_sampler则是将sampler生成的indices打包分组,得到一个又一个batch的index。生成的index是遍历Dataset所需的索引。例如下面示例中,BatchSamplerSequentialSampler生成的index按照指定的batch size分组。

a = list(BatchSampler(SequentialSampler(range(10)), batch_size=3, drop_last=False))
print(a)
  • 1
  • 2

运行结果如下:

[[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]
  • 1

以下进一步介绍几种较为常见的Sampler:SequentialSamplerRandomSamplerBatchSampler

3.1 SequentialSampler

SequentialSampler初始化形式:torch.utils.data.SequentialSampler(data_source)
SequentialSampler按顺序对数据集采样。其原理是首先在初始化的时候拿到数据集data_source,之后在__iter__方法中首先得到一个和data_source一样长度的range可迭代器。每次只会返回一个索引值。SequentialSamplerDataLoader提供了顺序遍历dataset的方式。

示例代码:

a = [1, 5, 7, 9, 10086]
b = torch.utils.data.SequentialSampler(a)
for x in b:
    print(x)
  • 1
  • 2
  • 3
  • 4

运行结果:

0
1
2
3
4
  • 1
  • 2
  • 3
  • 4
  • 5

3.2 RandomSampler

RandomSampler初始化形式:torch.utils.data.RandomSampler(data_source, replacement=False, num_samples=None, generator=None)RandomSampler初始化参数除了data_source还有以下2个。

  • num_samples: 指定采样的数量,默认是所有。
  • replacement: 默认是False,若为True,则表示可以重复采样,即同一个样本可以重复采样,这样可能导致有的样本采样不到。所以此时可以设置num_samples来增加采样数量使得每个样本都可能被采样到。

示例代码:

    torch.manual_seed(42) # 固定住 seed, 否则每次都会生成不同的结果indexs
    a = [1, 5, 7, 9, 10086]
    b = torch.utils.data.RandomSampler(a)
    for x in b:
        print(x)
  • 1
  • 2
  • 3
  • 4
  • 5

运行结果:

1
3
2
4
0
  • 1
  • 2
  • 3
  • 4
  • 5

3.3 WeightedSampler

WeightedSampler是根据给定的权重进行采样。
WeightedRandomSampler初始化形式:torch.utils.data.WeightedRandomSampler(weights, num_samples, replacement=True, generator=None),各参数说明如下:

  • weights (sequence) – 权重列表, 无需权重列表的和为1
  • num_samples (int) – 抽取的样本数
  • replacement (bool) – 与RandomSampler中的一样
  • generator (Generator) – 用于采样的发生器Generator

示例代码:

    torch.manual_seed(4)
    a= list(WeightedRandomSampler([0.1, 0.9, 0.4, 0.7, 3.0, 0.6], 5, replacement=True))
    print(a)
    b = list(WeightedRandomSampler([0.1, 0.9, 0.4, 0.7, 3.0, 0.6], 5, replacement=False))
    print(b)
  • 1
  • 2
  • 3
  • 4
  • 5

运行结果:

[4, 4, 1, 5, 5]
[4, 2, 1, 5, 3]
  • 1
  • 2

3.4 SubsetRandomSampler

SubsetRandomSampler是从给定的索引列表中抽样元素,该过程不重复采样。
SubsetRandomSampler初始化形式:torch.utils.data.SubsetRandomSampler(indices, generator=None)。有如下参数:

  • indices (sequence) – 索引列表
  • generator (Generator) – 用于采样的发生器Generator

这个采样器常见的使用场景是将训练集划分成训练集和验证集,示例如下:

    batch_size = 2
    validation_split = .2
    shuffle_dataset = True
    random_seed = 42
    Data = torch.arange(20)
    Data = Data.reshape(10, 2)
    Label = torch.arange(10)
    Label = Label.reshape(10,1)
    
    # 创建子类
    class CustomDataset(Dataset):
        # 初始化,定义数据内容和标签
        def __init__(self, Data, Label):
            self.Data = Data
            self.Label = Label

        # 返回数据集大小
        def __len__(self):
            return len(self.Data)

        # 得到数据内容和标签
        def __getitem__(self, index):
            data = torch.LongTensor(self.Data[index])
            label = torch.LongTensor(self.Label[index]) #torch.IntTensor
            return data, label

    dataset = CustomDataset(Data, Label)
    dataset_size = len(dataset)
    indices = list(range(dataset_size))
    split = int(np.floor(validation_split * dataset_size))
    if shuffle_dataset:
        np.random.seed(random_seed)
        np.random.shuffle(indices)
    train_indices, val_indices = indices[split:], indices[:split]

    # Creating PT data samplers and loaders:
    train_sampler = SubsetRandomSampler(train_indices)
    valid_sampler = SubsetRandomSampler(val_indices)

    train_loader = torch.utils.data.DataLoader(dataset, batch_size=batch_size,
                                               sampler=train_sampler)
    validation_loader = torch.utils.data.DataLoader(dataset, batch_size=batch_size,
                                                    sampler=valid_sampler)

    # Usage Example:
    print("train data:")
    for batch_index, (data, labels) in enumerate(train_loader):
        print(data, labels)

    print("\nvalidation data:")
    for batch_index, (data, labels) in enumerate(validation_loader):
        print(data, labels)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52

运行结果如下:

train data:
tensor([[ 4,  5],
        [10, 11]]) tensor([[2],
        [5]])
tensor([[18, 19],
        [ 6,  7]]) tensor([[9],
        [3]])
tensor([[14, 15],
        [12, 13]]) tensor([[7],
        [6]])
tensor([[0, 1],
        [8, 9]]) tensor([[0],
        [4]])

validation data:
tensor([[16, 17],
        [ 2,  3]]) tensor([[8],
        [1]])
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18

PS:上述使用SubsetRandomSampler的时候关键是获取索引列表,可以使用np.random.choice生成:

np.random.choice(indices, dataset_size)
#numpy.random.choice(a, size=None, replace=True, p=None)
#从a(只要是ndarray都可以,但必须是一维的)中随机抽取数字,并组成指定大小(size)的数组
#replace:True表示可以取相同数字,False表示不可以取相同数字
#数组p:与数组a相对应,表示取数组a中每个元素的概率,默认为选取每个元素的概率相同。
  • 1
  • 2
  • 3
  • 4
  • 5

3.5 DistributedSampler

DistributedSampler用于在多机多卡情况下分布式训练数据的读取是一个问题,不同的卡读取到的数据应该是不同的。dataparallel的做法是直接将batch切分到不同的卡,这种方法对于多机来说不可取,因为多机之间直接进行数据传输会严重影响效率。于是有了利用sampler确保dataloader只会load到整个数据集的一个特定子集的做法。DistributedSampler就是做这件事的,可以约束数据只加载数据集的子集。它为每一个子进程划分出一部分数据集,以避免不同进程之间数据重复。DistributedSampler一般与torch.nn.parallel.DistributedDataParallel搭配使用。在这种情况下,每个进程都可以将DistributedSampler实例作为DataLoader的sampler,并加载专属于它的原始数据集的子集。

DistributedSampler初始化形式:torch.utils.data.distributed.DistributedSampler(dataset, num_replicas=None, rank=None, shuffle=True, seed=0, drop_last=False)。具体参数定义如下:

  • dataset – 待采样的Dataset.
  • num_replicas (int, optional) – 分布式训练过程中使用的进程数,一般是使用world_size,world size指进程总数,在这里就是我们使用的卡数。
  • rank (int, optional) – 指当前进程序号。
  • shuffle (bool, optional) – 如果为True (默认), sampler则会shuffle索引。
  • seed (int, optional) – 用于对sampler进行shuffle的seed ,其前提是shuffle=True. 该seed数字在分布式组中的所有进程中是相同的。默认值为0。
  • drop_last (bool, optional) – 默认值为False。如果为True, sampler 丢弃尾部的数据,使其在各副本上均匀地可分。如果为False,sampler中将添加额外的索引,使数据在多个副本之间均匀分割。

注意:
在分布式模式下,创建DataLoader迭代器之前需要在每个epoch开始时先调用set_epoch()方法,从而令shuffling在多个epoch中生效。否则,总是会使用相同的顺序。
示例代码:

sampler = DistributedSampler(dataset) if is_distributed else None
loader = DataLoader(dataset, shuffle=(sampler is None), sampler=sampler)
for epoch in range(start_epoch, n_epochs):
    if is_distributed:
        sampler.set_epoch(epoch)
    train(loader)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

4. TensorDataset

TensorDataset 可以用来对 tensor 进行打包,其功能类似 python 中的 zip,将输入的tensors捆绑在一起组成元祖。该类通过每一个 tensor 的第一个维度进行索引。因此,该类中的 tensor 第一维度必须相等。
Pytorch 中 TensorDataset 类的定义如下:

class TensorDataset(Dataset[Tuple[Tensor, ...]]):
    r"""Dataset wrapping tensors.

    Each sample will be retrieved by indexing tensors along the first dimension.

    Arguments:
        *tensors (Tensor): tensors that have the same size of the first dimension.
    """
    tensors: Tuple[Tensor, ...]

    def __init__(self, *tensors: Tensor) -> None:
        assert all(tensors[0].size(0) == tensor.size(0) for tensor in tensors), "Size mismatch between tensors"
        self.tensors = tensors

    def __getitem__(self, index):
        return tuple(tensor[index] for tensor in self.tensors)

    def __len__(self):
        return self.tensors[0].size(0)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

通过代码可以看出TensorDatasetDataset的子类,已经重载了__len____getitem__方法。__getitem__表示每个tensor取相同的索引,然后将这个结果组成一个元组。

示例代码:

    data = torch.arange(12)
    label = torch.arange(12,0,-1)
    print("data=", data)
    print("label=", label)
    data_label = TensorDataset(data, label)
    print("data_label len=", data_label.__len__())
    print("data_label[0]=",data_label[0])


    data_2 = data.reshape(3,4)
    label_2 = torch.arange(3)
    print("data2=", data_2)
    print("label2=", label_2)
    data_label_2 = TensorDataset(data_2, label_2)
    print("data_label_2 len=", data_label_2.__len__())
    print("data_label_2[1]=",data_label_2[1])
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

运行结果:

data= tensor([ 0,  1,  2,  3,  4,  5,  6,  7,  8,  9, 10, 11])
label= tensor([12, 11, 10,  9,  8,  7,  6,  5,  4,  3,  2,  1])
data_label len= 12
data_label[0]= (tensor(0), tensor(12))
data2= tensor([[ 0,  1,  2,  3],
        [ 4,  5,  6,  7],
        [ 8,  9, 10, 11]])
label2= tensor([0, 1, 2])
data_label_2 len= 3
data_label_2[1]= (tensor([4, 5, 6, 7]), tensor(1))

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/Gausst松鼠会/article/detail/123880
推荐阅读
相关标签
  

闽ICP备14008679号