Pytorch：数据读取机制（DataLoader与Dataset）

大家好，欢迎来到IT知识分享网。

一、Pytorch的数据读取机制

数据模块中，DataLoader和DataSet就是数据读取子模块中的核心机制。
数据读取主要包含以下 3 个方面：

• 读取哪些数据：每个 Iteration 读取一个 Batchsize 大小的数据，每个 Iteration 应该读取哪些数据。
• 从哪里读取数据：如何找到硬盘中的数据，应该在哪里设置文件路径参数
• 如何读取数据：不同的文件需要使用不同的读取方法和库。

也就是说要使用Pytorch读取数据，首先应该新建一个类MyDataset，这个类要继承Dataset类并且实现里面的__getitem__方法，该方法用于定义如何接收一个索引idx， 返回一个样本对应的data和label。 此外还需要实现__len__，该方法用于计算样本数据，__len__返回总的样本的个数。

由于DataLoader是一个可迭代对象，当构建完成后可以简要查看读取的数据，以验证数据格式。

下面简要地介绍DataLoader和Dataset：

1.Dataload

torch.utils.data.Dataloader

功能：构建可迭代的数据装载器。训练的过程中，每一次iteration从DataLoader中获取一个batch_size大小的数据。
主要修改的参数如下表所示：

Epoch、Iteration、Batchsize之间的关系:

1:所有的样本数据都输入到模型中，称为一个epoch

2:一个Batch的样本输入到模型中，称为一个Iteration

3:一个批次的大小，一个Epoch=Batchsize*Iteration

例如： 假设样本总数80， Batchsize是8， 1Epoch=10 Iteration。 假设样本总数是87， Batchsize是8， 如果drop_last=True（丢弃最后一个样本数量不足batch_size批次数据，因为最后的余数为7）, 那么1Epoch=10Iteration, 如果等于False， 那么1Epoch=11Iteration, 最后1个Iteration有7个样本。

测试代码在后面实现的时候附上

2.Dataset

torch.utils.data.Dataset

功能：用来定义数据从哪里读取以及如何读取。Dataset抽象类，所有自定义的Dataset需要继承它，并且复写

主要修改的参数如下表所示：

测试代码在后面实现的时候附上

3.实现

自定义类

# 自定义类 class RMBDataset(Dataset): def __init__(self, data_dir, transform=None): """ rmb面额分类任务的Dataset :param data_dir: str, 数据集所在路径 :param transform: torch.transform，数据预处理 """ self.label_name = { 
   "1": 0, "100": 1} self.data_info = self.get_img_info(data_dir) # data_info存储所有图片路径和标签，在DataLoader中通过index读取样本 self.transform = transform def __getitem__(self, index): path_img, label = self.data_info[index] img = Image.open(path_img).convert('RGB') # 0~255 if self.transform is not None: img = self.transform(img) # 在这里做transform，转为tensor等等 return img, label def __len__(self): return len(self.data_info) @staticmethod def get_img_info(data_dir): data_info = list() for root, dirs, _ in os.walk(data_dir): # 遍历类别 for sub_dir in dirs: img_names = os.listdir(os.path.join(root, sub_dir)) img_names = list(filter(lambda x: x.endswith('.jpg'), img_names)) # 遍历图片 for i in range(len(img_names)): img_name = img_names[i] path_img = os.path.join(root, sub_dir, img_name) label = rmb_label[sub_dir] data_info.append((path_img, int(label))) return data_info 

划分数据集

import os import random import shutil def makedir(new_dir): # 辅助函数，用于验证地址 if not os.path.exists(new_dir): os.makedirs(new_dir) # if __name__ == '__main__': random.seed(1) #dataset_dir = os.path.join("..", "..", "data", "RMB_data")  #相对路径，一般用/来表示，’.’ 表示py文件当前所处的文件夹的绝对路径， #’..’ 表示py文件当前所处的文件夹上一级文件夹的绝对路径 # 注意要使用“\\”，否则“\”会被当成转义字符 dataset_dir = os.path.join( "data", "RMB_data") # 路径根据自己的文件所在位置设置。 split_dir = os.path.join("..", "..", "data", "rmb_split") train_dir = os.path.join(split_dir, "train") valid_dir = os.path.join(split_dir, "valid") test_dir = os.path.join(split_dir, "test") # 训练集和测试集、验证集比例 train_pct = 0.8 valid_pct = 0.1 test_pct = 0.1 for root, dirs, files in os.walk(dataset_dir): for sub_dir in dirs: imgs = os.listdir(os.path.join(root, sub_dir)) imgs = list(filter(lambda x: x.endswith('.jpg'), imgs)) random.shuffle(imgs) img_count = len(imgs) train_point = int(img_count * train_pct) valid_point = int(img_count * (train_pct + valid_pct)) for i in range(img_count): if i < train_point: out_dir = os.path.join(train_dir, sub_dir) elif i < valid_point: out_dir = os.path.join(valid_dir, sub_dir) else: out_dir = os.path.join(test_dir, sub_dir) makedir(out_dir) target_path = os.path.join(out_dir, imgs[i]) src_path = os.path.join(dataset_dir, sub_dir, imgs[i]) shutil.copy(src_path, target_path) print('Class:{}, train:{}, valid:{}, test:{}'.format(sub_dir, train_point, valid_point-train_point, img_count-valid_point)) 

因为基础比较差，作者在路径这一块卡了小半天，小菜鸡最后还是解决啦，学习了以下os.path.join()和相对路径以及绝对路径的表示，希望也能够帮到其他小朋友不用踩雷。

前面提到

借鉴参考博客作者的话，看源代码最好是先把逻辑关系给看懂， 然后再具体深入进去看具体细节。前面提到过重点是__ getitem__ 方法的实现，会发现有一个data_info[index]， 这个方法的大概意思就是通过提供的index经过一系列的处理后返回一个布尔型逻辑值。为了知道data_info是怎么来的，需要检索一下类的初始化部分__ init __，在看到赋值语句self.data_info = self.get_img_info(data_dir)后，又需要继续查看get_img_info(data_dir)。仔细查看这个函数的内容后发现，这个函数的参数是数据所在的路径data_dir，然后根据数据的路径去找到对应的数据，最后返回这个数据的位置和label，返回的格式是一个list，并且每一个list的元素是一个元组，格式就是[(样本1_loc, label_1), (样本2_loc, label_2), …(样本n_loc, label_n)]。这个list呢就是data_info的拿到的list，于是便可以通过索引index访问list中的元素(样本i_loc, label_i)了。
最后再回到__getitem __方法，便能够知道大概的意思了。第一行是通过索引index调用data_info方法得到一个样本图片的路径和label，然后第二行就是根据路径去找到这个图片并转化为RGB形式，最后第三行就是利用自定义的图像增广方法transform来对图片进行处理（如果图片是非空的话），最后返回样本的张量形式（上一节提到过tensor是Pytorch特有的一种数据形式）和label。
了解完大致的逻辑后便可以进行学习具体的细节实现了，这里我也偷个懒，就不继续叙述了。

因为Dataloader的源码太长啦，具体实现就不细说了，先来看看tran_loader做的主要事情：输出了一个<torch.utils.data.dataloader.DataLoader object at 0x000001D8C284DBC8>， 可以发现这是一个DataLoader对象，显然这是一个可迭代的对象。 那么便可以知道下边训练部分的代码就要遍历这个train_loade了, 然后每一次取一批数据进行训练啦。

实现

# ============================ step 1/5 数据 ============================ # split_dir = os.path.join("..", "..", "data", "rmb_split") split_dir = os.path.join("data", "rmb_split") train_dir = os.path.join(split_dir, "train") valid_dir = os.path.join(split_dir, "valid") norm_mean = [0.485, 0.456, 0.406] #均值 norm_std = [0.229, 0.224, 0.225] #标准差 #Compose是将一系列transforms方法进行有序地组合 train_transform = transforms.Compose([ transforms.Resize((32, 32)), #缩放为32*32大小 transforms.RandomCrop(32, padding=4), #随机裁剪 transforms.ToTensor(), #图像数据转换成张量，同时会进行归一化操作，把像素值从0-255归一化到0-1 transforms.Normalize(norm_mean, norm_std), #数据标准化，把数据均值变为0，标准差变为1 ]) #验证集不需要随即裁剪 valid_transform = transforms.Compose([ transforms.Resize((32, 32)), transforms.ToTensor(), transforms.Normalize(norm_mean, norm_std), ]) # 构建MyDataset实例 train_data = RMBDataset(data_dir=train_dir, transform=train_transform) valid_data = RMBDataset(data_dir=valid_dir, transform=valid_transform) # 构建DataLoder train_loader = DataLoader(dataset=train_data, batch_size=BATCH_SIZE, shuffle=True) #shuffle设置为True valid_loader = DataLoader(dataset=valid_data, batch_size=BATCH_SIZE) # ============================ step 2/5 模型 ============================ #初始化卷积神经网络LeNet net = LeNet(classes=2) net.initialize_weights() # ============================ step 3/5 损失函数 ============================ criterion = nn.CrossEntropyLoss() #选择损失函数,分类任务通常采用交叉熵损失函数 # ============================ step 4/5 优化器 ============================ optimizer = optim.SGD(net.parameters(), lr=LR, momentum=0.9) #选择优化器，随机梯度下降 scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.1) # 设置学习率下降策略 # ============================ step 5/5 训练 ============================ train_curve = list() valid_curve = list() for epoch in range(MAX_EPOCH): loss_mean = 0. correct = 0. total = 0. net.train() #enumerate返回枚举对象，比如0 ‘Spring’, 1 'Summer', 3 'Fall', 4 'Winter' for i, data in enumerate(train_loader): #不断从DataLoader中获取一个batch size大小的数据 # forward inputs, labels = data outputs = net(inputs) # backward optimizer.zero_grad() loss = criterion(outputs, labels) loss.backward() # update weights optimizer.step() # 统计分类情况 _, predicted = torch.max(outputs.data, 1) total += labels.size(0) correct += (predicted == labels).squeeze().sum().numpy() # 打印训练信息 loss_mean += loss.item() train_curve.append(loss.item()) if (i+1) % log_interval == 0: loss_mean = loss_mean / log_interval print("Training:Epoch[{:0>3}/{:0>3}] Iteration[{:0>3}/{:0>3}] Loss: {:.4f} Acc:{:.2%}".format( epoch, MAX_EPOCH, i+1, len(train_loader), loss_mean, correct / total)) loss_mean = 0. scheduler.step() # 更新学习率 # validate the model if (epoch+1) % val_interval == 0: correct_val = 0. total_val = 0. loss_val = 0. net.eval() with torch.no_grad(): for j, data in enumerate(valid_loader): inputs, labels = data outputs = net(inputs) loss = criterion(outputs, labels) _, predicted = torch.max(outputs.data, 1) total_val += labels.size(0) correct_val += (predicted == labels).squeeze().sum().numpy() loss_val += loss.item() loss_val_epoch = loss_val / len(valid_loader) valid_curve.append(loss_val_epoch) # valid_curve.append(loss.item()) # 改，记录整个epoch样本的loss，注意要取平均 print("Valid:\t Epoch[{:0>3}/{:0>3}] Iteration[{:0>3}/{:0>3}] Loss: {:.4f} Acc:{:.2%}".format( epoch, MAX_EPOCH, j+1, len(valid_loader), loss_val_epoch, correct_val / total_val)) train_x = range(len(train_curve)) train_y = train_curve train_iters = len(train_loader) valid_x = np.arange(1, len(valid_curve)+1) * train_iters*val_interval # 由于valid中记录的是epochloss，需要对记录点进行转换到iterations valid_y = valid_curve plt.plot(train_x, train_y, label='Train') plt.plot(valid_x, valid_y, label='Valid') plt.legend(loc='upper right') plt.ylabel('loss value') plt.xlabel('Iteration') plt.show() # ============================ inference ============================ BASE_DIR = os.path.dirname(os.path.abspath(__file__)) test_dir = os.path.join(BASE_DIR, "test_data") test_data = RMBDataset(data_dir=test_dir, transform=valid_transform) valid_loader = DataLoader(dataset=test_data, batch_size=1) for i, data in enumerate(valid_loader): # forward inputs, labels = data outputs = net(inputs) _, predicted = torch.max(outputs.data, 1) rmb = 1 if predicted.numpy()[0] == 0 else 100 print("模型获得{}元".format(rmb)) 

Lenet回顾

由于运行代码的时候，发现用了Lenet网络，正巧就回顾一下Lenet的一些基本定义。

总体来看，(LeNet（LeNet-5）由两个部分组成：)

• 卷积编码器：由两个卷积层组成;
• 全连接层密集块：由三个全连接层组成。

为了将卷积块的输出传递给稠密块，必须在小批量中展平每个样本。换言之，将这个四维输入转换成全连接层所期望的二维输入。这里的二维表示的第一个维度索引小批量中的样本，第二个维度给出每个样本的平面向量表示。LeNet的稠密块有三个全连接层，分别有120、84和10个输出。因为在执行分类任务，所以输出层的10维对应于最后输出结果的数量。

以上就是Pytorch读取机制DataLoader和Dataset的原理和一个demo的实现啦~

参考资料