• pytorch Nvidia 数据预处理加速

    目录

    安装 不支持Windows：

    官方说明：

    预处理加速：

    学习笔记：

    参考：

    深度学习预处理工具---DALI详解_nvidia.dali.fn_扫地的小何尚的博客-CSDN博客

    安装 不支持Windows：

    官方说明：

    Installation — NVIDIA DALI 1.30.0 documentation

    pip install nvidia-pyindex
    pip install nvidia-dali-cuda110


    import nvidia.dali.ops
    import nvidia.dali.types
     
    from nvidia.dali.pipeline import Pipeline
    from nvidia.dali.plugin.pytorch import DALIGenericIterator
     

    官网下载地址：看起来么有windows版本，

    Index of /compute/redist///nvidia-dali-cuda110

    预处理加速：

    Nvidia Dali： 强大的数据增强库_笔记大全_设计学院

    学习笔记：

    对于深度学习任务,训练速度决定了模型的迭代速度，而训练速度又取决于数据预处理和网络的前向和后向耗时。
    对于识别任务，batch size通常较大，并且需要做数据增强，因此常常导致训练速度的瓶颈在数据读取和预处理上，尤其对于小网络而言。
    对于数据读取耗时的提升，粗暴且有效的解决办法是使用固态硬盘，或者将数据直接拷贝至/tmp文件夹（内存空间换时间）。
    对于数据预处理的耗时，则可以通过使用Nvidia官方开发的Dali预处理加速工具包，将预处理放在cpu/gpu上进行加速。pytorch1.6版本内置了Dali，无需自己安装。

    官方的Dali交程较为简单，实际训练通常要根据任务需要自定义Dataloader，并于分布式训练结合使用。这里将展示一个使用Dali定义DataLoader的例子，功能是返回序列图像，并对序列图像做常见的统一预处理操作。
    `

    1. from nvidia.dali.plugin.pytorch import DALIGenericIterator
    2.  
    3. from nvidia.dali.types import DALIImageType
    4. import cv2
    5. from nvidia.dali.plugin.pytorch import DALIClassificationIterator
    6. from nvidia.dali.pipeline import Pipeline
    7. import nvidia.dali.ops as ops
    8. import nvidia.dali.types as types
    9. from sklearn.utils import shuffle
    10. import numpy as np
    11. from torchvision import transforms
    12. import torch.utils.data as torchdata
    13. import random
    14. from pathlib import Path
    15. import torch
    16.  
    17. class TRAIN_INPUT_ITER(object):
    18.     def __init__(self, batch_size, num_class,seq_len, sample_rate, num_shards=1, shard_id=0,root_dir=Path('') ,list_file='', is_training=True):
    19.         self.batch_size = batch_size
    20.         self.num_class = num_class
    21.         self.seq_len = seq_len
    22.         self.sample_rate = sample_rate
    23.         self.num_shards = num_shards
    24.         self.shard_id = shard_id
    25.         self.train = is_training
    26.         self.image_name_formatter = lambda x: f'image_{x:05d}.jpg'
    27.         self.root_dir = root_dir
    28.         with open(list_file,'r') as f:
    29.             self.ori_lines = f.readlines()
    30.  
    31.     def __iter__(self):
    32.         self.i = 0
    33.         bucket = len(self.ori_lines)//self.num_shards
    34.         self.n = bucket
    35.         return self
    36.  
    37.     def __next__(self):
    38.         batch = [[] for _ in range(self.seq_len)]
    39.         labels = []
    40.         for _ in range(self.batch_size):
    41.             # self.sample_rate = random.randint(1,2)
    42.             if self.train and self.i % self.n == 0:
    43.                 bucket = len(self.ori_lines)//self.num_shards
    44.                 self.ori_lines= shuffle(self.ori_lines, random_state=0)
    45.                 self.lines = self.ori_lines[self.shard_id*bucket:(self.shard_id+1)*bucket]
    46.             line = self.lines[self.i].strip()
    47.             dir_name,start_f,end_f, label = line.split(' ')
    48.             start_f = int(start_f)
    49.             end_f = int(end_f)
    50.             label = int(label)
    51.             begin_frame = random.randint(start_f,max(end_f-self.sample_rate*self.seq_len,start_f))
    52.             begin_frame = max(1,begin_frame)
    53.             last_frame = None
    54.             for k in range(self.seq_len):
    55.                 filename = self.root_dir/dir_name/self.image_name_formatter(begin_frame+self.sample_rate*k)
    56.                 if filename.exists():
    57.                     f = open(filename,'rb')
    58.                     last_frame = filename
    59.                 elif last_frame is not None:
    60.                     f = open(last_frame,'rb')
    61.                 else:
    62.                     print('{} does not exist'.format(filename))
    63.                     raise IOError
    64.                 batch[k].append(np.frombuffer(f.read(), dtype = np.uint8))
    65.             if random.randint(0,1)%2 == 0:
    66.                 end_frame = start_f + random.randint(0,self.sample_rate*self.seq_len//2)
    67.                 begin_frame = max(1,end_frame-self.sample_rate*self.seq_len)
    68.             else:
    69.                 begin_frame = end_f - random.randint(0,self.sample_rate*self.seq_len//2)
    70.                 begin_frame = max(1,begin_frame)
    71.                 end_frame = begin_frame + self.sample_rate*self.seq_len
    72.             last_frame = None
    73.             for k in range(self.seq_len):
    74.                 filename = self.root_dir/dir_name/self.image_name_formatter(begin_frame+self.sample_rate*k)
    75.                 if filename.exists():
    76.                     f = open(filename,'rb')
    77.                     last_frame = filename
    78.                 elif last_frame is not None:
    79.                     f = open(last_frame,'rb')
    80.                 else:
    81.                     print('{} does not exist'.format(filename))
    82.                     raise IOError
    83.                 batch[k].append(np.frombuffer(f.read(), dtype = np.uint8))
    84.  
    85.             labels.append(np.array([label], dtype = np.uint8))
    86.             if label==8 or label == 9:
    87.                 labels.append(np.array([label], dtype = np.uint8))
    88.             else:
    89.                 labels.append(np.array([self.num_class-1], dtype = np.uint8))
    90.  
    91.             self.i = (self.i + 1) % self.n
    92.         return (batch, labels)
    93.     
    94.     next = __next__
    95.  
    96.  
    97. class VAL_INPUT_ITER(object):
    98.     def __init__(self, batch_size, num_class,seq_len, sample_rate, num_shards=1, shard_id=0,root_dir=Path('') ,list_file='', is_training=False):
    99.         self.batch_size = batch_size
    100.         self.num_class = num_class
    101.         self.seq_len = seq_len
    102.         self.sample_rate = sample_rate
    103.         self.num_shards = num_shards
    104.         self.shard_id = shard_id
    105.         self.train = is_training
    106.         self.image_name_formatter = lambda x: f'image_{x:05d}.jpg'
    107.         self.root_dir = root_dir
    108.         with open(list_file,'r') as f:
    109.             self.ori_lines = f.readlines()
    110.             self.ori_lines= shuffle(self.ori_lines, random_state=0)
    111.  
    112.     def __iter__(self):
    113.         self.i = 0
    114.         bucket= len(self.ori_lines)//self.num_shards
    115.         self.n = bucket
    116.         return self
    117.  
    118.     def __next__(self):
    119.         batch = [[] for _ in range(self.seq_len)]
    120.         labels = []
    121.         for _ in range(self.batch_size):
    122.             # self.sample_rate = random.randint(1,2)
    123.             if self.train and self.i % self.n == 0:
    124.                 bucket = len(self.ori_lines)//self.num_shards
    125.                 self.ori_lines= shuffle(self.ori_lines, random_state=0)
    126.                 self.lines = self.ori_lines[self.shard_id*bucket:(self.shard_id+1)*bucket]
    127.             if self.i % self.n == 0:
    128.                 bucket = len(self.ori_lines)//self.num_shards
    129.                 self.lines = self.ori_lines[self.shard_id*bucket:(self.shard_id+1)*bucket]
    130.             line = self.lines[self.i].strip()
    131.             dir_name,start_f,end_f, label = line.split(' ')
    132.             start_f = int(start_f)
    133.             end_f = int(end_f)
    134.             label = int(label)
    135.             begin_frame = random.randint(start_f,max(end_f-self.sample_rate*self.seq_len,start_f))
    136.             begin_frame = max(1,begin_frame)
    137.             last_frame = None
    138.             for k in range(self.seq_len):
    139.                 filename = self.root_dir/dir_name/self.image_name_formatter(begin_frame+self.sample_rate*k)
    140.                 if filename.exists():
    141.                     f = open(filename,'rb')
    142.                     last_frame = filename
    143.                 elif last_frame is not None:
    144.                     f = open(last_frame,'rb')
    145.                 else:
    146.                     print('{} does not exist'.format(filename))
    147.                     raise IOError
    148.                 batch[k].append(np.frombuffer(f.read(), dtype = np.uint8))
    149.             labels.append(np.array([label], dtype = np.uint8))
    150.             self.i = (self.i + 1) % self.n
    151.         return (batch, labels)
    152.     next = __next__
    153.  
    154. class HybridPipe(Pipeline):
    155.     def __init__(self, batch_size, num_class,seq_len, sample_rate, num_shards,shard_id,root_dir, list_file, num_threads, device_id=0, dali_cpu=True,size = (224,224),is_gray = True,is_training = True):
    156.         super(HybridPipe, self).__init__(batch_size, num_threads, device_id, seed=12 + device_id)
    157.         if is_training:
    158.             self.external_data = TRAIN_INPUT_ITER(batch_size//2, num_class,seq_len,sample_rate,num_shards,shard_id,root_dir, list_file,is_training)
    159.         else:
    160.             self.external_data = VAL_INPUT_ITER(batch_size, num_class,seq_len,sample_rate,num_shards,shard_id,root_dir, list_file,is_training)
    161.         # self.external_data = VAL_INPUT_ITER(batch_size, num_class,seq_len,sample_rate,num_shards,shard_id,root_dir, list_file,is_training)
    162.         self.seq_len = seq_len
    163.         self.training = is_training
    164.         self.iterator = iter(self.external_data)
    165.         self.inputs = [ops.ExternalSource() for _ in range(seq_len)]
    166.         self.input_labels = ops.ExternalSource()
    167.         self.is_gray = is_gray
    168.  
    169.         decoder_device = 'cpu' if dali_cpu else 'mixed'
    170.  
    171.         self.decode = ops.ImageDecoder(device=decoder_device, output_type=types.RGB)
    172.         if self.is_gray:
    173.             self.space_converter = ops.ColorSpaceConversion(device='gpu',image_type=types.RGB,output_type=types.GRAY)
    174.         self.resize = ops.Resize(device='gpu', size=size)
    175.         self.cast_fp32 = ops.Cast(device='gpu',dtype = types.FLOAT)
    176.         if self.training:
    177.             self.crop_coin = ops.CoinFlip(probability=0.5)
    178.             self.crop_pos_x = ops.Uniform(range=(0., 1.))
    179.             self.crop_pos_y = ops.Uniform(range=(0., 1.))
    180.             self.crop_h = ops.Uniform(range=(256*0.85,256))
    181.             self.crop_w = ops.Uniform(range=(256*0.85,256))
    182.             self.crmn = ops.CropMirrorNormalize(device="gpu",output_layout=types.NHWC)
    183.  
    184.             self.u_rotate = ops.Uniform(range=(-8, 8))
    185.             self.rotate = ops.Rotate(device='gpu',keep_size=True)
    186.  
    187.             self.brightness = ops.Uniform(range=(0.9,1.1))
    188.             self.contrast = ops.Uniform(range=(0.9,1.1))
    189.             self.saturation = ops.Uniform(range=(0.9,1.1))
    190.             self.hue = ops.Uniform(range=(-0.3,0.3))
    191.             self.color_jitter = ops.ColorTwist(device='gpu')
    192.         else:
    193.             self.crmn = ops.CropMirrorNormalize(device="gpu",crop=(224,224),output_layout=types.NHWC)
    194.     
    195.  
    196.     def define_graph(self):
    197.         self.batch_data = [i() for i in self.inputs]
    198.         self.labels = self.input_labels()
    199.         out = self.decode(self.batch_data)
    200.         out = [out_elem.gpu() for out_elem in out]
    201.         if self.training:
    202.             out = self.color_jitter(out,brightness=self.brightness(),contrast=self.contrast())
    203.         if self.is_gray:
    204.             out = self.space_converter(out)
    205.         if self.training:
    206.             out = self.rotate(out,angle=self.u_rotate())
    207.             out = self.crmn(out,crop_h=self.crop_h(),crop_w=self.crop_w(),crop_pos_x=self.crop_pos_x(),crop_pos_y=self.crop_pos_y(),mirror=self.crop_coin())
    208.         else:
    209.             out = self.crmn(out)
    210.         out = self.resize(out)
    211.         if not self.training:
    212.             out = self.cast_fp32(out)
    213.         return (*out, self.labels)
    214.     
    215.     def iter_setup(self):
    216.         try:
    217.             (batch_data, labels) = self.iterator.next()
    218.             for i in range(self.seq_len):
    219.                 self.feed_input(self.batch_data[i], batch_data[i])
    220.             self.feed_input(self.labels, labels)
    221.  
    222.         except StopIteration:
    223.             self.iterator = iter(self.external_data)
    224.             raise StopIteration
    225.  
    226. def dali_loader(batch_size,
    227.                 num_class,
    228.                 seq_len,
    229.                 sample_rate,
    230.                 num_shards,
    231.                 shard_id,
    232.                 root_dir,
    233.                 list_file,
    234.                 num_workers,
    235.                 device_id,
    236.                 dali_cpu=True,
    237.                 size = (224,224),
    238.                 is_gray = True,
    239.                 is_training=True):
    240.     print('##########',root_dir)
    241.     pipe = HybridPipe(batch_size,num_class,seq_len,sample_rate,num_shards,shard_id,root_dir,
    242.                         list_file,num_workers,device_id=device_id,
    243.                         dali_cpu=dali_cpu,size = size,is_gray=is_gray,is_training=is_training)
    244.     # pipe.build()
    245.     names = []
    246.     for i in range(seq_len):
    247.         names.append(f'data{i}')
    248.     names.append('label')
    249.     print('##############',names)
    250.     loader = DALIGenericIterator(pipe,names,pipe.external_data.n,last_batch_padded=True, fill_last_batch=True)
    251.     return loade
  • 相关阅读:
    如何查询IP地址的位置?
    如何搭建一部引人入胜的短剧小程序
    【Linux】进程控制基础知识
    Spring注解驱动之声明式事务源码分析
    408真题-2021
    【第3章】MyBatis-Plus持久层接口之Service Interface(上)
    优先级队列（堆）【Java】
    [补题记录] Complete the Permutation（贪心、set）
    金仓数据库 KingbaseES 插件参考手册(17. dbms_metadata)
    1459. 矩形面积
  • 原文地址：https://blog.csdn.net/jacke121/article/details/133692941