【MindSpore易点通】数据处理经验总结

多进程转MindRecord

背景信息

MindRecord数据在MindSpore中读取性能更优，推荐用户将其他格式的数据集转换为MindRecord格式。

经验总结

利用multiprocessing.Pool接口，实现多进程运行。将Cifar10图片格式数据集，转换成MindRecord格式，并进行性能对比。与单进程方式相比，多进程中需进行如下配置（具体差异详见代码附件）：

total_data_length = len(os.listdir(os.path.join(argv.data_define_path,
 
                            argv.path)))
 
part_num = math.ceil(total_data_length / argv.multi_num)
 
left = total_data_length % part_numif left == 0:
 
    left = part_num
 
total_index_with_length = []for i in range(argv.multi_num - 1):
 
    total_index_with_length.append([i * part_num, part_num])
 
total_index_with_length.append([(argv.multi_num - 1) * part_num, left])
 
pool_list = []for i, index_with_length in enumerate(total_index_with_length):
 
    pool_list.append((argv, index_with_length[0], index_with_length[1], i))with Pool(processes=argv.multi_num) as pool:
 
    pool.map(process_single_writer, pool_list)

其中argv.multi_num为进程数；process_single_writer和单进程中的转换方式相同；pool_list为每个进程传入参数。

注：当数据集较大时，推荐在训练环境中利用训练环境机器的本地SSD（开发环境中使用EFS会影响数据转换性能），并选用8卡规格进行转换（若选用1卡的规格，CPU/内存/SSD空间（一共3.3T） 都只会被分到1/8 ）。需要注意的是8卡任务会起8次脚本导致重复转换，可以通过RANK_ID来控制，使仅有一个脚本正常执行；

性能对比：多进程转MindRecord: 8886.35imgs/sec ； 单进程转MindRecord：2133.79imgs/sec

一个Epoch最后一个Batch的数据不足

背景信息

一个Epoch的迭代次数=图片总数/Batch Size，如果不能整除，就会出现最后一个Batch的数据不足。目前如果Batch Size数据不足会报错。

经验总结

如果迭代一个Epoch的末尾训练报错结束，可以检查是否该问题导致的，可以通过如下配置规避drop_remainder=true

代码示例如下：

import mindspore.dataset as ds
# data is an instance of Dataset object
# declare an apply_func function which returns a Dataset objectdef apply_func(ds):
 
    ds = ds.batch(2, drop_remainder=true)
 
    return ds# use apply to call apply_func
 
data = data.apply(apply_func)

TFRecord数据读取

背景信息

TFRecord数据文件是一种将图像数据和标签统一存储的二进制文件，能更好的利用内存，在TensorFlow中快速的复制，移动，读取，存储等。

经验总结

1. TFRecord的读取

FP32图片数据使用tostring保存为TFRecord，若直接使用TFRecordDataset读取为UINT8格式，而MindSpore中暂时缺少tf.decode_raw这样类似功能的算子，需要自己手动进行数据转化，转化代码如下：

def trans_dtype(data, target):
 
    trans_to_float32 = lambda x: np.frombuffer(np.ndarray.tobytes(x), dtype=np.float32)
 
    input_size = 128
 
    data = np.reshape(trans_to_float32(data), (input_size, input_size, 10))
 
    target = np.reshape(trans_to_float32(target), (input_size, input_size, 10))
 
    return data, target
 
 
 
tfdataset = tfdataset.map(input_columns=['data', 'target'], operations=trans_dtype)

1. 自动生成Schema与预先定义Schema

训练数据含有两批不同大小的数据（Key值不同），如A中包含{a,b,c}3种Key及其值，B种包含{a,b,c,d}4种Key及其值，目前训练任务中只需要两批数据中{a,b}2种Key及其值即可。 以读取TFRecord为例：

DATA #A+B

tfdataset = de.TFRecordDataset(dataset_files=DATA) # (1)

DATA #B+A

tfdataset = de.TFRecordDataset(dataset_files=DATA) # (2)

以上两种写法中，(2)写法会报错。其关键在于DATA的构造，在读取数据过程中，如果不进行Schema的定义，MindSpore会在读取第一个数据的时候根据读取到的数据本身自行定义Schema。所以如果先读取B类数据，自动生成的Schema会包含{a,b,c,d}4个Key，而之后读取A类数据的时候得不到key=d的值，此时便会报错。所以较为恰当的做法是事先在程序中定义好Schema，如：

DATA # B+A或A+B

schema = de.Schema()
 
schema.add_column('a')
 
schema.add_column('b')
 
tfdataset = de.TFRecordDataset(dataset_files=DATA, schema=schema)

1. NHWC到NCHW的转化

数据读入时是NHWC的形式，需转化为NCHW的形式送入网络

data = data.transpose(2,0,1)
 
target = target.transpose(2,0,1) # (1)
 
import mindspore.transforms.py_transforms as T
 
transforms = T.ComposeOp([T.HWC2CHW()])
 
tfdataset = tfdataset.map(input_columns='target', operations=())
 
tfdataset = tfdataset.map(input_columns='data', operations=transforms()) # (2)

1. 得到的数据Shape是正确的，但是数据会被打乱，图片写出不正常；(2)使用T.HWC2CHW()则可以得到Shape正确数据顺序正确的图片，图片写出正常。

MindData数据处理流程推荐顺序

背景信息

MindData包含的数据处理操作包括Repeat、Batch、Shuffle、Map和Zip，一般训练过程中都会用到Repeat、Batch、Shuffle和Map的操作。

经验总结

在实际使用过程中，需要组合使用这几个操作时，为达到最优性能，推荐按照如下顺序：数据集加载并shuffle -> Map -> Batch -> Repeat。 原因是shuffle操作需要填充足够数据之后才会往下走，如果是先Load和Map的话，Shuffle很容易受到Map的阻塞导致Pipeline无法并行处理。

示例代码如下：

import mindspore.dataset as dsimport mindspore.dataset.transforms.vision.py_transforms as transforms
 
 
 
DATA_DIR = "custom_dataset_dir/"
 
imagefolder_dataset = ds.ImageFolderDatasetV2(dataset_dir, shuffle=True)
 
resize_op = transforms.Resize(size=(500,500))# map()
 
dataset.map(input_columns="image", operations=resize_op)# batch()
 
dataset = dataset.batch(32, drop_remainder=True)# repeat()
 
dataset = dataset.repeat(10)

数据分布式方式

背景信息

PyTorch的数据分布式方式是ddp（distributeddataparallel）或者dp（dataparallel），而MindSpore的数据分布式方式是ddp（distributeddataparallel）。

经验总结内容

dp和ddp的区别就是数据集是集中在一台机器还是分发到每台机器。差别是用户对batchsize的设置，如果是dp，batchsize=total_batchsize。如果是ddp，batchsize=per_gpu_batch。因此MindSpore的batchsize=per_gpu_batch。