MindSpore数据集加载-GeneratorDataset功能及常见问题

MindSpore支持用户通过自定义的方式构造输入的数据源，然后接入到MindData的流处理流程中，通过迭代该数据源获取数据集进行训练，有点类似PyTorch的DataLoader。

相关的API可以参考：mindspore.dataset.GeneratorDataset

本文主要介绍GeneratorDataset的常用功能，用户常见的问题及解决方法。

###############################################

GeneratorDataset自定义数据集主要可以分为3类：

• 可随机访问的自定义数据集
• 生成器(python generator)式自定义数据集
• 可迭代式的自定义数据集

###############################################

构造可随机访问的数据集

先看一个例子

import numpy as np
import mindspore.dataset as ds
 
class DatasetGenerator:
    def __init__(self):
        self.data = [np.array([i]) for i in range(10)]
 
    def __getitem__(self, item):
        return self.data[item]
 
    def __len__(self):
        return 10
 
dataset = ds.GeneratorDataset(DatasetGenerator(), ["col1"])
for data in dataset.create_dict_iterator(output_numpy=True, num_epochs=1):
    print(data["col1"])

可随机访问，意为数据集本身每一条数据都可以通过索引直接访问。因此构造的数据源本身是持有全量数据集的，如上述DatasetGenerator在__init__方法中自定义了。

常见问题1

• 错误代码：

  class DatasetGenerator:
      def __init__(self):
          self.data = [np.array([i]) for i in range(10)]
   
      def __getitem__(self, item):
          return self.data[item]
  

• 错误提示：

  RuntimeError: Attempt to construct a random access dataset, '__len__' method is required!
  

• 错误分析/修正方法： 错误提示缺少__len__方法，为什么需要这个方法呢？如上述所示，这是一个可随机访问的数据集，因此可以通过随机的索引访问任意数据，所以需要一个索引的范围来确定随机的范围，因此__len__方法是必须的。所以需要为DatasetGenerator加上此方法。

常见问题2

• 错误代码：

  class DatasetGenerator:
      def __init__(self):
          self.data = [np.array([i]) for i in range(10)]
   
      def __getitem__(self, item):
          self.data["data"]
          return self.data[item]
   
      def __len__(self):
          return 10
  

• 错误提示/修正方法：

  RuntimeError: Exception thrown from PyFunc. TypeError: list indices must be integers or slices, not str
  

• 错误分析： 错误提示执行PyFunc的时候出错了（Exception thrown from PyFunc）。一般遇到这种错误，是因为出现了Python语法错误，MindSpore同时会在这句Error后附上抛出的Python异常，比如这里就是TypeError: list indices must be integers or slices, not str。遇到这种情况，第一时间先检查自定义的逻辑是否出现了语法错误，比如检查DatasetGenerator的__getitem__方法，发现对list的访问用了str下标，导致了一个python的异常。一般来说，可以直接定义这个类DatasetGenerator，单独调一下各个方法看是否出现了问题，快速排查。

常见问题3

• 错误代码：

  class DatasetGenerator:
    def __init__(self):
        self.data = [np.array([i]) for i in range(10)]
   
    def __getitem__(self, item):
        return self.data[item]
   
    def __len__(self):
        return 10
   
  dataset = ds.GeneratorDataset(DatasetGenerator(), ["col1", "col2"])
  for data in dataset.create_dict_iterator(output_numpy=True, num_epochs=1):
      print(data["col1"])
  

• 错误提示：

  RuntimeError: Exception thrown from PyFunc. Invalid python function, the 'source' of 'GeneratorDataset' should return same number of NumPy arrays as specified in column_names, the size of column_names is:2 and number of returned NumPy array is:1
  

• 错误分析/修正方法： 错误提示size of column_names is:2 and number of returned NumPy array is:1。显然，2不等于1对吧，再对比一下代码，__getitem__返回的是1个元素，可是GeneratorDataset定义的input_columns参数却是["col1", "col2"]，显然这里会不匹配。

生成器(python generator)式自定义数据集

先看一个例子

import numpy as np
import mindspore.dataset as ds
 
def my_generator(num):
    for i in range(num):
        yield np.array([i])
 
dataset = ds.GeneratorDataset(lambda: my_generator(10), ["col1"])
for data in dataset.create_dict_iterator(output_numpy=True, num_epochs=1):
    print(data["col1"])

所谓生成器，就是指python generator。构造一个生成器传入到GeneratorDataset，MindSpore每次会从生成器中读取一条数据返回。请注意，此时shuffle是失效的，因为GeneratorDataset每次只能获取生成器的下一条数据，且不能知晓究竟有多少条数据，因此无法做到索引上的随机，上述例子无论运行多少遍输出顺序都是一样的。

常见问题1

• 错误代码：

  def my_generator(num):
      for i in range(num):
          yield np.array([i])
   
  dataset = ds.GeneratorDataset(my_generator(10), ["col1"])
  print("get_dataset_size", dataset.get_dataset_size())
  for data in dataset.create_dict_iterator(output_numpy=True, num_epochs=1):
      print("mydata:", data["col1"])
  

• 错误提示： 不是ERROR，但会得到一个WARNING，同时发现mydata及相关的数据打印不出来

  get_dataset_size 10
  [WARNING] ME(24815:140518424966976,MainProcess):2022-06-10-11:38:11.314.445 [mindspore/dataset/engine/iterators.py:143] No records available.
  

• 错误分析/修正方法： python generator本身是特殊的，其本身的数据如果消耗完了，在对其进行访问，会一直得到StopIteration异常，而且不能重头开始生成。因此上述如果传入的是单一个generator实例，会导致其被消耗完了就没了，使得create_dict_iterator的时候再去访问generato会发现没有数据，从而打印出了这个warning。

  修改方法： dataset = ds.GeneratorDataset(lambda: my_generator(10), ["col1"]) 构建一个lambda函数，每次都会生成一个新的generator，从而保证数据可以重新迭代。

可迭代式的自定义数据集

先看一个例子

import numpy as np
import mindspore.dataset as ds
 
class IterableDataset:
    def __init__(self):
        self.count = 0
        self.max = 10
 
    def __iter__(self):
        return self
 
    def __next__(self):
        if self.count >= self.max:
            raise StopIteration
        self.count += 1
        return (np.array(self.count),)
 
dataset = ds.GeneratorDataset(IterableDataset(), ["col1"])
for data in dataset.create_dict_iterator(output_numpy=True, num_epochs=1):
    print(data["col1"])

可迭代式的数据集，其实与上面的生成器式数据集的原理是一样的。只是一个是python generator，一个是iterable class，都可以通过nex()t方法迭代地获取数据。因此这里同样构造一个可迭代类传入到GeneratorDataset，MindSpore每次会从生成器中读取一条数据返回。请注意，此时shuffle是失效的，因为GeneratorDataset每次只能获取生成器的下一条数据，且不能知晓究竟有多少条数据，因此无法做到索引上的随机，上述例子无论运行多少遍输出顺序都是一样的。

常见问题1

• 错误代码：

  class IterableDataset:
    def __init__(self):
        self.count = 0
        self.max = 10
   
    def __next__(self):
        if self.count >= self.max:
            raise StopIteration
        self.count += 1
        return (np.array(self.count),)
   
  dataset = ds.GeneratorDataset(IterableDataset(), ["col1"])
  for data in dataset.create_dict_iterator(output_numpy=True, num_epochs=1):
      print(data["col1"])
  

• 错误提示：

  TypeError: Input `source` function of GeneratorDataset should be callable, iterable or random accessible, commonly it should implement one of the method like yield, __getitem__ or __next__(__iter__).
  复制
  

• 错误分析/修正方法： 错误代码中构建的IterableDataset()其实是有类型缺失的。主要是缺少__getitem__方法或__iter__和__next__方法，从而无法判断此类是属于可随机访问数据的类，还是可迭代访问的类。由于缺失明显的类属性定义，MindSpore无法判断用户的预期行为是什么，具体可以参考这篇教程的样例进一步选择所需要的数据集类：https://www.mindspore.cn/tutorials/zh-CN/r1.7/advanced/dataset/custom.html。

常见问题2

• 错误代码：

  class IterableDataset:
    def __init__(self):
        self.count = 0
        self.max = 10
   
    def __iter__(self):
        return self
   
    def __next__(self):
        self.count += 1
        return (np.array(self.count),)
   
  dataset = ds.GeneratorDataset(IterableDataset(), ["col1"])
  for data in dataset.create_dict_iterator(output_numpy=True, num_epochs=1):
      print(data["col1"])
  

• 错误提示： 无错误提示，但是会发现数据集无限迭代，一直在打印

• 错误分析/修正方法： 可迭代式的数据集类不同于可随机访问数据集类的定义，其没有__len__属性，因此本身是不知道数据集到底迭代到什么时候应该结束。其也不同于python generator，也没有一个具体的数据内容范围信息。因此，想要在某个时间点/数据点结束此类的迭代，我们需要构造一些“异常”作为返回的信号。MindSpore可识别的结束信号，就是StopIteration，这个信号跟python generator的结束信号是一样的（比如一个python generator对象a，对a一直调用next(a)方法，当a数据消耗完了会抛出异常StopIteration）。因此StopIteration可以当成可迭代类数据集迭代结束的信号，从而结束返回。

  修改方法： 在__next__方法中，设定一定的迭代次数条件，然后return StopIteration。