【深度学习】DNN Regression 代码实现与详解

相关说明

• 数据集分成两个文件，train和test。训练时，将train分为训练集和验证集，用于训练模型以及判断模型训练的好坏。
• test数据集用于最终测试模型的通用性，即所训练出来的模型是否“过拟合”。

1.下载数据集

tr_path = 'covid.train.csv'  # 设置训练集数据存储的地址
tt_path = 'covid.test.csv'   # 设置测试集数据存储的地址

# !gdown为Google Colab所支持的特殊指令，用于下载存储在云盘中数据
!gdown --id '19CCyCgJrUxtvgZF53vnctJiOJ23T5mqF' --output covid.train.csv
!gdown --id '1CE240jLm2npU-tdz81-oVKEF3T2yfT1O' --output covid.test.csv

下载数据不方便的同学，用下面的链接下载即可。
链接：https://pan.baidu.com/s/1iBWYkSy-Jj8UwHYnSSS73w
提取码：amva

2.导入相关的包

# 导入后续需要使用到的相关包
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader

# 导入对数据集进行相关处理的包
import numpy as np
import csv
import os

# 导入绘制图像相关的包
import matplotlib.pyplot as plt
from matplotlib.pyplot import figure

# 设置一个随机种子，以保证程序的可重复性
# 相关知识：使用相同的随机数种子，能够保证每次实验所生成的随机数相同
myseed = 42069  
torch.backends.cudnn.deterministic = True	# 设置为True，保证每次训练使用的卷积算法一致
# 算法一致：算法运行在相同的软硬件的前提下，如果具有相同的输入，则输出相同
# Note：确定性算法往往比不确定性算法有更坏的性能表现

torch.backends.cudnn.benchmark = False		# 布尔值，为真将使cuDNN对多个卷积算法进行基准测试，并选择最快的算法。
# cuDNN 是英伟达专门为深度神经网络所开发出来的 GPU 加速库，针对卷积、池化等等常见操作做了非常多的底层优化
# 如果卷积网络结构不是动态变化的，网络的输入 (batch size，图像的大小，输入的通道) 是固定的，设置为True。由于本文并未涉及卷积运算，所以设置为False

# 设置numpy、torch、torch.cuda的随机数种子
np.random.seed(myseed)
torch.manual_seed(myseed)
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(myseed)

3.定义基本函数

def get_device():
    ''' 判断GPU是否可用，可用则返回cuda，否则返回cpu'''
    return 'cuda' if torch.cuda.is_available() else 'cpu'

def plot_learning_curve(loss_record, title=''):
    ''' 训练结束后，用于绘制整个训练过程中Loss值的变化 '''
    total_steps = len(loss_record['train'])
    # 
    x_1 = range(total_steps)
    x_2 = x_1[::len(loss_record['train']) // len(loss_record['dev'])]
    figure(figsize=(6, 4))			                      # 设置图表的宽、高(以英尺为单位)
    plt.plot(x_1, loss_record['train'], c='tab:red', label='train')
    plt.plot(x_2, loss_record['dev'], c='tab:cyan', label='dev')
    plt.ylim(0.0, 5.)									  # 限制y轴大小为0~5
    plt.xlabel('Training steps')
    plt.ylabel('MSE loss')
    plt.title('Learning curve of {}'.format(title))
    plt.legend()										  # 使修改的label生效
    plt.show()


def plot_pred(dv_set, model, device, lim=35., preds=None, targets=None):
    ''' 绘制训练后的DNN网络的预测结果 '''
    if preds is None or targets is None:
        model.eval()									  # 设置模型围为测试模式
        preds, targets = [], []
        for x, y in dv_set:
            x, y = x.to(device), y.to(device)
            with torch.no_grad():						  # 关闭梯度计算，降低内存消耗，加快运行速度
                pred = model(x)                     
                preds.append(pred.detach().cpu())         # detach()运算：用于将Tensor从计算图中分离出来 cpu()运算：将Tensor数据传回CPU
                targets.append(y.detach().cpu())         
        preds = torch.cat(preds, dim=0).numpy()          # torch.cat()：在给定维中(dim指定)连接给定序列的seq张量。所有张量必须具有相同的形状（连接维度中除外）或为空。
        targets = torch.cat(targets, dim=0).numpy()

    figure(figsize=(5, 5))
    # plt.scatter()用于绘制散点图
    plt.scatter(targets, preds, c='r', alpha=0.5)
    plt.plot([-0.2, lim], [-0.2, lim], c='b')
    plt.xlim(-0.2, lim)
    plt.ylim(-0.2, lim)
    plt.xlabel('ground truth value')
    plt.ylabel('predicted value')
    plt.title('Ground Truth v.s. Prediction')
    plt.show()

4.定义数据处理类

自定义数据集类必须实现三个函数： __init__()、__len__()和__getitem__()。本文在类COVID19Dataset有所体现。

• __init__()：本函数仅在实例化类时被调用一次。用于完成一些初始化的操作，如读取数据集的数据、分割数据集等。
• __len__()：调用该函数返回当前数据集中数据个数。
• __getitem__(idx)：返回当前数据集中对应idx的数据。对于训练集，返回样本和对应的标签；对于测试集，通常仅返回样本。

class COVID19Dataset(Dataset):
    ''' 用于加载和预处理COVID19 数据集的类'''
    def __init__(self,path,mode='train',target_only=False):
        self.mode = mode

        # 读取数据，保存格式为numpy数组
        with open(path, 'r') as fp:
            data = list(csv.reader(fp))
            data = np.array(data[1:])[:, 1:].astype(float)
        # target_only为默认值，则选取所有93个特征作为训练数据；
        if not target_only:
            feats = list(range(93))
        else:
            pass

        if mode == 'test':
            # 由于测试集不含有标签数据，故仅对data进行操作即可
            data = data[:, feats]
            self.data = torch.FloatTensor(data)
        else:
            # 针对训练集的操作
            # 读取的原始数据: 2700 x 94 (40 states + day 1 (18) + day 2 (18) + day 3 (18))
            # 处理后，data为2700 x 93,target为2700 x 1
            target = data[:, -1]
            data = data[:, feats]
            
            # 将训练集划分为训练集和验证集，此处按照：训练集：验证集=9：1 的比例进行
            if mode == 'train':
                indices = [i for i in range(len(data)) if i % 10 != 0]
            elif mode == 'dev':
                indices = [i for i in range(len(data)) if i % 10 == 0]
            
            # 将numpy array格式的数据均转化为torch.FloatTensor类型
            self.data = torch.FloatTensor(data[indices])
            self.target = torch.FloatTensor(target[indices])

        # 归一化特征；这种做法通常有利于提升模型训练的效果
        self.data[:, 40:] = \
            (self.data[:, 40:] - self.data[:, 40:].mean(dim=0, keepdim=True)) \
            / self.data[:, 40:].std(dim=0, keepdim=True)

        self.dim = self.data.shape[1]

        print('Finished reading the {} set of COVID19 Dataset ({} samples found, each dim = {})'
              .format(mode, len(self.data), self.dim))

    def __getitem__(self, index):
        # 根据传入的index值返回数据
        if self.mode in ['train', 'dev']:
            return self.data[index], self.target[index]
        else:
            return self.data[index]

    def __len__(self):
        # 返回数据集的长度(尺寸)
        return len(self.data)

5.定义数据加载器"DataLoader"

数据加载器的意义：前一步构造的数据集实现了每一次返回一个样本和对应的标签，但在训练模型时，我们希望一次给模型“喂”进去一批样本minibatches”，同时在每个epoch按照不同的次序导入数据shuffle 。上述做法能够起到降低模型过拟合，以及利用Python的批处理能力加速训练的目的。

DataLoader可作为迭代器使用。每次迭代返回由batch_size控制的数据规模。

def prep_dataloader(path, mode, batch_size, n_jobs=0, target_only=False):
    '''构造一个数据集，再将数据集传入数据加载器'''
    dataset = COVID19Dataset(path, mode=mode, target_only=target_only) 
    # num_workers(可选):用于数据加载的子进程数。0表示将在主进程中加载数据。（默认值：0）
    # pin_memory(可选) :如果为True，数据加载程序将在返回张量之前将张量复制到CUDA固定内存中。
    # drop_last(可选)  :如果数据集大小不能被批大小(batch_size)整除，则设置为True以删除最后一个不完整的批。如果为False，并且数据集的大小不能被批大小整除，则最后一批将更小。（默认值：False）
    dataloader = DataLoader(
        dataset, batch_size,
        shuffle=(mode == 'train'), drop_last=False,
        num_workers=n_jobs, pin_memory=True)                            # 构造数据加载器
    return dataloader

6.定义深度神经网络“DNN”

class NeuralNet(nn.Module):
    ''' 一个简单的深度神经网络(均由全连接层构成)'''
    # torch.nn名称空间提供了构建自己的神经网络所需的所有构建块。
    # PyTorch中的有关神经网络每个模块都是nn.module的子类。
    # 神经网络本身就是一个由其他模块（层）组成的模块。
    # 这种嵌套结构允许轻松构建和管理复杂的体系结构。
    def __init__(self, input_dim):
        # 每个继承nn.Module的子类都需要在forward()方法中实现对于输入数据的操作。
        super(NeuralNet, self).__init__()
        # Sequential是模块的有序容器。
        # 数据以定义的相同顺序通过所有模块。您可以使用顺序容器快速的组合一个网络。
        self.net = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 1)
        )

        # 定义损失函数
        # reduction(可选):指定要应用于输出的缩减："none"|"mean"|"sum"
        	# "none"：不应用缩减
        	# "mean"：输出的总和将除以输出中的元素数
        	# "sum" : 将对输出进行求和。(默认值)
        self.criterion = nn.MSELoss(reduction='mean')

    def forward(self, x):
        ''' 定义前向传递函数'''
        # .squeeze() 返回一个删除输入张量中维度大小为1的张量
        return self.net(x).squeeze(1)

    def cal_loss(self, pred, target):
        ''' 定义损失计算函数 '''
        return self.criterion(pred, target)

7.定义训练函数

def train(tr_set, dv_set, model, config, device):
    '''训练 DNN '''
    n_epochs = config['n_epochs']  # 设置最大的训练轮次
    # 设置优化器
    optimizer = getattr(torch.optim, config['optimizer'])(
        model.parameters(), **config['optim_hparas'])
    min_mse = 1000.
    loss_record = {'train': [], 'dev': []}      # 记录训练过程中的损失值
    early_stop_cnt = 0
    epoch = 0
    while epoch < n_epochs:
        model.train()                           # 将模型设置为训练模式
        for x, y in tr_set:                     # 开始迭代数据加载器
            optimizer.zero_grad()               # 设置梯度值为0
            x, y = x.to(device), y.to(device)   # 将数据转移到device中，加速运算
            pred = model(x)                     # 前向传播
            mse_loss = model.cal_loss(pred, y)  # 计算损失
            mse_loss.backward()                 # 启动反向传播，计算梯度
            optimizer.step()                    # 利用计算出来的梯度，更新参数
            loss_record['train'].append(mse_loss.detach().cpu().item())

        # 在完成一个epoch的训练后，在验证集上测试模型的效果
        dev_mse = dev(dv_set, model, device)
        if dev_mse < min_mse:
            # 若在验证集上得到更好的效果，则及时保存模型的参数
            min_mse = dev_mse
            print('Saving model (epoch = {:4d}, loss = {:.4f})'
                .format(epoch + 1, min_mse))
            torch.save(model.state_dict(), config['save_path'])  
            early_stop_cnt = 0
        else:
            early_stop_cnt += 1							# 统计模型效果连续不变好的次数

        epoch += 1
        loss_record['dev'].append(dev_mse)
        if early_stop_cnt > config['early_stop']:
            # 如果模型连续不变好的次数大于预设值，则停止训练
            # 这往往代表模型已经不能够训练得到更好的结果，及时停止训练是一个较好的策略
            break

    print('Finished training after {} epochs'.format(epoch))
    return min_mse, loss_record

8.定义验证函数

def dev(dv_set, model, device):
    model.eval()                                # 设置模型为测试模式
    total_loss = 0
    for x, y in dv_set:                         
        x, y = x.to(device), y.to(device)       
        with torch.no_grad():                   # 取消梯度计算(加快运行速度)
            pred = model(x)                     # 前向计算
            mse_loss = model.cal_loss(pred, y)  # 计算损失
        total_loss += mse_loss.detach().cpu().item() * len(x)  # 累加损失值
    total_loss = total_loss / len(dv_set.dataset)              # 计算平均的损失值

    return total_loss

9.定义测试函数

def test(tt_set, model, device):
    model.eval()                                
    preds = []
    for x in tt_set:                            
        x = x.to(device)                        
        with torch.no_grad():                   
            pred = model(x)                     
            preds.append(pred.detach().cpu())   # 记录每一批数据的预测值
    preds = torch.cat(preds, dim=0).numpy()     # 融合每一批数据的预测值，并将其转化为numpy数据
    return preds

10.配置超参数“hyperparameters”

device = get_device()                 # 检验当前的硬件环境(CPU/GPU)
os.makedirs('models', exist_ok=True)  # 创建存储模型的文件夹
target_only = False                   # 使用默认的特征(在本文中，即使用全部93个特征)

config = {
    'n_epochs': 3000,                # 设置最大的训练批次
    'batch_size': 270,               # 设置数据加载器的batch_size，即每次迭代返回的数据数量
    'optimizer': 'SGD',              # 优化算法(optimizer in torch.optim)
    'optim_hparas': {                # 优化器的超参数设置 (具体取决于你所调用的优化器)
        'lr': 0.001,                 # SGD算法的学习率
        'momentum': 0.9              # SGD算法的momentum
    },
    'early_stop': 200,               # 提前结束批次(距离你模型性能上一次提升的批次数)
    'save_path': 'models/model.pth'  # 模型保存路径
}

11.创建数据加载器

tr_set = prep_dataloader(tr_path, 'train', config['batch_size'], target_only=target_only)
dv_set = prep_dataloader(tr_path, 'dev', config['batch_size'], target_only=target_only)
tt_set = prep_dataloader(tt_path, 'test', config['batch_size'], target_only=target_only)

12.创建DNN

model = NeuralNet(tr_set.dataset.dim).to(device)  # 构造DNN，并导入GPU

13.调用训练函数，开始训练

model_loss, model_loss_record = train(tr_set, dv_set, model, config, device)

14.绘制学习图

plot_learning_curve(model_loss_record, title='deep model')

15.保存模型、验证模型

del model
model = NeuralNet(tr_set.dataset.dim).to(device)
ckpt = torch.load(config['save_path'], map_location='cpu')  # 载入先前保存的模型参数
model.load_state_dict(ckpt)
plot_pred(dv_set, model, device)  

16.保存结果

def save_pred(preds, file):
    ''' Save predictions to specified file '''
    print('Saving results to {}'.format(file))
    with open(file, 'w') as fp:
        writer = csv.writer(fp)
        writer.writerow(['id', 'tested_positive'])
        for i, p in enumerate(preds):
            writer.writerow([i, p])

preds = test(tt_set, model, device)  # 预测测试集的结果
save_pred(preds, 'pred.csv')         # 保存预测结果到 pred.csv

16.参考文档

• 知乎：torch.backends.cudnn.benchmark ?!