分布式加载数据训练神经网络

训练

import multiprocessing.managers
from multiprocessing import Queuetask_queue = Queue()def return_task():return task_queueclass QueueManager(multiprocessing.managers.BaseManager):passimport torch
import torch.nn as nnimport librosa
import numpy as npclass ParallelModel(nn.Module):def __init__(self, num_emotions):super(ParallelModel, self).__init__()# conv blockself.conv2Dblock = nn.Sequential(# 1. conv blocknn.Conv2d(in_channels=1,out_channels=16,kernel_size=3,stride=1,padding=1),nn.BatchNorm2d(16),nn.ReLU(),nn.MaxPool2d(kernel_size=2, stride=2),nn.Dropout(p=0.3),# 2. conv blocknn.Conv2d(in_channels=16,out_channels=32,kernel_size=3,stride=1,padding=1),nn.BatchNorm2d(32),nn.ReLU(),nn.MaxPool2d(kernel_size=4, stride=4),nn.Dropout(p=0.3),# 3. conv blocknn.Conv2d(in_channels=32,out_channels=64,kernel_size=3,stride=1,padding=1),nn.BatchNorm2d(64),nn.ReLU(),nn.MaxPool2d(kernel_size=4, stride=4),nn.Dropout(p=0.3),# 4. conv blocknn.Conv2d(in_channels=64,out_channels=64,kernel_size=3,stride=1,padding=1),nn.BatchNorm2d(64),nn.ReLU(),nn.MaxPool2d(kernel_size=4, stride=4),nn.Dropout(p=0.3))# Transformer blockself.transf_maxpool = nn.MaxPool2d(kernel_size=[2, 4], stride=[2, 4])transf_layer = nn.TransformerEncoderLayer(d_model=64, nhead=4, dim_feedforward=512, dropout=0.4,activation='relu')self.transf_encoder = nn.TransformerEncoder(transf_layer, num_layers=4)# Linear softmax layerself.out_linear = nn.Linear(320, num_emotions)self.dropout_linear = nn.Dropout(p=0)self.out_softmax = nn.Softmax(dim=1)def forward(self, x):# conv embeddingconv_embedding = self.conv2Dblock(x)  # (b,channel,freq,time)conv_embedding = torch.flatten(conv_embedding, start_dim=1)  # do not flatten batch dimension# transformer embeddingx_reduced = self.transf_maxpool(x)x_reduced = torch.squeeze(x_reduced, 1)x_reduced = x_reduced.permute(2, 0, 1)  # requires shape = (time,batch,embedding)transf_out = self.transf_encoder(x_reduced)transf_embedding = torch.mean(transf_out, dim=0)# concatenatecomplete_embedding = torch.cat([conv_embedding, transf_embedding], dim=1)# final Linearoutput_logits = self.out_linear(complete_embedding)output_logits = self.dropout_linear(output_logits)output_softmax = self.out_softmax(output_logits)return output_logits, output_softmaxdef getMELspectrogram(audio, sample_rate):mel_spec = librosa.feature.melspectrogram(y=audio,sr=sample_rate,n_fft=1024,win_length=512,window='hamming',hop_length=256,n_mels=128,fmax=sample_rate / 2)mel_spec_db = librosa.power_to_db(mel_spec, ref=np.max)return mel_spec_dbif __name__ == "__main__":model = ParallelModel(2)model.train()model.cuda()loss_f = torch.nn.CrossEntropyLoss()# 暂时是测试训练且不适用动态lr和正则化optime = torch.optim.Adam(model.parameters(), lr=0.001, weight_decay=0.0001)# 生成训练数据块每次训练的时候进行torch.load 每次生成后要将下面两句话注释掉# data_block_name=gen_epocch_batch_sizes(8,64)# torch.save({"data_block_name":data_block_name},"data_block_name")# 加入动态学习率加入logs后期方便可视化 加入验证和测试代码scheduler_exp = torch.optim.lr_scheduler.ExponentialLR(optime, 0.95)avg_loss_ = []avg_acc_ = []# 开启分布式支持multiprocessing.freeze_support()# 注册可以访问队列并得到结果的函数QueueManager.register('get_task', callable=return_task)# QueueManager.register('get_result')manager = QueueManager(address=('192.168.0.106', 8888), authkey='password'.encode('utf-8'))manager.start()task = manager.get_task()# result=manager.get_result()epoch_b=0while epoch_b<12344:epoch_b+=1# 从队列中取出任务# 训练代码放这里data = task.get()for data, label in data:out = model(torch.Tensor(data).cuda())loss = loss_f(out[0], torch.Tensor(label).long().cuda())avg_loss_.append(loss)avg_acc_.append(sum(torch.max(out[1].cpu(), -1)[1] == torch.Tensor(label)) / 64)optime.zero_grad()loss.backward()optime.step()print("\r  avg_loss {} avg_acc {} ".format( sum(avg_loss_) / len(avg_loss_),sum(avg_acc_) / len(avg_acc_)), end="")torch.save(model.state_dict(), "best.pth")print()scheduler_exp.step()print('client_get', len(data))

数据加载


import multiprocessing.managersfrom multiprocessing import Queueimport librosa
import numpy as np
# 任务队列class QueueManager(multiprocessing.managers.BaseManager):passdef get_pcen(fpath):y, sr = librosa.load(fpath, sr=16000)S = librosa.feature.melspectrogram(y, sr=sr, power=1, n_fft=1024, hop_length=512, n_mels=128)pcen_S = librosa.pcen(S).Tlog_S = librosa.amplitude_to_db(S, ref=np.max)return pcen_S  ,log_S
def read_label():root_path = "D:/110time/"# print(s)with open(root_path + "zong_new.txt", "r", encoding="utf-8") as f:data = f.readlines()path_list_wav = [root_path + i.split("\t")[0] for i in data if "non" not in i]path_list_lab = [i.split("\t")[1].strip().replace("/noise", "") for i in data if "non" not in i]path_list_lab_bin = [1 if "就" in i else 0 for i in path_list_lab]voc_n_d = {i: d for i, d in enumerate(sorted(set("".join(path_list_lab))))}voc_d_n = {d: i for i, d in enumerate(sorted(set("".join(path_list_lab))))}path_list_lab_n = [[voc_d_n[i_] for i_ in i] for i in path_list_lab]max_l_len = max([len(i) for i in path_list_lab])return path_list_wav,path_list_lab_bin# 将数据padding
def gen_pcen_padding(batch_size=32,log_flog=False):path,label=read_label()epoch_batch_size=[]epoch_batch_size_label=[]train_data=[]for i,j in zip(path,label):train_data.append((i,j))# 打乱np.random.shuffle(train_data)for i,j in train_data:try:data = get_pcen(i)except:continuepadding_data = np.zeros([ 1, 128,512])# print("\r {} ".format(i),end="")if log_flog:padding_data_log = padding_data.copy()padding_data_log[:,  :, :data[0].T.shape[1]] = data[0].Tepoch_batch_size.append(padding_data_log)epoch_batch_size_label.append(j)else:padding_data[:, :, :data[1].shape[1]] = data[1]epoch_batch_size.append(padding_data)epoch_batch_size_label.append(j)if len(epoch_batch_size)==batch_size:yield np.stack(epoch_batch_size,axis=0),epoch_batch_size_labelepoch_batch_size=[]epoch_batch_size_label = []return np.stack(epoch_batch_size,axis=0),epoch_batch_size_label
# 将一定量的batch数据提前使用torch.save 存在本地数量不要太多要不然内存不够
def gen_epocch_batch_sizes(epoch_batch_size,batch_size):# 这里采取512条数据进行存储一次也就是16个32epoch_batch_size_data=[]name_=0name_list=[]for i in gen_pcen_padding(batch_size=batch_size):name_+=1epoch_batch_size_data.append(i)if len(epoch_batch_size_data)==epoch_batch_size:name_list.append("epoch_batch_{}.pth".format(name_))# print("gen_block")print("gen data_block {} ".format(name_))# torch.save({"data":epoch_batch_size_data},"epoch_batch_{}.pth".format(name_))yield epoch_batch_size_dataepoch_batch_size_data=[]if __name__ == "__main__":# 开启分布式支持multiprocessing.freeze_support()# 注册可以访问队列并得到结果的函数QueueManager.register('get_task')# QueueManager.register('get_result', callable=return_result)manager = QueueManager(address=('192.168.0.106', 8888), authkey='password'.encode('utf-8'))manager.connect()# 任务队列task = manager.get_task()# 结果队列# result = manager.get_result()# 小于m 个 batchsize 就开始放入输入  大于n个batchsize 就停止放入data_block_name = gen_epocch_batch_sizes(32, 64)# 这里大小看内存大小while task.qsize()<10:print(task.qsize())for i in data_block_name:task.put(i)while task.qsize()>=9:print(task.qsize())# for i in range(1000):#     # 将结果从队列中取出#     res = result.get()#     print('get_data', res)# manager.shutdown()

分布式加载数据训练神经网络相关推荐

C++ 和 OpenCV 实现卷积神经网络并加载 Keras 训练好的参数进行预测
C++ 和 OpenCV 实现卷积神经网络并加载 Keras 训练好的参数进行预测一. 背景二. Keras 定义神经网络结构 channels_first 与 channels_last cha ...
PyTorch训练中Dataset多线程加载数据，比Dataloader里设置多个workers还要快
PyTorch训练中Dataset多线程加载数据,而不是在DataLoader 背景与需求现在做深度学习的越来越多人都有用PyTorch,他容易上手,而且API相对TF友好的不要太多.今天就给大家带 ...
【PyTorch训练中Dataset多线程加载数据，比Dataloader里设置多个workers还要快】
文章目录一.引言二.背景与需求三.方法的实现四.代码与数据测试五.测试结果 5.1.Max elapse 5.2.Multi Load Max elapse 5.3.Min elapse 5 ...
在NVIDIA A100 GPU中使用DALI和新的硬件JPEG解码器快速加载数据
在NVIDIA A100 GPU中使用DALI和新的硬件JPEG解码器快速加载数据如今,最流行的拍照设备智能手机可以捕获高达4K UHD的图像(3840×2160图像),原始数据超过25 MB.即使 ...
mxnet.gluon 加载预训练
import mxnet as mx from mxnet.gluon import nn from mxnet import gluon,nd,autograd,init from mxnet.gl ...
在TensorFlow中使用pipeline加载数据
正文共2028个字,6张图,预计阅读时间6分钟. 前面对TensorFlow的多线程做了测试,接下来就利用多线程和Queue pipeline地加载数据.数据流如下图所示: 首先,A.B.C三个文件通 ...
PyTorch 加载预训练权重
前言使用PyTorch官方提供的权重或者其他第三方提供的权重对相同模型的参数进行初始化,在数据量较少的前提下,可以帮助模型更快地收敛到最优点,达到更好的效果,即迁移学习. 在大部分的迁移学习场景 ...
TensorFlow加载数据的方式
tensorflow作为符号编程框架,需要先构建数据流图,再读取数据,然后再进行训练.tensorflow提供了以下三种方式来加载数据: 预加载数据(preloaded data):在tensorfl ...
7.3 TensorFlow笔记(基础篇):加载数据之从队列中读取
前言整体步骤在TensorFlow中进行模型训练时,在官网给出的三种读取方式,中最好的文件读取方式就是将利用队列进行文件读取,而且步骤有两步: 1. 把样本数据写入TFRecords二进制文件 2 ...

分布式加载数据训练神经网络

分布式加载数据训练神经网络相关推荐

最新文章

热门文章