mindspore.dataset.engine.datasets 源代码

# Copyright 2022-2023 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""
1. This file is an abstraction of the dataset loading class. It contains
some basic dataset operations(skip, filter, map, batch, ...).
2. Specific dataset loading classes can be found in datasets_vision.py, datasets_text.py,
datasets_audio.py, datasets_standard_format.py and dataets_user_defined.py files.
    datasets_vision.py: contains vision dataset loading classes.
    datasets_text.py: contains text dataset loading classes.
    datasets_audio.py: contains audio dataset loading classes.
    datasets_standard_format.py: contains standard format loading classes which
                                 any other kinds of datasets can be converted to.
    dataets_user_defined.py: contains basic classes that help users to define
                             flexible ways to load dataset.
"""
import atexit
import glob
import json
import os
import queue
import signal
import stat
import subprocess
import warnings

import gc
import time
import uuid
import multiprocessing
from enum import Enum
from importlib import import_module
import sys
import threading

import copy
import weakref
import platform
import psutil

import mindspore._c_dataengine as cde
from mindspore._c_expression import typing

from mindspore import log as logger
from mindspore.parallel._ps_context import _is_role_pserver, _is_role_sched, _get_ps_context,\
                                           _enable_distributed_mindrt
from mindspore.dataset.engine.offload import GetOffloadModel

import mindspore.dataset.transforms.c_transforms as c_transforms
import mindspore.dataset.transforms.py_transforms as py_transforms
import mindspore.dataset.transforms as transforms
from mindspore.dataset.text.utils import SentencePieceModel, DE_C_INTER_SENTENCEPIECE_MODE
from mindspore.parallel._utils import _get_device_num
from mindspore.dataset.debug import DebugHook

from mindspore.dataset.engine import samplers
from .iterators import DictIterator, TupleIterator, DummyIterator, check_iterator_cleanup, _set_iterator_cleanup, \
    ITERATORS_LIST, _unset_iterator_cleanup
from .queue import _SharedQueue, _Queue
from .validators import check_batch, check_shuffle, check_map, check_filter, check_repeat, check_skip, check_zip, \
    check_rename, check_device_send, check_take, check_output_shape, check_project, \
    check_sync_wait, check_zip_dataset, check_add_column, check_concat, check_split, check_bucket_batch_by_length, \
    check_save, check_tuple_iterator, check_dict_iterator, check_schema, check_to_device_send, check_padded_batch, \
    check_total_batch
from ..core.config import get_callback_timeout, _init_device_info, get_enable_shared_mem, get_num_parallel_workers, \
    get_enable_watchdog, get_seed, set_seed, get_debug_mode, get_multiprocessing_timeout_interval, _get_debug_hook_list
from ..core.datatypes import mstype_to_detype
from ..core.validator_helpers import replace_none
from ..core.py_util_helpers import ExceptionHandler
from ..transforms.py_transforms_util import FuncWrapper, Implementation
from ..vision.transforms import ToNumpy
from ...mindrecord.config import _get_enc_key, _get_enc_mode, _get_hash_mode, encrypt, append_hash_to_file

try:
    context = import_module("mindspore.context")
except ModuleNotFoundError:
    context = None

if platform.system().lower() == "darwin" and multiprocessing.get_start_method() != "fork":
    multiprocessing.set_start_method("fork", True)

OffloadToManualOffloadMode = {
    None: cde.ManualOffloadMode.UNSPECIFIED,
    False: cde.ManualOffloadMode.DISABLED,
    True: cde.ManualOffloadMode.ENABLED
}

_train_dataset = None


def _set_training_dataset(dataset):
    """
    Set the dataset to be used when training recovery has occurred.

    Args:
        dataset: the training dataset or iterator
    """
    global _train_dataset
    _train_dataset = dataset


def _get_training_dataset():
    """
    Get the dataset to be used when training recovery has occurred.

    Returns:
        training dataset/iterator
    """
    return _train_dataset


def _reset_training_dataset(global_step, dataset_size):
    """
    Reset the training dataset to the given global step.

    Args:
        global_step (int): Number of global steps that have completed training.
            Dataset will provide data from its next step after reset.
        dataset_size (int): Number of steps per epoch.
    """
    dataset = _get_training_dataset()
    if dataset is not None:
        dataset._reset(global_step, dataset_size)  # pylint: disable=protected-access
    else:
        raise RuntimeError("Training dataset is not set.")


[文档]class Shuffle(str, Enum): """Specify the shuffle mode. - ``Shuffle.GLOBAL`` : Shuffle both the files and samples. - ``Shuffle.FILES`` : Shuffle files only. - ``Shuffle.INFILE`` : Shuffle data within each file. """ GLOBAL: str = "global" FILES: str = "files" INFILE: str = "infile"
ShuffleToShuffleMode = {Shuffle.FILES: cde.ShuffleMode.FILES, Shuffle.GLOBAL: cde.ShuffleMode.GLOBAL, Shuffle.INFILE: cde.ShuffleMode.INFILE} def shuffle_to_shuffle_mode(shuffle): """ Shuffle Enum to Shuffle Mode Args: shuffle (Shuffle): shuffle flag to shuffle mode in C layer Returns: ShuffleMode, shuffle mode """ shuffle_mode = cde.ShuffleMode.GLOBAL # Global shuffle if not isinstance(shuffle, Shuffle): if shuffle is None or shuffle: shuffle_mode = cde.ShuffleMode.GLOBAL # Global shuffle else: shuffle_mode = cde.ShuffleMode.FALSE # No shuffle else: shuffle_mode = ShuffleToShuffleMode[shuffle] return shuffle_mode def shuffle_to_bool(shuffle): """ Shuffle Enum to bool Args: shuffle (Shuffle): shuffle flag to bool Returns: bool, True / False """ if shuffle is not None and not isinstance(shuffle, (bool, Shuffle)): raise TypeError("shuffle must be of boolean or enum of 'Shuffle' values like 'Shuffle.GLOBAL' or " "'Shuffle.FILES' or 'Shuffle.INFILE'.") shuffle_bool = True if not isinstance(shuffle, Shuffle): if shuffle is None: shuffle_bool = None elif shuffle: shuffle_bool = True else: shuffle_bool = False else: shuffle_bool = True return shuffle_bool @check_zip def zip(datasets): """ Zip the datasets in the input tuple of datasets. Args: datasets (tuple[Dataset]): A tuple of datasets to be zipped together. The number of datasets must be more than 1. Returns: Dataset, a new dataset with the above operation applied. Raises: ValueError: If the number of datasets is 1. TypeError: If datasets is not a tuple. Examples: >>> # Create a dataset which is the combination of dataset_1 and dataset_2 >>> import mindspore.dataset as ds >>> >>> dataset_1 = ds.GeneratorDataset([1], "column1") >>> dataset_2 = ds.GeneratorDataset([2], "column2") >>> dataset = ds.zip((dataset_1, dataset_2)) """ if len(datasets) <= 1: raise ValueError( "Can't zip empty or just one dataset!") for dataset in datasets: if not isinstance(dataset, Dataset): raise TypeError("Invalid dataset, expected Dataset object, but got %s!" % type(dataset)) return ZipDataset(datasets) def _get_operator_process(): """ Inner implemented method, mainly for passing sub-process id in C layer Returns: dict, mapping dict of operation id and corresponding process id. """ global _OP_PROCESS process_info = _OP_PROCESS op_process = dict() keys = process_info.keys() fetched_all = True for key in keys: try: op_process[key] = list(process_info[key][1]) item_full = (len(process_info[key][1]) == process_info[key][0]) except KeyError as err: raise err fetched_all = fetched_all and item_full return op_process, fetched_all def _set_dataset_permissions(file_name, num_files): """ set saved dataset files' permissions to 600 the rule of dataset filenames should be the same as those in C++. """ num_digits = len(str(num_files - 1)) if num_files == 1: paths = [file_name] else: paths = ["{}{}".format(file_name, str(x).rjust(num_digits, '0')) for x in range(num_files)] for item in paths: if os.path.exists(item): os.chmod(item, stat.S_IRUSR | stat.S_IWUSR) index_file = item + ".db" if os.path.exists(index_file): os.chmod(index_file, stat.S_IRUSR | stat.S_IWUSR) class Dataset: """ Abstract class to represent a dataset in DataEngine's data pipeline. This class is the base class of SourceDataset and Dataset, and represents a node in the data flow graph. Dataset ----------------------------------------------------------- | | | | VisionBaseDataset TextBaseDataset AudioBaseDataset | - - - | | | | | ---------------------------------------- | UnionBaseDataset | | SourceDataset - | MappableDataset DatasetOperation: MapDataset(UnionBaseDataset) BatchDataset(UnionBaseDataset) PaddedBatchDataset(UnionBaseDataset) BucketBatchByLengthDataset(UnionBaseDataset) ShuffleDataset(UnionBaseDataset) FilterDataset(UnionBaseDataset) RepeatDataset(UnionBaseDataset) SkipDataset(UnionBaseDataset) TakeDataset(UnionBaseDataset) ZipDataset(UnionBaseDataset) ConcatDataset(UnionBaseDataset) RenameDataset(UnionBaseDataset) ProjectDataset(UnionBaseDataset) SyncWaitDataset(UnionBaseDataset) Impl Dataset - vision: ImageFolderDataset(MappableDataset, VisionBaseDataset) USPSDataset(SourceDataset, VisionBaseDataset) Impl Dataset - text: TextFileDataset(SourceDataset, TextBaseDataset) YahooAnswersDataset(SourceDataset, TextBaseDataset) Impl Dataset - audio: LJSpeechDataset(MappableDataset, AudioBaseDataset) TedliumDataset(MappableDataset, AudioBaseDataset) Impl Dataset - standard: MindDataset(MappableDataset, UnionBaseDataset) TFRecordDataset(SourceDataset, UnionBaseDataset) Impl Dataset - user defined: GeneratorDataset(MappableDataset, UnionBaseDataset) NumpySlicesDataset(GeneratorDataset) Args: num_parallel_workers (int, optional): Number of workers to process the dataset in parallel. Default: ``None``. """ def __init__(self, children=None, num_parallel_workers=None, cache=None): # Note: children and parent are internal variables, not recommended for external using. self.children = replace_none(children, []) if isinstance(self.children, tuple): self.children = list(self.children) if not isinstance(self.children, list): self.children = [self.children] self.parent = [] for child in self.children: child.parent.append(weakref.ref(self)) self.num_parallel_workers = num_parallel_workers self.cache = cache self._device_iter = 0 self._input_indexs = () self.saved_output_types = None self.saved_output_shapes = None self.estimated_output_shapes = None self.runtime_context = None self._col_names = None self.dataset_size = None self._batch_size = None self._num_classes = None self._repeat_count = None self._class_indexing = None self._sync = False self._global_step = None @staticmethod def _get_operator_id(dataset): """ Internal method to iterate the tree and obtain op_id of each operation. Returns: Dataset, the root dataset of the tree. """ op_name = dict() generator_process = dict() op_name[str(dataset)] = 0 op_id = 1 def process_name(datasets, operator_id): if not datasets: return 0 temp = [] for item in datasets: for d in item.children: temp.append(d) op_name[str(d)] = operator_id from mindspore.dataset.engine.datasets_user_defined import GeneratorDataset if isinstance(d, GeneratorDataset) and d.sample_fn and d.sample_fn.pids: generator_process[operator_id] = [d.num_parallel_workers, set(d.sample_fn.pids)] operator_id = operator_id + 1 return process_name(temp, operator_id) process_name([dataset], op_id) if generator_process: global _OP_PROCESS _OP_PROCESS.update(generator_process) return op_name def create_ir_tree(self, getter_mode=False): """ Internal method to build an IR tree. Args: getter_mode (bool, optional): Whether to build IR tree in pull mode. Default: ``False``. Returns: Union[DatasetNode, Dataset], the root node of the IR tree and the root dataset of the IR tree. """ parent = self.parent self.parent = [] dataset = copy.deepcopy(self) global _OP_NAME _OP_NAME = Dataset._get_operator_id(dataset) ir_tree = dataset.parse_tree(getter_mode) self.parent = parent _init_device_info() return ir_tree, dataset def parse_tree(self, getter_mode=False): """ Internal method to parse the API tree into an IR tree. Args: getter_mode (bool, optional): Whether to build IR tree in pull mode. Default: ``False``. Returns: DatasetNode, the root node of the IR tree. """ if len(self.parent) > 1: raise ValueError("The data pipeline is not a tree (i.e., one node has 2 consumers)") ir_children = [d.parse_tree(getter_mode) for d in self.children] # Bootstrap can only be performed on a copy of the original dataset node. # Bootstrap on original dataset node will make all iterators share the same process pool self.pre_parse(getter_mode) self.iterator_bootstrap() ir_node = self.parse(ir_children) ir_node = self.post_parse(ir_node) return ir_node def __safe_deepcopy__(self, memodict, exclude=()): if id(self) in memodict: return memodict[id(self)] cls = self.__class__ new_op = cls.__new__(cls) memodict[id(self)] = new_op for arg, value in self.__dict__.items(): if arg in exclude: setattr(new_op, arg, value) else: try: setattr(new_op, arg, copy.deepcopy(value, memodict)) except TypeError: setattr(new_op, arg, value) return new_op @staticmethod def _noop_mode(): if _is_role_sched(): return True return False def iterator_bootstrap(self): pass def __add__(self, datasets): return self.concat(datasets)
[文档] def to_json(self, filename=""): """ Serialize a pipeline into JSON string and dump into file if filename is provided. Args: filename (str): filename of JSON file to be saved as. Default: ``""``. Returns: str, JSON string of the pipeline. Examples: >>> import mindspore.dataset as ds >>> mnist_dataset_dir = "/path/to/mnist_dataset_directory" >>> dataset = ds.MnistDataset(dataset_dir=mnist_dataset_dir) >>> dataset_json = dataset.to_json("/path/to/mnist_dataset_pipeline.json") """ ir_tree, _ = self.create_ir_tree() return json.loads(ir_tree.to_json(filename))
# The decorator has been deleted.
[文档] def bucket_batch_by_length(self, column_names, bucket_boundaries, bucket_batch_sizes, element_length_function=None, pad_info=None, pad_to_bucket_boundary=False, drop_remainder=False): """ Bucket elements according to their lengths. Each bucket will be padded and batched when they are full. A length function is called on each row in the dataset. The row is then bucketed based on its length and bucket boundaries. When a bucket reaches its corresponding size specified in bucket_batch_sizes, the entire bucket will be padded according to pad_info, and then form a batch. Refer to the following figure for the execution process: .. image:: bucket_batch_by_length_en.png Args: column_names (list[str]): Columns passed to element_length_function. bucket_boundaries (list[int]): A list consisting of the upper boundaries of the buckets. Must be strictly increasing. If there are n boundaries, n+1 buckets are created: One bucket for [0, bucket_boundaries[0]), one bucket for [bucket_boundaries[i], bucket_boundaries[i+1]) for each 0<i<n-1, and the last bucket for [bucket_boundaries[n-1], inf). bucket_batch_sizes (list[int]): A list consisting of the batch sizes for each bucket. Must contain len(bucket_boundaries)+1 elements. element_length_function (Callable, optional): A function that takes in M arguments where M = len(column_names) and returns an integer. If no value provided, parameter M the len(column_names) must be 1, and the size of the first dimension of that column will be taken as the length. Default: ``None``. pad_info (dict, optional): The information about how to batch each column. The key corresponds to the column name, and the value must be a tuple of 2 elements. The first element corresponds to the shape to pad to, and the second element corresponds to the value to pad with. If a column is not specified, then that column will be padded to the longest in the current batch, and 0 will be used as the padding value. Any None dimensions will be padded to the longest in the current batch, unless if `pad_to_bucket_boundary` is ``True``. If no padding is wanted, set `pad_info` to ``None``. Default: ``None``. pad_to_bucket_boundary (bool, optional): If ``True``, will pad each None dimension in `pad_info` to the bucket_boundary minus 1. If there are any elements that fall into the last bucket, an error will occur. Default: ``False``. drop_remainder (bool, optional): If ``True``, will drop the last batch for each bucket if it is not a full batch. Default: ``False``. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> # Create a dataset where certain counts rows are combined into a batch >>> # and drops the last incomplete batch if there is one. >>> import mindspore.dataset as ds >>> import numpy as np >>> def generate_2_columns(n): ... for i in range(n): ... yield (np.array([i]), np.array([j for j in range(i + 1)])) >>> >>> column_names = ["col1", "col2"] >>> dataset = ds.GeneratorDataset(generate_2_columns(8), column_names) >>> bucket_boundaries = [5, 10] >>> bucket_batch_sizes = [2, 1, 1] >>> element_length_function = (lambda col1, col2: max(len(col1), len(col2))) >>> # Will pad col2 to shape [bucket_boundaries[i]] where i is the >>> # index of the bucket that is currently being batched. >>> pad_info = {"col2": ([None], -1)} >>> pad_to_bucket_boundary = True >>> dataset = dataset.bucket_batch_by_length(column_names, bucket_boundaries, ... bucket_batch_sizes, ... element_length_function, pad_info, ... pad_to_bucket_boundary) """ return BucketBatchByLengthDataset(self, column_names, bucket_boundaries, bucket_batch_sizes, element_length_function, pad_info, pad_to_bucket_boundary, drop_remainder)
[文档] @check_batch def batch(self, batch_size, drop_remainder=False, num_parallel_workers=None, **kwargs): """ Combine batch_size number of consecutive rows into batch which apply per_batch_map to the samples first. For any column, all the elements within that column must have the same shape. Refer to the following figure for the execution process: .. image:: batch_en.png Note: The order of using repeat and batch reflects the number of batches and per_batch_map. It is recommended that the repeat operation applied after the batch operation finished. Args: batch_size (Union[int, Callable]): The number of rows each batch is created with. An int or callable object which takes exactly 1 parameter, BatchInfo. drop_remainder (bool, optional): Determines whether or not to drop the last block whose data row number is less than batch size. Default: ``False`` . If ``True`` , and if there are less than `batch_size` rows available to make the last batch, then those rows will be dropped and not propagated to the child node. num_parallel_workers (int, optional): Number of workers(threads) to process the dataset in parallel. Default: ``None`` . **kwargs: - per_batch_map (Callable[[List[numpy.ndarray], ..., List[numpy.ndarray], BatchInfo], \ (List[numpy.ndarray], ..., List[numpy.ndarray])], optional): Per batch map callable. Default: ``None``. A callable which takes (List[numpy.ndarray], ..., List[numpy.ndarray], BatchInfo) as input parameters. Each list[numpy.ndarray] represents a batch of numpy.ndarray on a given column. The number of lists should match with the number of entries in input_columns. The last parameter of the callable should always be a BatchInfo object. Per_batch_map should return (list[numpy.ndarray], list[numpy.ndarray], ...). The length of each list in output should be the same as the input. output_columns is required if the number of output lists is different from input. - input_columns (Union[str, list[str]], optional): List of names of the input columns. The size of the list should match with signature of `per_batch_map` callable. Default: ``None`` . - output_columns (Union[str, list[str]], optional): List of names assigned to the columns outputted by the last operation. This parameter is mandatory if len(input_columns) != len(output_columns). The size of this list must match the number of output columns of the last operation. Default: ``None`` , output columns will have the same name as the input columns, i.e., the columns will be replaced. - python_multiprocessing (bool, optional): Parallelize Python function `per_batch_map` with multi-processing or multi-threading mode, ``True`` means multi-processing, ``False`` means multi-threading If `per_batch_map` is a I/O bound task, use multi-threading mode. If `per_batch_map` is a CPU bound task, it is recommended to use multi-processing mode. Default: ``False`` , use python multi-threading mode. - max_rowsize(Union[int, list[int]], optional): Maximum size of row in MB that is used for shared memory allocation to copy data between processes, the total occupied shared memory will increase as ``num_parallel_workers`` and :func:`mindspore.dataset.config.set_prefetch_size` increase. If set to -1, shared memory will be dynamically allocated with the actual size of data. This is only used if ``python_multiprocessing`` is set to True. If it is an int value, it represents ``input_columns`` and ``output_columns`` use this value as the unit to create shared memory. If it is a list, the first element represents the ``input_columns`` use this value as the unit to create shared memory, and the second element represents ``output_columns`` use this value as the unit to create shared memory. Default: 16. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> # 1) Create a dataset where every 5 rows are combined into a batch >>> # and drops the last incomplete batch if there is one. >>> import mindspore.dataset as ds >>> from PIL import Image >>> >>> cifar10_dataset_dir = "/path/to/cifar10_dataset_directory" >>> dataset = ds.Cifar10Dataset(dataset_dir=cifar10_dataset_dir, num_samples=10) >>> dataset = dataset.batch(5, True) >>> >>> # 2) resize image according to its batch number, if it's 5-th batch, resize to (5^2, 5^2) = (25, 25) >>> def np_resize(col, BatchInfo): ... output = col.copy() ... s = (BatchInfo.get_batch_num() + 1) ** 2 ... index = 0 ... for c in col: ... img = Image.fromarray(c.astype('uint8')).convert('RGB') ... img = img.resize((s, s)) ... output[index] = np.array(img) ... index += 1 ... return (output,) >>> dataset = dataset.batch(batch_size=8, input_columns=["image"], per_batch_map=np_resize) >>> >>> # 3) Create a dataset where its batch size is dynamic >>> # Define a callable batch size function and let batch size increase 1 each time. >>> def add_one(BatchInfo): ... return BatchInfo.get_batch_num() + 1 >>> dataset = dataset.batch(batch_size=add_one, drop_remainder=True) """ return BatchDataset(self, batch_size, drop_remainder, num_parallel_workers, **kwargs)
[文档] @check_padded_batch def padded_batch(self, batch_size, drop_remainder=False, num_parallel_workers=None, pad_info=None): """ Combine batch_size number of consecutive rows into batch which apply pad_info to the samples first. Refer to the following figure for the execution process: .. image:: padded_batch_en.png Note: The order of using repeat and padded_batch reflects the number of batches. It is recommended that the repeat operation applied after the padded_batch operation finished. Args: batch_size (Union[int, Callable]): The number of rows each batch is created with. An int or callable object which takes exactly 1 parameter, BatchInfo. drop_remainder (bool, optional): Determines whether or not to drop the last block whose data row number is less than batch size. Default: ``False``. If ``True``, and if there are less than batch_size rows available to make the last batch, then those rows will be dropped and not propagated to the child node. num_parallel_workers (int, optional): Number of workers(threads) to process the dataset in parallel. Default: ``None``. pad_info (dict, optional): The pad information about how to batch each column. The key corresponds to the column name, and the value must be a tuple of 2 elements. The first element corresponds to the shape to pad to, and the second element corresponds to the value to pad with. If a column is not specified, then that column will be padded to the longest in the current batch, and 0 will be used as the padding value. If ``pad_info={"col1": ([224, 224], 0)}``, expand the data column named ``col1`` to shape (224, 224), and fill in the missing values with 0. If ``pad_info={}``, all samples in the batch will be filled to the shape with the largest sample in the current batch. If ``pad_info={"col1": (None, 100)}``, all samples in the batch will be filled to the shape with the largest sample in the current batch, and fill in the missing values with 100. If no padding is wanted, set `pad_info` to ``None``. Default: ``None``. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> # 1) Pad every sample to the largest sample's shape and batch the samples >>> import mindspore.dataset as ds >>> dataset = ds.NumpySlicesDataset([[1], [1, 2], [1, 2, 3], [1, 2, 3, 4]], "column1") >>> dataset = dataset.padded_batch(2, True, pad_info={}) >>> >>> # 2) Create a dataset where every 3 rows are combined into a batch >>> # and drops the last incomplete batch if there is one. >>> dataset = ds.NumpySlicesDataset([i for i in range(10)], "column1") >>> dataset = dataset.padded_batch(3, True) >>> >>> # 3) Create a dataset where its batch size is dynamic >>> # Define a callable batch size function and let batch size increase 1 each time. >>> def add_one(BatchInfo): ... return BatchInfo.get_batch_num() + 1 >>> dataset = dataset.padded_batch(batch_size=add_one, drop_remainder=True) """ return PaddedBatchDataset(self, batch_size, drop_remainder, num_parallel_workers, pad_info)
[文档] @check_sync_wait def sync_wait(self, condition_name, num_batch=1, callback=None): """ Add a blocking condition to the input Dataset and a synchronize action will be applied. Args: condition_name (str): The condition name that is used to toggle sending next row. num_batch (int): the number of batches without blocking at the start of each epoch. Default: ``1``. callback (function): The callback function that will be invoked when sync_update is called. Default: ``None``. Returns: Dataset, a new dataset with the above operation applied. Raises: RuntimeError: If condition name already exists. Examples: >>> import mindspore.dataset as ds >>> import numpy as np >>> def gen(): ... for i in range(100): ... yield (np.array(i),) >>> >>> class Augment: ... def __init__(self, loss): ... self.loss = loss ... ... def preprocess(self, input_): ... return input_ ... ... def update(self, data): ... self.loss = data["loss"] >>> >>> batch_size = 4 >>> dataset = ds.GeneratorDataset(gen, column_names=["input"]) >>> >>> aug = Augment(0) >>> dataset = dataset.sync_wait(condition_name="policy", callback=aug.update) >>> dataset = dataset.map(operations=[aug.preprocess], input_columns=["input"]) >>> dataset = dataset.batch(batch_size) >>> count = 0 >>> for data in dataset.create_dict_iterator(num_epochs=1, output_numpy=True): ... assert data["input"][0] == count ... count += batch_size ... data = {"loss": count} ... dataset.sync_update(condition_name="policy", data=data) """ return SyncWaitDataset(self, condition_name, num_batch, callback)
[文档] @check_shuffle def shuffle(self, buffer_size): """ Shuffle the dataset by creating a cache with the size of `buffer_size` . 1. Make a shuffle buffer that contains the first `buffer_size` rows. 2. Randomly select an element from the shuffle buffer to be the next row propagated to the child node. 3. Get the next row (if any) from the parent node and put it in the shuffle buffer. 4. Repeat steps 2 and 3 until there are no more rows left in the shuffle buffer. A random seed can be provided to be used on the first epoch via `dataset.config.set_seed` . In every subsequent epoch, the seed is changed to a new one, randomly generated value. Args: buffer_size (int): The size of the buffer (must be larger than 1) for shuffling. Setting `buffer_size` equal to the number of rows in the entire dataset will result in a global shuffle. Returns: Dataset, a new dataset with the above operation applied. Raises: RuntimeError: If exist sync operations before shuffle. Examples: >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> >>> # Optionally set the seed for fixed randomness >>> ds.config.set_seed(58) >>> >>> # Create a shuffled dataset using a shuffle buffer of size 4 >>> dataset = dataset.shuffle(4) """ return ShuffleDataset(self, buffer_size)
[文档] def flat_map(self, func): """ Map `func` to each row in dataset and flatten the result. Args: func (function): A function that must take one `numpy.ndarray` as an argument and return a `Dataset` . Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> # 1) flat_map on one column dataset >>> dataset = ds.NumpySlicesDataset([[0, 1], [2, 3]], shuffle=False) >>> >>> def repeat(array): ... # create a NumpySlicesDataset with the array ... data = ds.NumpySlicesDataset(array, shuffle=False) ... # repeat the dataset twice ... data = data.repeat(2) ... return data >>> >>> dataset = dataset.flat_map(repeat) >>> # [0, 1, 0, 1, 2, 3, 2, 3] >>> >>> # 2) flat_map on multi column dataset >>> dataset = ds.NumpySlicesDataset(([[0, 1], [2, 3]], [[0, -1], [-2, -3]]), shuffle=False) >>> >>> def plus_and_minus(col1, col2): ... # apply different methods on columns ... data = ds.NumpySlicesDataset((col1 + 1, col2 - 1), shuffle=False) ... return data >>> >>> dataset = dataset.flat_map(plus_and_minus) >>> # ([1, 2, 3, 4], [-1, -2, -3, -4]) Raises: TypeError: If `func` is not a function. TypeError: If `func` doesn't return a Dataset. """ dataset = None if not hasattr(func, '__call__'): logger.critical("func must be a function.") raise TypeError("func must be a function.") for row_data in self.create_tuple_iterator(num_epochs=1, output_numpy=True): if dataset is None: dataset = func(*row_data) else: dataset += func(*row_data) if not isinstance(dataset, Dataset): logger.critical("flat_map must return a Dataset object.") raise TypeError("flat_map must return a Dataset object.") return dataset
[文档] @check_map def map(self, operations, input_columns=None, output_columns=None, column_order=None, num_parallel_workers=None, **kwargs): """ Apply each operation in operations to this dataset. Each operation will be passed one or more columns from the dataset as input, and one or more columns will be outputted. The first operation will be passed the columns specified in input_columns as input. If there is more than one operation in operations, the outputted columns of the previous operation are used as the input columns for the next operation. The columns outputted by the very last operation will be assigned names specified by `output_columns` , and if not specified, the column name of output column is same as that of `input_columns` . - If you use transformations ( `vision transform <https://mindspore.cn/docs/en/r2.3.0/api_python/mindspore.\ dataset.transforms.html#module-mindspore.dataset.vision>`_ , `nlp transform <https://mindspore.cn/docs/en/r2.3.0/api_python/mindspore.\ dataset.transforms.html#module-mindspore.dataset.text>`_ , `audio transform <https://mindspore.cn/docs/en/r2.3.0/api_python/mindspore.\ dataset.transforms.html#module-mindspore.dataset.audio>`_ ) provided by mindspore dataset, please use the following parameters: .. image:: map_parameter_en.png - If you use user-defined transform as PyFunc (Python Func), please use the following parameters: .. image:: map_parameter_pyfunc_en.png Args: operations (Union[list[TensorOperation], list[functions]]): List of operations to be applied on the dataset. Operations are applied in the order they appear in this list. input_columns (Union[str, list[str]], optional): List of the names of the columns that will be passed to the first operation as input. The size of this list must match the number of input columns expected by the first operation. Default: ``None``, the first operation will be passed however many columns that are required, starting from the first column. output_columns (Union[str, list[str]], optional): List of names assigned to the columns outputted by the last operation. This parameter is mandatory if len(input_columns) != len(output_columns). The size of this list must match the number of output columns of the last operation. Default: ``None``, output columns will have the same name as the input columns, i.e., the columns will be replaced. num_parallel_workers (int, optional): Number of threads used to process the dataset in parallel. Default: ``None``, the value from the configuration will be used. **kwargs: - python_multiprocessing (bool, optional): Parallelize Python operations with multiple worker processes. This option could be beneficial if the Python operation is computational heavy. Default: ``False``. - max_rowsize (Union[int, list[int]], optional): Maximum size of row in MB that is used for shared memory allocation to copy data between processes, the total occupied shared memory will increase as ``num_parallel_workers`` and :func:`mindspore.dataset.config.set_prefetch_size` increase. If set to -1, shared memory will be dynamically allocated with the actual size of data. This is only used if ``python_multiprocessing`` is set to True. If it is an int value, it represents ``input_columns`` and ``output_columns`` use this value as the unit to create shared memory. If it is a list, the first element represents the ``input_columns`` use this value as the unit to create shared memory, and the second element represents ``output_columns`` use this value as the unit to create shared memory. Default: 16. - cache (DatasetCache, optional): Use tensor caching service to speed up dataset processing. Default: ``None``, which means no cache is used. - callbacks (DSCallback, list[DSCallback], optional): List of Dataset callbacks to be called. Default: ``None``. - offload (bool, optional): Flag to indicate whether offload is used. Default: ``None``. Note: - Input `operations` accepts TensorOperations defined in mindspore.dataset part, plus user-defined Python functions (PyFuncs). - Do not add network computing operators from mindspore.nn and mindspore.ops or others into this `operations` . Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> import mindspore.dataset.vision as vision >>> # dataset is an instance of Dataset which has 2 columns, "image" and "label". >>> # image is of type bytes type which can be decoded to RGB >>> # label is of type int32 >>> cifar10_dataset_dir = "/path/to/cifar10_dataset_directory" >>> dataset = ds.Cifar10Dataset(dataset_dir=cifar10_dataset_dir) >>> >>> # Define two operations, where each operation accepts 1 input column and outputs 1 column. >>> decode_op = vision.Decode(to_pil=False) >>> random_jitter_op = vision.RandomColorAdjust(brightness=(0.8, 0.8), contrast=(1, 1), ... saturation=(1, 1), hue=(0, 0)) >>> >>> # 1) Simple map example. >>> >>> # Apply decode_op on column "image". >>> dataset = dataset.map(operations=[decode_op], input_columns=["image"]) >>> >>> # Decode and rename column "image" to "decoded_image". >>> dataset = dataset.map(operations=[decode_op], input_columns=["image"], output_columns=["decoded_image"]) >>> >>> # A simple example for user defined python function transform. >>> dataset = ds.NumpySlicesDataset(data=[[0, 1, 2]], column_names=["data"]) >>> dataset = dataset.map(operations=[(lambda x: x - 1)], input_columns=["data"]) >>> >>> # 2) Map example with more than one operation. >>> >>> # Create a dataset where the images are decoded, then randomly color jittered. >>> # decode_op takes column "image" as input and outputs one column. The column >>> # outputted by decode_op is passed as input to random_jitter_op. >>> # random_jitter_op will output one column. Column "image" will be replaced by >>> # the column outputted by random_jitter_op (the very last operation). All other >>> # columns are unchanged. >>> dataset = dataset.map(operations=[decode_op, random_jitter_op], input_columns=["image"]) >>> >>> # Rename the column outputted by random_jitter_op to "image_mapped". >>> dataset = dataset.map(operations=[decode_op, random_jitter_op], input_columns=["image"], ... output_columns=["image_mapped"]) >>> >>> # Map with multiple operations using pyfunc and rename column's name >>> dataset = ds.NumpySlicesDataset(data=[[0, 1, 2]], column_names=["data"]) >>> dataset = dataset.map(operations=[(lambda x: x * x), (lambda x: x - 1)], input_columns=["data"], ... output_columns=["data_mapped"]) >>> >>> # 3) Example where number of input columns is not equal to number of output columns. >>> >>> # operations[0] is a lambda that takes 2 columns as input and outputs 3 columns. >>> # operations[1] is a lambda that takes 3 columns as input and outputs 1 column. >>> # operations[2] is a lambda that takes 1 column as input and outputs 4 columns. >>> # >>> # Note: The number of output columns of operation[i] must equal the number of >>> # input columns of operation[i+1]. Otherwise, this map call will also result >>> # in an error. >>> operations = [(lambda x, y: (x, x + y, x + y + 1)), ... (lambda x, y, z: x * y * z), ... (lambda x: (x % 2, x % 3, x % 5, x % 7))] >>> dataset = ds.NumpySlicesDataset(data=([[0, 1, 2]], [[3, 4, 5]]), column_names=["x", "y"]) >>> dataset = dataset.map(operations, input_columns=["x", "y"], ... output_columns=["mod2", "mod3", "mod5", "mod7"]) """ if hasattr(self, 'operator_mixed') and getattr(self, 'operator_mixed') is True: num_parallel_workers = 1 logger.warning( "Input 'operations' of 'map' includes network computing operators like in mindspore.nn, mindspore.ops, " "mindspore.numpy module and etc, which do not support multi-thread compiling, recommend to replace it " "with python implemented operator like numpy etc. Here decrease 'num_parallel_workers' into 1.") return MapDataset(self, operations, input_columns, output_columns, num_parallel_workers, **kwargs)
[文档] @check_filter def filter(self, predicate, input_columns=None, num_parallel_workers=None): """ Filter dataset by prediction. Args: predicate (callable): Python callable which returns a boolean value. If False then filter the element. input_columns (Union[str, list[str]], optional): List of names of the input columns. If not provided or provided with ``None``, the predicate will be applied on all columns in the dataset. Default: ``None``. num_parallel_workers (int, optional): Number of workers to process the dataset in parallel. Default: ``None``. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> # generator data(0 ~ 19) >>> # filter the data that greater than or equal to 11 >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(20)], "data") >>> dataset = dataset.filter(predicate=lambda data: data < 11, input_columns = ["data"]) """ return FilterDataset(self, predicate, input_columns, num_parallel_workers)
[文档] @check_repeat def repeat(self, count=None): """ Repeat this dataset `count` times. Repeat infinitely if the `count` is ``None`` or ``-1``. Note: The order of using repeat and batch reflects the number of batches. It is recommended that the repeat operation is used after the batch operation. Args: count (int): Number of times the dataset is going to be repeated. Default: ``None``. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> >>> # Create a dataset with 10 elements >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> ori_size = dataset.get_dataset_size() >>> >>> # Repeat the dataset 50 times. >>> dataset = dataset.repeat(50) >>> repeated_size = dataset.get_dataset_size() >>> print("ori_size", ori_size, ", repeated_size", repeated_size) ori_size 10 , repeated_size 500 >>> >>> # Since the original dataset size is less than batch_size, thus no data is returned >>> dataset1 = ds.GeneratorDataset([i for i in range(10)], "column1") >>> dataset1 = dataset1.batch(batch_size=20, drop_remainder=True) >>> dataset1 = dataset1.repeat(6) >>> >>> # Repeat the original dataset to 60 elements, thus 3 batches are returned >>> dataset2 = ds.GeneratorDataset([i for i in range(10)], "column1") >>> dataset2 = dataset2.repeat(6) >>> dataset2 = dataset2.batch(batch_size=20, drop_remainder=True) >>> print("dataset1 size", dataset1.get_dataset_size(), ", dataset2 size", dataset2.get_dataset_size()) dataset1 size 0 , dataset2 size 3 """ return RepeatDataset(self, count)
[文档] @check_skip def skip(self, count): """ Skip the first N elements of this dataset. Args: count (int): Number of elements in the dataset to be skipped. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> # Skip first 3 elements of dataset and retain 7 elements. >>> dataset = dataset.skip(3) """ return SkipDataset(self, count)
[文档] @check_take def take(self, count=-1): """ Take the first specified number of samples from the dataset. Args: count (int, optional): The desired number of samples to take. If the value exceeds the total number of samples in the dataset, all data will be returned. Default: ``-1`` , will return all data. Note: When there are operations that will change the number of samples of the dataset in the data pipeline, the location of the `take` operation can change its effect. For example, `batch` operation will combine the successive samples of the specified `batch_size` into 1 sample, so `.batch(batch_size).take(1)` will be equivalent to `.take(batch_size).batch(batch_size)`. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> mnist_dataset_dir = "/path/to/mnist_dataset_directory" >>> dataset = ds.MnistDataset(dataset_dir=mnist_dataset_dir) >>> # Take 50 samples from MNIST dataset. >>> dataset = dataset.take(50) """ return TakeDataset(self, count)
def _get_absolute_split_sizes(self, sizes): """ Internal method called by split to calculate absolute split sizes and to do some error checking after calculating absolute split sizes. Returns: int, absolute split sizes of the dataset. """ # Call get_dataset_size here and check input here because # don't want to call this once in check_split and another time in # here again dataset_size = self.get_dataset_size() if dataset_size is None or dataset_size <= 0: raise RuntimeError("dataset_size is unknown, unable to split.") if not isinstance(sizes, list): raise RuntimeError("sizes must be a list.") all_int = all(isinstance(item, int) for item in sizes) if all_int: sizes_sum = sum(sizes) if sizes_sum != dataset_size: raise RuntimeError("Sum of split sizes {} is not equal to dataset size {}." .format(sizes_sum, dataset_size)) return sizes absolute_sizes = [] for item in sizes: absolute_size = int(round(item * dataset_size)) if absolute_size == 0: raise RuntimeError("Split percentage {} is too small.".format(item)) absolute_sizes.append(absolute_size) absolute_sizes_sum = sum(absolute_sizes) # if we still need more rows, give them to the first split. # if we have too many rows, remove the extras from the first split that has # enough rows. size_difference = int(dataset_size - absolute_sizes_sum) if size_difference > 0: absolute_sizes[0] += size_difference else: for i, _ in enumerate(absolute_sizes): if absolute_sizes[i] + size_difference > 0: absolute_sizes[i] += size_difference break if sum(absolute_sizes) != dataset_size: raise RuntimeError("Sum of calculated split sizes {} is not equal to dataset size {}." .format(absolute_sizes_sum, dataset_size)) return absolute_sizes
[文档] @check_split def split(self, sizes, randomize=True): """ Split the dataset into smaller, non-overlapping datasets. Args: sizes (Union[list[int], list[float]]): If a list of integers [s1, s2, …, sn] is provided, the dataset will be split into n datasets of size s1, size s2, …, size sn respectively. If the sum of all input sizes does not equal the original dataset size, an error will throw. If a list of floats [f1, f2, …, fn] is provided, all floats must be between 0 and 1 and must sum to 1, otherwise an error will throw. The dataset will be split into n Datasets of size round(f1*K), round(f2*K), …, round(fn*K) where K is the size of the original dataset. If after rounding: - Any size equals 0, an error will occur. - The sum of split sizes < K, the difference of K - sigma(round(fi * k)) will be added to the first split. - The sum of split sizes > K, the difference of sigma(round(fi * K)) - K will be removed from the first large enough split such that it will have at least 1 row after removing the difference. randomize (bool, optional): Determines whether or not to split the data randomly. Default: ``True``. If True, the data will be randomly split. Otherwise, each split will be created with consecutive rows from the dataset. Note: 1. Dataset cannot be sharded if split is going to be called. 2. It is strongly recommended to not shuffle the dataset, but use randomize=True instead. Shuffling the dataset may not be deterministic, which means the data in each split will be different in each epoch. Returns: Tuple[Dataset], a tuple of new datasets split from the original one. Raises: RuntimeError: If get_dataset_size returns None or is not supported for this dataset. RuntimeError: If `sizes` is list of integers and sum of all elements in sizes does not equal the dataset size. RuntimeError: If `sizes` is list of float and there is a split with size 0 after calculations. RuntimeError: If the dataset is sharded prior to calling split. ValueError: If `sizes` is list of float and not all floats are between 0 and 1, or if the floats don't sum to 1. Examples: >>> # Split the data into train part and test part. >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> train_dataset, test_dataset = dataset.split([0.9, 0.1]) """ if self.is_shuffled(): logger.warning("Dataset is shuffled before split.") if self.is_sharded(): raise RuntimeError("Dataset should not be sharded before split.") absolute_sizes = self._get_absolute_split_sizes(sizes) splits = [] rows_to_skip = 0 for size in absolute_sizes: ds = copy.deepcopy(self) if randomize: # want to shuffle the same way every epoch before split # in alter_tree, shuffle buffer is minimum 10000, so use 10000 here ds = ds.shuffle(10000) ds.reshuffle_each_epoch = False if rows_to_skip > 0: ds = ds.skip(rows_to_skip) ds = ds.take(size) splits.append(ds) rows_to_skip += size return tuple(splits)
[文档] @check_zip_dataset def zip(self, datasets): """ Zip the datasets in the sense of input tuple of datasets. Columns in the input datasets must have different name. Args: datasets (Union[Dataset, tuple[Dataset]]): A tuple of datasets or a single class Dataset to be zipped together with this dataset. Returns: Dataset, a new dataset with the above operation applied. Raises: TypeError: The parameter is not dataset object or tuple of dataset objects. Examples: >>> # Create a dataset which is the combination of dataset_1 and dataset_2 >>> import mindspore.dataset as ds >>> dataset_1 = ds.GeneratorDataset([1, 2, 3], "column1") >>> dataset_2 = ds.GeneratorDataset([1, 2, 3], "column2") >>> dataset = dataset_1.zip(dataset_2) """ if isinstance(datasets, tuple): datasets = (self, *datasets) elif isinstance(datasets, Dataset): datasets = (self, datasets) else: raise TypeError("Invalid datasets, expected Dataset object or tuple of Dataset, but got %s!" % datasets) return ZipDataset(datasets)
[文档] @check_concat def concat(self, datasets): """ Concatenate the dataset objects in the input list. Performing "+" operation on dataset objects can achieve the same effect. For a dataset concatenated by many other dataset objects, it returns the data in the order of datasets passed in. If you want to change the data order(such as random selection from each dataset instead of in sequence), apply `use_sampler` method on the concatenated dataset object. Currently `use_sampler` supports `dataset.DistributedSampler` for sharding selection from each dataset or `dataset.RandomSampler` for random selection from each dataset, see examples below. Note: The column name, and rank and type of the column data must be the same in the input datasets. Args: datasets (Union[list, Dataset]): A list of datasets or a single class Dataset to be concatenated together with this dataset. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> dataset_1 = ds.GeneratorDataset([1, 2, 3], "column1", shuffle=False) >>> dataset_2 = ds.GeneratorDataset([4, 5, 6], "column1", shuffle=False) >>> >>> # Create a dataset by concatenating dataset_1 and dataset_2 with "+" operator >>> dataset = dataset_1 + dataset_2 >>> # Create a dataset by concatenating dataset_1 and dataset_2 with concat operation >>> dataset = dataset_1.concat(dataset_2) >>> >>> # Check the data order of dataset >>> dataset_1 = ds.GeneratorDataset([1, 2, 3], "column1", shuffle=False) >>> dataset_2 = ds.GeneratorDataset([4, 5, 6], "column1", shuffle=False) >>> dataset = dataset_1 + dataset_2 >>> result = list(dataset) >>> # [[Tensor(shape=[], dtype=Int64, value= 1)], [Tensor(shape=[], dtype=Int64, value= 2)], >>> # [Tensor(shape=[], dtype=Int64, value= 3)], [Tensor(shape=[], dtype=Int64, value= 4)], >>> # [Tensor(shape=[], dtype=Int64, value= 5)], [Tensor(shape=[], dtype=Int64, value= 6)]] >>> >>> # Change the data order of concatenated dataset with sharding selection >>> dataset_1 = ds.GeneratorDataset([1, 2, 3], "column1", shuffle=False) >>> dataset_2 = ds.GeneratorDataset([4, 5, 6], "column1", shuffle=False) >>> dataset = dataset_1.concat(dataset_2) >>> dataset.use_sampler(ds.DistributedSampler(num_shards=2, shard_id=1, shuffle=False)) >>> result = list(dataset) >>> # [[Tensor(shape=[], dtype=Int64, value= 2)], [Tensor(shape=[], dtype=Int64, value= 4)], >>> # [Tensor(shape=[], dtype=Int64, value= 6)]] >>> >>> # Change the data order of concatenated dataset with random selection >>> dataset_1 = ds.GeneratorDataset([1, 2, 3], "column1", shuffle=False) >>> dataset_2 = ds.GeneratorDataset([4, 5, 6], "column1", shuffle=False) >>> dataset = dataset_1.concat(dataset_2) >>> dataset.use_sampler(ds.RandomSampler()) >>> result = list(dataset) >>> # [[Tensor(shape=[], dtype=Int64, value= 1)], [Tensor(shape=[], dtype=Int64, value= 4)], >>> # [Tensor(shape=[], dtype=Int64, value= 2)], [Tensor(shape=[], dtype=Int64, value= 5)], >>> # [Tensor(shape=[], dtype=Int64, value= 6)], [Tensor(shape=[], dtype=Int64, value= 3)]] """ if isinstance(datasets, Dataset): datasets = [self] + [datasets] elif isinstance(datasets, list): datasets = [self] + datasets else: raise TypeError("Invalid datasets, expected Dataset object or list of Dataset, but got %s!" % datasets) return ConcatDataset(datasets)
[文档] @check_rename def rename(self, input_columns, output_columns): """ Rename the columns in input datasets. Args: input_columns (Union[str, list[str]]): List of names of the input columns. output_columns (Union[str, list[str]]): List of names of the output columns. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> input_columns = ["input_col1", "input_col2", "input_col3"] >>> output_columns = ["output_col1", "output_col2", "output_col3"] >>> >>> # Create a dataset with 3 columns >>> dataset = ds.GeneratorDataset([(1, 2, 3), (3, 4, 5), (5, 6, 7)], column_names=input_columns) >>> >>> # Rename "input_col1" to "output_col1", "input_col2" to "output_col2", "input_col3" to "output_col3" >>> dataset = dataset.rename(input_columns=input_columns, output_columns=output_columns) """ return RenameDataset(self, input_columns, output_columns)
[文档] @check_project def project(self, columns): """ The specified columns will be selected from the dataset and passed into the pipeline with the order specified. The other columns are discarded. Args: columns(Union[str, list[str]]): List of names of the columns to project. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> # Create a dataset with 3 columns >>> input_columns = ["column1", "column2", "column3"] >>> dataset = ds.GeneratorDataset([(1, 2, 3), (3, 4, 5), (5, 6, 7)], column_names=input_columns) >>> >>> columns_to_project = ["column3", "column1", "column2"] >>> # in that order, regardless of the original order of columns. >>> dataset = dataset.project(columns=columns_to_project) """ return ProjectDataset(self, columns)
[文档] def apply(self, apply_func): """ Apply a function in this dataset. Args: apply_func (function): A function that must take one `Dataset` as an argument and return a preprocessed `Dataset` . Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> >>> # Declare an apply_func function which returns a Dataset object >>> def apply_func(data): ... data = data.batch(2) ... return data >>> >>> # Use apply to call apply_func >>> dataset = dataset.apply(apply_func) Raises: TypeError: If apply_func is not a function. TypeError: If apply_func doesn't return a Dataset. """ if not hasattr(apply_func, '__call__'): raise TypeError("apply_func must be a function.") dataset = apply_func(self) if not isinstance(dataset, Dataset): raise TypeError("apply_func must return a dataset.") return dataset
@check_device_send def device_que(self, send_epoch_end=True, create_data_info_queue=False, queue_name=""): """ Return a transferred Dataset that transfers data through a device. Args: send_epoch_end (bool, optional): Whether to send end of sequence to device or not. Default: ``True``. create_data_info_queue (bool, optional): Whether to create queue which stores types and shapes of data or not. Default: ``False``. queue_name (str, optional): Name of queue which connects dataset processing and model computing. Default: ``""``. Note: If device is Ascend, features of data will be transferred one by one. The limitation of data transmission per time is 256M. Returns: Dataset, a new dataset with the above operation applied. Examples: >>> import mindspore.dataset as ds >>> import time >>> >>> data = ds.TFRecordDataset('/path/to/TF_FILES', '/path/to/TF_SCHEMA_FILE', shuffle=ds.Shuffle.FILES) >>> data = data.device_que() >>> data.send() >>> time.sleep(0.1) >>> data.stop_send() """ return TransferDataset(self, send_epoch_end, create_data_info_queue, queue_name)
[文档] @check_save def save(self, file_name, num_files=1, file_type='mindrecord'): """ Save the dynamic data processed by the dataset pipeline in common dataset format. Supported dataset formats: ``'mindrecord'`` only. And you can use :class:`mindspore.dataset.MindDataset` API to read the saved file(s). Implicit type casting exists when saving data as ``'mindrecord'`` . The transform table shows how to do type casting. .. list-table:: Implicit Type Casting when Saving as `mindrecord` :widths: 25 25 50 :header-rows: 1 * - Type in `dataset` - Type in `mindrecord` - Details * - bool - int32 - transform to int32 * - int8 - int32 - * - uint8 - int32 - * - int16 - int32 - * - uint16 - int32 - * - int32 - int32 - * - uint32 - int64 - * - int64 - int64 - * - uint64 - int64 - Maybe reverse * - float16 - float32 - * - float32 - float32 - * - float64 - float64 - * - string - string - Multi-dimensional string not supported * - bytes - bytes - Multi-dimensional bytes not supported Note: 1. To save the samples in order, set dataset's `shuffle` to ``False`` and `num_files` to ``1``. 2. Before calling the function, do not use batch operation, repeat operation or data augmentation operations with random attribute in map operation. 3. When array dimension is variable, one-dimensional arrays or multi-dimensional arrays with variable dimension 0 are supported. 4. MindRecord does not support multi-dimensional string or multi-dimensional bytes. Args: file_name (str): Path to dataset file. num_files (int, optional): Number of dataset files. Default: ``1`` . file_type (str, optional): Dataset format. Default: ``'mindrecord'`` . Examples: >>> import mindspore.dataset as ds >>> import numpy as np >>> >>> def generator_1d(): ... for i in range(10): ... yield (np.array([i]),) >>> >>> # apply dataset operations >>> d1 = ds.GeneratorDataset(generator_1d, ["data"], shuffle=False) >>> d1.save('/path/to/save_file') """ if (_get_enc_key() is not None or _get_hash_mode() is not None) and num_files > 1: raise RuntimeError("When encode mode or hash check is enabled, " + "the automatic sharding function is unavailable.") ir_tree, api_tree = self.create_ir_tree() runtime_context = cde.PythonRuntimeContext() runtime_context.Init() consumer = cde.PythonSaveToDisk(file_name, num_files, file_type) consumer.Init(ir_tree) runtime_context.AssignConsumer(consumer) consumer.Save() if _get_hash_mode() is not None: append_hash_to_file(file_name) append_hash_to_file(file_name + ".db") if _get_enc_key() is not None: encrypt(file_name, _get_enc_key(), _get_enc_mode()) encrypt(file_name + ".db", _get_enc_key(), _get_enc_mode()) _set_dataset_permissions(file_name, num_files) del api_tree
[文档] @check_tuple_iterator def create_tuple_iterator(self, columns=None, num_epochs=-1, output_numpy=False, do_copy=True): """ Create an iterator over the dataset that yields samples of type list, whose elements are the data for each column. Args: columns (list[str], optional): Specify the output columns and the order. Default: ``None``, keep all the output columns and their original order. num_epochs (int, optional): The number of epochs to iterate over the entire dataset. Default: ``-1`` , the dataset can be iterated indefinitely. output_numpy (bool, optional): Whether to keep the output data as NumPy ndarray, or convert it to Tensor. Default: ``False`` . do_copy (bool, optional): Whether to copy the data when converting output to Tensor, or reuse the buffer for better performance, only works when `output_numpy` is ``False`` . Default: ``True`` . Returns: Iterator, a dataset iterator that yields samples of type list. Examples: >>> import mindspore.dataset as ds >>> >>> dataset = ds.GeneratorDataset([i for i in range(10)], "data") >>> num_epochs = 3 >>> iterator = dataset.create_tuple_iterator(num_epochs=num_epochs) >>> for epoch in range(num_epochs): ... for item in iterator: ... # output is of type tuple ... print(type(item)) ... break ... break <class 'list'> """ if output_numpy is None: output_numpy = False if Dataset._noop_mode(): return DummyIterator(self, 'tuple', output_numpy) return TupleIterator(self, columns, num_epochs, output_numpy, do_copy)
[文档] @check_dict_iterator def create_dict_iterator(self, num_epochs=-1, output_numpy=False, do_copy=True): """ Create an iterator over the dataset that yields samples of type dict, while the key is the column name and the value is the data. Args: num_epochs (int, optional): The number of epochs to iterate over the entire dataset. Default: ``-1`` , the dataset can be iterated indefinitely. output_numpy (bool, optional): Whether to keep the output data as NumPy ndarray, or convert it to Tensor. Default: ``False`` . do_copy (bool, optional): Whether to copy the data when converting output to Tensor, or reuse the buffer for better performance, only works when `output_numpy` is ``False`` . Default: ``True`` . Returns: Iterator, a dataset iterator that yields samples of type dict. Examples: >>> import mindspore.dataset as ds >>> >>> dataset = ds.GeneratorDataset([i for i in range(10)], "data") >>> num_epochs = 3 >>> iterator = dataset.create_dict_iterator(num_epochs=num_epochs) >>> for epoch in range(num_epochs): ... for item in iterator: ... # output is of type dict ... print(type(item)) ... break ... break <class 'dict'> """ if output_numpy is None: output_numpy = False if Dataset._noop_mode(): return DummyIterator(self, 'dict', output_numpy) return DictIterator(self, num_epochs, output_numpy, do_copy)
def __iter__(self): """Create an iterator over the dataset.""" return self.create_tuple_iterator(num_epochs=1) @property def input_indexs(self): """ Get the column index, which represents the corresponding relationship between the data column order and the network when using the sink mode. Returns: int, tuple of the input index information. Examples: >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> # set input_indexs >>> dataset.input_indexs = 10 >>> print(dataset.input_indexs) 10 """ if self._input_indexs != (): return self._input_indexs # find input_indexes of children children_input_index = [child.input_indexs for child in self.children] # in case of more than one child, return the first input_indexes for cix in children_input_index: if cix != (): return cix # if all children's input_indexes are () or the node is a leaf return self._input_indexs @input_indexs.setter def input_indexs(self, value): self._input_indexs = value def copy_batch_size(self, value): self._batch_size = value def _init_tree_getters(self, getter_mode=True): """ Get pipeline information. Args: getter_mode (bool, optional): Whether to build IR tree in pull mode. Default: ``True``. """ ir_tree, api_tree = self.create_ir_tree(getter_mode) runtime_context = cde.PythonRuntimeContext() runtime_context.Init() getter = cde.TreeGetters() getter.Init(ir_tree) runtime_context.AssignConsumer(getter) return getter, runtime_context, api_tree def __init_size_getter(self): """ Get pipeline information. """ ir_tree, api_tree = self.create_ir_tree() runtime_context = cde.PythonRuntimeContext() runtime_context.Init() getter = cde.DatasetSizeGetters() getter.Init(ir_tree) runtime_context.AssignConsumer(getter) return getter, runtime_context, api_tree
[文档] def get_col_names(self): """ Return the names of the columns in dataset. Returns: list, list of column names in the dataset. Examples: >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> col_names = dataset.get_col_names() >>> print(col_names) ['column1'] """ if self._col_names is None: runtime_getter = self._init_tree_getters() self._col_names = runtime_getter[0].GetColumnNames() return self._col_names
[文档] @check_output_shape def output_shapes(self, estimate=False): """ Get the shapes of output data. Args: estimate (bool): If `estimate` is ``False`` , will return the shapes of first data row. Otherwise, will iterate the whole dataset and return the estimated shapes of data row, where dynamic shape is marked as None (used in dynamic data shapes scenario). Default: ``False`` . Returns: list, list of shapes of each column. Examples: >>> import mindspore.dataset as ds >>> import numpy as np >>> >>> def generator1(): ... for i in range(1, 100): ... yield np.ones((16, 83, 83)), np.array([i]) >>> >>> dataset = ds.GeneratorDataset(generator1, ["data1", "data2"]) >>> output_shapes = dataset.output_shapes() >>> print(output_shapes) [[16, 83, 83], [1]] """ # cache single shape if not estimate and self.saved_output_shapes is not None: return self.saved_output_shapes # cache estimate shape if estimate and self.estimated_output_shapes is not None: return self.estimated_output_shapes # We have a hang problem when two-level pipeline with multiprocessing, we need to extend the life cycle # of runtime_context. We found this hang problem only occur on output_types and output_shapes. runtime_getter = self._init_tree_getters() self.runtime_context = runtime_getter[1] api_tree = runtime_getter[2] output_shapes = runtime_getter[0].GetOutputShapes(estimate) del api_tree # Need to terminate the runtime context to avoid the occasional hang problem for # Python (with multiprocessing enabled) in sink mode. self.runtime_context.Terminate() del self.runtime_context if estimate: self.estimated_output_shapes = output_shapes else: self.saved_output_shapes = output_shapes return output_shapes
[文档] def output_types(self): """ Get the types of output data. Returns: list, list of data types. Examples: >>> import mindspore.dataset as ds >>> import numpy as np >>> >>> def generator1(): ... for i in range(1, 100): ... yield np.ones((16, 83, 83)).astype(np.float32), np.array([i]).astype(np.int32) >>> >>> dataset = ds.GeneratorDataset(generator1, ["data1", "data2"]) >>> output_types = dataset.output_types() >>> print(output_types) [dtype('float32'), dtype('int32')] """ if self.saved_output_types is None: runtime_getter = self._init_tree_getters() # We have a hang problem when two-level pipeline with multiprocessing, we need to extend the life cycle # of runtime_context. We found this hang problem only occur on output_types and output_shapes. self.runtime_context = runtime_getter[1] api_tree = runtime_getter[2] self.saved_output_types = runtime_getter[0].GetOutputTypes() del api_tree # Need to terminate the runtime context to avoid the occasional hang problem for # Python (with multiprocessing enabled) in sink mode. self.runtime_context.Terminate() del self.runtime_context return self.saved_output_types
[文档] def get_dataset_size(self): """ Return the number of batches in an epoch. Returns: int, number of batches. Examples: >>> import mindspore.dataset as ds >>> import numpy as np >>> >>> # A generator return 66 samples >>> def generator1(): ... for i in range(66): ... yield np.ones((16, 83, 83)), np.array([i]) >>> >>> dataset = ds.GeneratorDataset(generator1, ["data1", "data2"]) >>> dataset_size = dataset.get_dataset_size() >>> print(dataset_size) 66 """ if self.dataset_size is None: runtime_getter = self.__init_size_getter() self.dataset_size = runtime_getter[0].GetDatasetSize(False) if self.dataset_size == 0: logger.warning("Got 0 sample from dataset pipeline, check if drop all data or load dataset fail.") return self.dataset_size
[文档] def num_classes(self): """ Get the number of classes in a dataset. Returns: int, number of classes. Examples: >>> import mindspore.dataset as ds >>> # Read image files >>> image_folder_dataset_dir = "/path/to/image_folder_dataset_directory" >>> dataset = ds.ImageFolderDataset(dataset_dir=image_folder_dataset_dir) >>> # Check how many classes exist in image folder >>> num_classes = dataset.num_classes() """ if self._num_classes is None: runtime_getter = self._init_tree_getters() self._num_classes = runtime_getter[0].GetNumClasses() if self._num_classes == -1: return None return self._num_classes
def get_sync_notifiers(self): if self.children: return self.children[0].get_sync_notifiers() return {} def disable_sync(self): if self.children: return self.children[0].disable_sync() return {} def is_sync(self): if self.children: return self.children[0].is_sync() return False
[文档] def sync_update(self, condition_name, num_batch=None, data=None): """ Release a blocking condition and trigger callback with given data. Args: condition_name (str): The condition name that is used to toggle sending next row. num_batch (Union[int, None]): The number of batches (rows) that are released. When `num_batch` is ``None``, it will default to the number specified by the `sync_wait` operation. Default: ``None``. data (Any): The data passed to the callback, user defined. Default: ``None``. Examples: >>> import numpy as np >>> import mindspore.dataset as ds >>> >>> def gen(): ... for i in range(100): ... yield (np.array(i),) >>> >>> class Augment: ... def __init__(self, loss): ... self.loss = loss ... ... def preprocess(self, input_): ... return input_ ... ... def update(self, data): ... self.loss = data["loss"] >>> >>> batch_size = 10 >>> dataset = ds.GeneratorDataset(gen, column_names=["input"]) >>> aug = Augment(0) >>> dataset = dataset.sync_wait(condition_name='', num_batch=1) >>> dataset = dataset.map(input_columns=["input"], operations=[aug.preprocess]) >>> dataset = dataset.batch(batch_size) >>> >>> count = 0 >>> for data in dataset.create_dict_iterator(num_epochs=1, output_numpy=True): ... count += 1 ... data = {"loss": count} ... dataset.sync_update(condition_name="", data=data) """ if (not isinstance(num_batch, int) and num_batch is not None) or \ (isinstance(num_batch, int) and num_batch <= 0): # throwing exception, disable all sync_wait in pipeline self.disable_sync() raise RuntimeError("Sync_update batch size can only be positive integer, got : {}.".format(num_batch)) notifiers_dict = self.get_sync_notifiers() if not isinstance(condition_name, str): raise TypeError("Argument condition_name with value {} is not of type str, but got {}." .format(condition_name, type(condition_name))) if condition_name not in notifiers_dict: # throwing exception, disable all sync_wait in pipeline self.disable_sync() raise RuntimeError("Condition name not found.") if num_batch is not None: num_batch *= self.get_batch_size() notifiers_dict[condition_name](num_batch, data)
[文档] def get_batch_size(self): """ Return the size of batch. Returns: int, the batch size of data. Examples: >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> dataset = dataset.batch(2) >>> batch_size = dataset.get_batch_size() >>> print(batch_size) 2 """ if self._batch_size is None: runtime_getter = self._init_tree_getters() self._batch_size = runtime_getter[0].GetBatchSize() if self._batch_size is None: self._batch_size = 1 return self._batch_size
[文档] def get_repeat_count(self): """ Get the replication times in RepeatDataset. Default: ``1`` . Returns: int, the count of repeat. Examples: >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> dataset = dataset.repeat(5) >>> repeat_count = dataset.get_repeat_count() >>> print(repeat_count) 5 """ if self._repeat_count is None: runtime_getter = self._init_tree_getters() self._repeat_count = runtime_getter[0].GetRepeatCount() if self._repeat_count is None: self._repeat_count = 1 return self._repeat_count
[文档] def get_class_indexing(self): """ Get the mapping dictionary from category names to category indexes. This dictionary can be used to look up which category name corresponds to a particular category index. Returns: Dict[str, int], the mappings from category names to category indexes. Examples: >>> import mindspore.dataset as ds >>> # Read image files >>> image_folder_dataset_dir = "/path/to/image_folder_dataset_directory" >>> dataset = ds.ImageFolderDataset(dataset_dir=image_folder_dataset_dir) >>> # Check how many classes exist in image folder >>> class_indexing = dataset.get_class_indexing() """ if self.children: return self.children[0].get_class_indexing() return {}
[文档] def reset(self): """ Reset the dataset for next epoch. Examples: >>> import mindspore.dataset as ds >>> mind_dataset_dir = ["/path/to/mind_dataset_file"] >>> dataset = ds.MindDataset(dataset_files=mind_dataset_dir) >>> for _ in range(5): ... num_iter = 0 ... for data in dataset.create_tuple_iterator(num_epochs=1, output_numpy=True): ... num_iter += 1 ... dataset.reset() """
def is_shuffled(self): """Returns True if the dataset or its children is shuffled.""" for input_dataset in self.children: if input_dataset.is_shuffled(): return True return False def is_sharded(self): """Returns True if the dataset or its children is sharded.""" for input_dataset in self.children: if input_dataset.is_sharded(): return True return False def parse(self, children=None): raise NotImplementedError("Dataset has to implement parse method.") def __len__(self): """ Get the length of dataset. Returns: int, the length of dataset. """ return self.get_dataset_size() @staticmethod def _update_data_shard(num_shards, shard_id): """ Update the shard number and shard id if necessary. This is normally used in distributed training mode like Parameter Server training. """ # If this is in distributed execution mode, # the shard number and shard id might need to be updated according to the process's rank or role. worker_num = _get_ps_context("worker_num") server_num = _get_ps_context("server_num") if _is_role_pserver() and _enable_distributed_mindrt() and (worker_num != server_num): num_shards = worker_num shard_id = 0 return num_shards, shard_id def pre_parse(self, getter_mode): if getter_mode: if hasattr(self, "python_multiprocessing"): self.python_multiprocessing = False if hasattr(self, "num_parallel_workers"): self.num_parallel_workers = 1 def post_parse(self, ir_node): if self.cache: ir_node = ir_node.set_cache_client(self.cache.cache_client) if self.num_parallel_workers: ir_node = ir_node.set_num_workers(self.num_parallel_workers) return ir_node def set_init_step(self, init_step): self._global_step = init_step def get_init_step(self): if self._global_step is not None: return self._global_step if len(self.children) == 1: return self.children[0].get_init_step() # When there are multiple children, we cannot tell from which child to get the initial step, # so we initialize from the beginning return 0 class VisionBaseDataset(Dataset): """ Abstract class to represent a vision source dataset which produces content to the data pipeline. """ def __init__(self, children=None, num_parallel_workers=None, cache=None): super().__init__(children=children, num_parallel_workers=num_parallel_workers, cache=cache) def parse(self, children=None): raise NotImplementedError("Dataset has to implement parse method.") class TextBaseDataset(Dataset): """ Abstract class to represent a text source dataset which produces content to the data pipeline. """ def __init__(self, children=None, num_parallel_workers=None, cache=None): super().__init__(children=children, num_parallel_workers=num_parallel_workers, cache=cache) def parse(self, children=None): raise NotImplementedError("Dataset has to implement parse method.") def build_vocab(self, columns, freq_range, top_k, special_tokens, special_first): """ Function to create a Vocab from source dataset. Desired source dataset is a text type dataset. Build a vocab from a dataset. This would collect all the unique words in a dataset and return a vocab which contains top_k most frequent words (if top_k is specified). Note: mindspore.dataset.Dataset.build_vocab is deprecated from version 2.0 and will be removed in a future version. Use mindspore.dataset.text.Vocab.from_dataset instead. Args: columns(Union[str, list[str]]): Column names to get words from. freq_range(tuple[int]): A tuple of integers (min_frequency, max_frequency). Words within the frequency range will be stored. Naturally 0 <= min_frequency <= max_frequency <= total_words. min_frequency/max_frequency can be set to default, which corresponds to 0/total_words separately. top_k(int): Number of words to be built into vocab. top_k most frequent words are taken. The top_k is taken after freq_range. If not enough top_k, all words will be taken special_tokens(list[str]): A list of strings, each one is a special token. special_first(bool): Whether special_tokens will be prepended/appended to vocab, If special_tokens is specified and special_first is set to default, special_tokens will be prepended. Returns: Vocab, vocab built from the dataset. """ warnings.warn("mindspore.dataset.Dataset.build_vocab is deprecated from version 2.0 " "and will be removed in a future version. " "Use mindspore.dataset.text.Vocab.from_dataset instead.", DeprecationWarning) def build_sentencepiece_vocab(self, columns, vocab_size, character_coverage, model_type, params): """ Function to create a SentencePieceVocab from source dataset. Desired source dataset is a text type dataset. Note: mindspore.dataset.Dataset.build_sentencepiece_vocab is deprecated from version 2.0 and will be removed in a future version. Use mindspore.dataset.text.SentencePieceVocab.from_dataset instead. Args: columns(list[str]): Column names to get words from. vocab_size(int): Vocabulary size. character_coverage(float): Percentage of characters covered by the model, must be between 0.98 and 1.0 Good defaults are: 0.9995 for languages with rich character sets like Japanese or Chinese character sets, and 1.0 for other languages with small character sets like English or Latin. model_type(SentencePieceModel): Model type. Choose from unigram (default), bpe, char, or word. The input sentence must be pretokenized when using word type. params(dict): Any extra optional parameters of sentencepiece library according to your raw data Returns: SentencePieceVocab, vocab built from the dataset. """ warnings.warn("mindspore.dataset.Dataset.build_sentencepiece_vocab is deprecated from version 2.0 " "and will be removed in a future version. " "Use mindspore.dataset.text.SentencePieceVocab.from_dataset instead.", DeprecationWarning) def _build_vocab(self, columns, freq_range, top_k, special_tokens, special_first): """ Function to create a Vocab from source dataset. Desired source dataset is a text type dataset. Build a vocab from a dataset. This would collect all the unique words in a dataset and return a vocab which contains top_k most frequent words (if top_k is specified). Args: columns(Union[str, list[str]]): Column names to get words from. freq_range(tuple[int]): A tuple of integers (min_frequency, max_frequency). Words within the frequency range will be stored. Naturally 0 <= min_frequency <= max_frequency <= total_words. min_frequency/max_frequency can be set to default, which corresponds to 0/total_words separately. top_k(int): Number of words to be built into vocab. top_k most frequent words are taken. The top_k is taken after freq_range. If not enough top_k, all words will be taken special_tokens(list[str]): A list of strings, each one is a special token. special_first(bool): Whether special_tokens will be prepended/appended to vocab, If special_tokens is specified and special_first is set to default, special_tokens will be prepended. Returns: Vocab, vocab built from the dataset. """ vocab = cde.Vocab() columns = replace_none(columns, []) if not isinstance(columns, list): columns = [columns] freq_range = replace_none(freq_range, (0, 9223372036854775807)) if freq_range[0] is None: freq_range = (0, freq_range[1]) if freq_range[1] is None: freq_range = (freq_range[0], 9223372036854775807) special_tokens = replace_none(special_tokens, []) top_k = replace_none(top_k, 9223372036854775807) ir_tree, api_tree = self.create_ir_tree() # vocab node vocab_node = cde.BuildVocabNode(ir_tree, vocab, columns, freq_range, top_k, special_tokens, special_first) runtime_context = cde.PythonRuntimeContext() runtime_context.Init() # build vocab consumer = cde.PythonBuildVocabConsumer() consumer.Init(vocab_node) runtime_context.AssignConsumer(consumer) consumer.Start() del api_tree return vocab def _build_sentencepiece_vocab(self, columns, vocab_size, character_coverage, model_type, params): """ Function to create a SentencePieceVocab from source dataset. Desired source dataset is a text type dataset. Args: columns(list[str]): Column names to get words from. vocab_size(int): Vocabulary size. character_coverage(float): Percentage of characters covered by the model, must be between 0.98 and 1.0 Good defaults are: 0.9995 for languages with rich character sets like Japanese or Chinese character sets, and 1.0 for other languages with small character sets like English or Latin. model_type(SentencePieceModel): Model type. Choose from unigram (default), bpe, char, or word. The input sentence must be pretokenized when using word type. params(dict): Any extra optional parameters of sentencepiece library according to your raw data Returns: SentencePieceVocab, vocab built from the dataset. """ if not isinstance(model_type, SentencePieceModel): raise TypeError("Argument model_type with value {0} is not of type SentencePieceModel, but got {1}." \ .format(model_type, type(model_type))) model_type = DE_C_INTER_SENTENCEPIECE_MODE[model_type] vocab = cde.SentencePieceVocab() ir_tree, api_tree = self.create_ir_tree() # vocab node vocab_node = cde.BuildSentenceVocabNode(ir_tree, vocab, columns, vocab_size, character_coverage, model_type, params) runtime_context = cde.PythonRuntimeContext() runtime_context.Init() # build vocab consumer = cde.PythonBuildVocabConsumer() consumer.Init(vocab_node) runtime_context.AssignConsumer(consumer) consumer.Start() del api_tree return vocab class AudioBaseDataset(Dataset): """ Abstract class to represent a audio source dataset which produces content to the data pipeline. """ def __init__(self, children=None, num_parallel_workers=None, cache=None): super().__init__(children=children, num_parallel_workers=num_parallel_workers, cache=cache) def parse(self, children=None): raise NotImplementedError("Dataset has to implement parse method.") class UnionBaseDataset(VisionBaseDataset, TextBaseDataset, AudioBaseDataset): """ Abstract class to represent a union source dataset which produces content to the data pipeline. """ def __init__(self, children=None, num_parallel_workers=None, cache=None): super().__init__(children=children, num_parallel_workers=num_parallel_workers, cache=cache) def parse(self, children=None): raise NotImplementedError("Dataset has to implement parse method.") class SourceDataset(Dataset): """ Abstract class to represent a source dataset which produces content to the data pipeline. """ def __init__(self, num_parallel_workers=None, num_samples=None, shuffle=True, num_shards=None, shard_id=None, cache=None): super().__init__(num_parallel_workers=num_parallel_workers, cache=cache) self.num_samples = replace_none(num_samples, 0) self.num_shards = replace_none(num_shards, 1) self.shard_id = replace_none(shard_id, 0) if shuffle is not None and not isinstance(shuffle, (bool, Shuffle)): raise TypeError("shuffle must be of boolean or enum of 'Shuffle' values like 'Shuffle.GLOBAL' or " "'Shuffle.FILES' or 'Shuffle.INFILE'.") self.shuffle_flag = 2 # Global shuffle if not isinstance(shuffle, Shuffle): if shuffle is None or shuffle: self.shuffle_flag = 2 # Global shuffle else: self.shuffle_flag = 0 # No shuffle else: if shuffle == Shuffle.GLOBAL: self.shuffle_flag = 2 # Global shuffle elif shuffle == Shuffle.FILES: self.shuffle_flag = 1 # Files shuffle elif shuffle == Shuffle.INFILE: self.shuffle_flag = 3 # Infile shuffle def parse(self, children=None): raise NotImplementedError("Dataset has to implement parse method.") @staticmethod def _find_files(patterns): """ Utility function to search for files with the given glob patterns. Args: patterns (Union[str, list[str]]): String or list of patterns to be searched. Returns: list, list of files. """ if not isinstance(patterns, list): patterns = [patterns] file_list = [] unmatched_patterns = [] for pattern in patterns: matches = [match for match in glob.glob(pattern, recursive=True) if os.path.isfile(match)] if matches: file_list.extend(matches) else: unmatched_patterns.append(pattern) if unmatched_patterns: raise ValueError("The following patterns did not match any files: {}.".format(unmatched_patterns)) if file_list: # not empty return file_list raise ValueError("The list of path names matching the patterns is empty.") def is_shuffled(self): return self.shuffle_flag > 0 def is_sharded(self): if self.num_shards is not None: return self.num_shards > 1 return False class MappableDataset(SourceDataset): """ Abstract class to represent a source dataset which supports use of samplers. """ def parse(self, children=None): raise NotImplementedError("Dataset has to implement parse method.") def __init__(self, num_parallel_workers=None, sampler=None, num_samples=None, shuffle=None, num_shards=None, shard_id=None, cache=None): num_shards, shard_id = self._update_data_shard(num_shards, shard_id) super().__init__(num_parallel_workers=num_parallel_workers, num_samples=num_samples, shuffle=shuffle, num_shards=num_shards, shard_id=shard_id, cache=cache) self.shuffle_flag = replace_none(shuffle, True) self.sampler = samplers.select_sampler(num_samples, sampler, shuffle, num_shards, shard_id)
[文档] def add_sampler(self, new_sampler): """ Add a child sampler for the current dataset. Args: new_sampler (Sampler): The child sampler to be added. Examples: >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> >>> new_sampler = ds.DistributedSampler(10, 2) >>> dataset.add_sampler(new_sampler) """ # Note: By adding a sampler, the sampled IDs will flow to the new_sampler # after first passing through the current samplers attached to this dataset. self.dataset_size = None new_sampler.add_child(self.sampler) self.sampler = new_sampler
[文档] def use_sampler(self, new_sampler): """ Replace the last child sampler of the current dataset, remaining the parent sampler unchanged. Args: new_sampler (Sampler): The new sampler to replace with. Examples: >>> import mindspore.dataset as ds >>> dataset = ds.GeneratorDataset([i for i in range(10)], "column1") >>> >>> # use a DistributedSampler instead >>> new_sampler = ds.DistributedSampler(10, 2) >>> dataset.use_sampler(new_sampler) """ if new_sampler is None: raise TypeError("Input sampler can not be None.") if not isinstance(new_sampler, (samplers.BuiltinSampler, samplers.Sampler)): raise TypeError("Input sampler is not an instance of a sampler.") self.dataset_size = None self.sampler = self.sampler.child_sampler self.add_sampler(new_sampler)
def is_shuffled(self): return self.sampler.is_shuffled() def is_sharded(self): return self.sampler.is_sharded() @check_split def split(self, sizes, randomize=True): """ Split the dataset into smaller, non-overlapping datasets. Args: sizes (Union[list[int], list[float]]): If a list of integers [s1, s2, …, sn] is provided, the dataset will be split into n datasets of size s1, size s2, …, size sn respectively. If the sum of all sizes does not equal the original dataset size, an error will occur. If a list of floats [f1, f2, …, fn] is provided, all floats must be between 0 and 1 and must sum to 1, otherwise an error will occur. The dataset will be split into n Datasets of size round(f1*K), round(f2*K), …, round(fn*K) where K is the size of the original dataset. If after rounding: - Any size equals 0, an error will occur. - The sum of split sizes < K, the difference will be added to the first split. - The sum of split sizes > K, the difference will be removed from the first large enough split such that it will have at least 1 row after removing the difference. randomize (bool, optional): Determines whether or not to split the data randomly. Default: ``True``. If ``True``, the data will be randomly split. Otherwise, each split will be created with consecutive rows from the dataset. Note: 1. There is an optimized split function, which will be called automatically when the dataset that calls this function is a MappableDataset. 2. Dataset should not be sharded if split is going to be called. Instead, create a :class:`mindspore.dataset.DistributedSampler` and specify a split to shard after splitting. If the dataset is sharded after a split, it is strongly recommended setting the same seed in each instance of execution, otherwise each shard may not be part of the same split (see Examples). 3. It is strongly recommended to not shuffle the dataset, but set `randomize` to ``True`` instead. Shuffling the dataset may not be deterministic, which means the data in each split will be different in each epoch. Furthermore, if sharding occurs after split, each shard may not be part of the same split. Returns: Tuple[Dataset], a tuple of new datasets split from the original one. Raises: RuntimeError: If get_dataset_size returns None or is not supported for this dataset. RuntimeError: If `sizes` is list of integers and sum of all elements in sizes does not equal the dataset size. RuntimeError: If `sizes` is list of float and there is a split with size 0 after calculations. RuntimeError: If the dataset is sharded prior to calling split. ValueError: If `sizes` is list of float and not all floats are between 0 and 1, or if the floats don't sum to 1. Examples: >>> import mindspore.dataset as ds >>> # Since many datasets have shuffle on by default, set shuffle to False if split will be called! >>> image_folder_dataset_dir = "/path/to/image_folder_dataset_directory" >>> dataset = ds.ImageFolderDataset(image_folder_dataset_dir, shuffle=False) >>> >>> # Set the seed, and tell split to use this seed when randomizing. >>> # This is needed because sharding will be done later >>> ds.config.set_seed(58) >>> train_dataset, test_dataset = dataset.split([0.9, 0.1]) >>> >>> # To shard the train dataset, use a DistributedSampler >>> train_sampler = ds.DistributedSampler(10, 2) >>> train_dataset.use_sampler(train_sampler) """ if self.is_shuffled(): logger.warning("Dataset is shuffled before split.") if self.is_sharded(): raise RuntimeError("Dataset should not be sharded before split.") absolute_sizes = self._get_absolute_split_sizes(sizes) splits = [] current_split_start_index = 0 for size in absolute_sizes: ds = copy.deepcopy(self) ds.dataset_size = None if randomize: # want to shuffle the same way every epoch before split, we are assuming # that the user will call set_seed random_sampler = samplers.RandomSampler() random_sampler.reshuffle_each_epoch = False ds.add_sampler(random_sampler) subset_sampler = samplers.SequentialSampler(current_split_start_index, size) ds.add_sampler(subset_sampler) # add sequential sampler, so that if user calls use_sampler, we will # get rid of the sequential sampler instead of something we need ds.add_sampler(samplers.SequentialSampler()) splits.append(ds) current_split_start_index += size return tuple(splits) class BucketBatchByLengthDataset(UnionBaseDataset): """ The result of applying BucketBatchByLength operation to the input dataset. """ def __init__(self, input_dataset, column_names, bucket_boundaries, bucket_batch_sizes, element_length_function, pad_info, pad_to_bucket_boundary, drop_remainder): super().__init__(children=input_dataset) self.column_names = to_list(column_names) self.bucket_boundaries = replace_none(bucket_boundaries, []) self.bucket_batch_sizes = replace_none(bucket_batch_sizes, []) self.element_length_function = element_length_function self.pad_info = replace_none(pad_info, {}) self.pad_to_bucket_boundary = replace_none(pad_to_bucket_boundary, False) self.drop_remainder = replace_none(drop_remainder, False) def parse(self, children=None): return cde.BucketBatchByLengthNode(children[0], self.column_names, self.bucket_boundaries, self.bucket_batch_sizes, self.element_length_function, self.pad_info, self.pad_to_bucket_boundary, self.drop_remainder) def _check_shm_usage(num_worker, queue_size, in_rowsize, out_rowsize): """ Check sufficient shared memory is available for shared memory queues when training in parallel mode. """ threshold_ratio = 0.8 # Verify available size only when using static shared memory on Linux if platform.system().lower() not in {"windows", "darwin"} and in_rowsize != -1 and out_rowsize != -1: device_num = _get_device_num() # In the cluster, _get_device_num indicates the number of the entire cluster. The maximum number of cards # on the ascend server is 8. if device_num > 1: device_num = min(device_num, 8) shm_estimate_usage = device_num * num_worker * \ (queue_size + 2) * (in_rowsize + out_rowsize) * 1024 * 1024 try: shm_available = psutil.disk_usage('/dev/shm').free if shm_estimate_usage >= threshold_ratio * shm_available: raise RuntimeError( "Insufficient shared memory available. Required: {}, Available: {}. " "The required memory can't exceed 80% of the available shared memory, " "it's recommended to reduce memory usage by following methods:\n" "1. reduce value of parameter max_rowsize or num_parallel_workers.\n" "2. reduce prefetch size by set_prefetch_size().\n" "3. disable shared memory by set_enable_shared_mem().".format(shm_estimate_usage, shm_available)) except FileNotFoundError: raise RuntimeError("Expected /dev/shm to exist.") class BatchDataset(UnionBaseDataset): """ The result of applying Batch operation to the input dataset. Args: input_dataset (Dataset): Input Dataset to be batched. batch_size (Union[int, function]): The number of rows each batch is created with. An int or callable which takes exactly 1 parameter, BatchInfo. drop_remainder (bool, optional): Determines whether or not to drop the last possibly incomplete batch. Default: ``False``. If True, and if there are less than batch_size rows available to make the last batch, then those rows will be dropped and not propagated to the child node. num_parallel_workers (int, optional): Number of workers to process the dataset in parallel. Default: ``None``. per_batch_map (callable, optional): Per batch map callable. A callable which takes (list[Tensor], list[Tensor], ..., BatchInfo) as input parameters. Each list[Tensor] represents a batch of Tensors on a given column. The number of lists should match with number of entries in input_columns. The last parameter of the callable must always be a BatchInfo object. input_columns (Union[str, list[str]], optional): List of names of the input columns. The size of the list must match with signature of per_batch_map callable. output_columns (Union[str, list[str]], optional): List of names assigned to the columns outputted by the last operation. This parameter is mandatory if len(input_columns) != len(output_columns). The size of this list must match the number of output columns of the last operation. Default: ``None``, output columns will have the same name as the input columns, i.e., the columns will be replaced. max_rowsize(Union[int, list[int]], optional): Maximum size of row in MB that is used for shared memory allocation to copy data between processes, the total occupied shared memory will increase as ``num_parallel_workers`` and :func:`mindspore.dataset.config.set_prefetch_size` increase. If set to -1, shared memory will be dynamically allocated with the actual size of data. This is only used if ``python_multiprocessing`` is set to True. If it is an int value, it represents ``input_columns`` and ``output_columns`` use this value as the unit to create shared memory. If it is a list, the first element represents the ``input_columns`` use this value as the unit to create shared memory, and the second element represents ``output_columns`` use this value as the unit to create shared memory. Default: 16. """ def __init__(self, input_dataset, batch_size, drop_remainder=False, num_parallel_workers=None, per_batch_map=None, input_columns=None, output_columns=None, python_multiprocessing=False, max_rowsize=16): super().__init__(children=input_dataset, num_parallel_workers=num_parallel_workers) if BatchDataset._is_ancestor_of_repeat(input_dataset): logger.warning("Repeat is located before batch, data from two epochs can be batched together.") BatchDataset._update_batch_size_for_syncwait(input_dataset, batch_size) # if batch_size is callable, set batch_size to 1 and batch_size_func to that callable function self.batch_size = batch_size if not callable(batch_size) else 1 self.batch_size_func = None if not callable(batch_size) else batch_size self.drop_remainder = replace_none(drop_remainder, False) self.per_batch_map = per_batch_map self.input_columns = to_list(input_columns) self.output_columns = to_list(output_columns) self.python_multiprocessing = python_multiprocessing self.process_pool = None if isinstance(max_rowsize, int): self.max_rowsize = [max_rowsize * self.batch_size] * 2 if max_rowsize != -1 else [max_rowsize, max_rowsize] else: self.max_rowsize = [max_rowsize[0] * self.batch_size, max_rowsize[1] * self.batch_size] def __del__(self): if hasattr(self, "process_pool") and self.process_pool is not None: self.process_pool.terminate() del self.process_pool def parse(self, children=None): return cde.BatchNode(children[0], self.batch_size, self.drop_remainder, False, self.input_columns, self.output_columns, self.batch_size_func, self.per_batch_map, {}, self.process_pool) @staticmethod def _is_ancestor_of_repeat(dataset): """ Utility function to find the case where repeat is used before batch. Args: dataset (Dataset): Dataset to be checked. Returns: bool, whether repeat is used before batch. """ if isinstance(dataset, RepeatDataset): return True flag = False for input_dataset in dataset.children: flag = flag | BatchDataset._is_ancestor_of_repeat(input_dataset) return flag @staticmethod def _update_batch_size_for_syncwait(dataset, batch_size): """ Utility function to notify batch size to sync_wait. Args: dataset (Dataset): Dataset to be checked. batch_size (int): batch size to notify. """ if isinstance(dataset, SyncWaitDataset): dataset.update_sync_batch_size(batch_size) for input_dataset in dataset.children: BatchDataset._update_batch_size_for_syncwait(input_dataset, batch_size) def __deepcopy__(self, memodict): return self.__safe_deepcopy__(memodict, exclude=("per_batch_map", "batch_size_func", "__transfer_dataset__")) # Iterator bootstrap will be called on iterator construction. # A deep copy of Dataset object is created prior of iterator_bootstrap. # This method will create per iterator process pool and bind pyfunc execution to the pool. def iterator_bootstrap(self): """ Per iterator bootstrap callback. """ if self.python_multiprocessing and platform.system().lower() == 'windows': logger.warning("Python multiprocessing is not supported on Windows platform.") if self.python_multiprocessing and get_debug_mode(): logger.warning("Python multiprocessing is not supported in debug mode." " Ignoring Python multiprocessing for batch operation.") self.python_multiprocessing = False if self.python_multiprocessing and platform.system().lower() != 'windows': if self.per_batch_map is None: logger.warning("per_batch_map is None so python_multiprocessing is ignored for batch.") return # If user didn't specify num_parallel_workers, set it to default if self.num_parallel_workers is None: self.num_parallel_workers = get_num_parallel_workers() self.process_pool = _PythonMultiprocessing(str(self), self.num_parallel_workers, [self.per_batch_map], self.max_rowsize) # Wrap per_batch_map into _PythonCallable self.per_batch_map = _PythonCallable(self.per_batch_map, 0, self.process_pool) else: if self.per_batch_map is not None: self.per_batch_map = FuncWrapper(self.per_batch_map)
[文档]class BatchInfo(cde.CBatchInfo): """ This class helps to get dataset information dynamically when the input of `batch_size` or `per_batch_map` in `batch` operation is a callable object. """
[文档] def get_batch_num(self): """ Return the batch number being processed in current epoch, start from 0. Examples: >>> # Create a dataset where its batch size is dynamic >>> # Define a callable batch size function and let batch size increase 1 each time. >>> import mindspore.dataset as ds >>> from mindspore.dataset import BatchInfo >>> >>> dataset = ds.GeneratorDataset([i for i in range(3)], "column1", shuffle=False) >>> def add_one(BatchInfo): ... return BatchInfo.get_batch_num() + 1 >>> dataset = dataset.batch(batch_size=add_one) >>> print(list(dataset)) [[Tensor(shape=[1], dtype=Int64, value= [0])], [Tensor(shape=[2], dtype=Int64, value= [1, 2])]] """ return
[文档] def get_epoch_num(self): """ Return the epoch number, start from 0. Examples: >>> # Create a dataset where its batch size is dynamic >>> # Define a callable batch size function and let batch size increase 1 each epoch. >>> import mindspore.dataset as ds >>> from mindspore.dataset import BatchInfo >>> >>> dataset = ds.GeneratorDataset([i for i in range(4)], "column1", shuffle=False) >>> def add_one_by_epoch(BatchInfo): ... return BatchInfo.get_epoch_num() + 1 >>> dataset = dataset.batch(batch_size=add_one_by_epoch) >>> >>> result = [] >>> epoch = 2 >>> iterator = dataset.create_tuple_iterator(num_epochs=epoch) >>> for i in range(epoch): ... result.extend(list(iterator)) >>> # result: >>> # [[Tensor(shape=[1], dtype=Int64, value= [0])], [Tensor(shape=[1], dtype=Int64, value= [1])], >>> # [Tensor(shape=[1], dtype=Int64, value= [2])], [Tensor(shape=[1], dtype=Int64, value= [3])], >>> # [Tensor(shape=[2], dtype=Int64, value= [0, 1])], [Tensor(shape=[2], dtype=Int64, value= [2, 3])]] """ return
class BlockReleasePair: """ The blocking condition class used by SyncWaitDataset. Args: init_release_rows (int): Number of lines to allow through the pipeline. callback (function): The callback function that will be called when release is called. Default: ``None``. """ def __init__(self, init_release_rows, callback=None): if isinstance(init_release_rows, int) and init_release_rows <= 0: raise ValueError("release_rows need to be greater than 0.") self.row_count = -init_release_rows self.cv = threading.Condition() self.callback = callback self.default_rows = init_release_rows self.disable = False def __deepcopy__(self, memodict): return self def reset(self): with self.cv: self.row_count = -self.default_rows self.cv.notify_all() def update_batched_size(self, batch_size): # sanity check if isinstance(batch_size, int) and batch_size <= 0: raise ValueError("batch_size need to be greater than 0.") # should only use before the pipeline creates self.row_count *= batch_size self.default_rows *= batch_size def block_func(self): """ Function for handing blocking condition. Returns: bool, True. """ with self.cv: # if disable is true, the always evaluate to true not_time_out = self.cv.wait_for(lambda: (self.row_count < 0 or self.disable), timeout=get_callback_timeout()) # time_out will be False if time out occurs if not not_time_out: logger.warning("Timeout happened in sync_wait, maybe dataset.sync_update(condition=...) " "is not added after dataset.create_dict_iterator(...), now disabling lock.") self.disable = True self.row_count += 1 return True def release_func(self, pass_rows=None, data=None): with self.cv: if pass_rows is None: pass_rows = self.default_rows self.row_count -= pass_rows if self.callback is not None: self.callback(data) self.cv.notify_all() def disable_lock(self): with self.cv: self.disable = True self.cv.notify_all() class PaddedBatchDataset(UnionBaseDataset): """ The result of applying Batch operation to the input dataset. Args: input_dataset (Dataset): Input Dataset to be batched. batch_size (Union[int, function]): The number of rows each batch is created with. An int or callable which takes exactly 1 parameter, BatchInfo. drop_remainder (bool, optional): Determines whether or not to drop the last possibly incomplete batch. Default: ``False``. If True, and if there are less than batch_size rows available to make the last batch, then those rows will be dropped and not propagated to the child node. num_parallel_workers (int, optional): Number of workers to process the dataset in parallel. Default: ``None``. pad_info (dict, optional): Whether to perform padding on selected columns. pad_info={"col1":([224,224],0)} will pad column with name "col1" to a tensor of size [224,224] and fill the missing with 0. """ def __init__(self, input_dataset, batch_size, drop_remainder=False, num_parallel_workers=None, pad_info=None): super().__init__(children=input_dataset, num_parallel_workers=num_parallel_workers) if PaddedBatchDataset._is_ancestor_of_repeat(input_dataset): logger.warning("Repeat is located before padded_batch, data from two epochs can be batched together.") PaddedBatchDataset._update_batch_size_for_syncwait(input_dataset, batch_size) # if batch_size is callable, set batch_size to 1 and batch_size_func to that callable function self.batch_size = batch_size if not callable(batch_size) else 1 self.batch_size_func = None if not callable(batch_size) else batch_size self.drop_remainder = replace_none(drop_remainder, False) self.pad = bool(pad_info is not None) self.pad_info = replace_none(pad_info, dict()) def parse(self, children=None): return cde.BatchNode(children[0], self.batch_size, self.drop_remainder, self.pad, [], [], self.batch_size_func, None, self.pad_info, None) @staticmethod def _is_ancestor_of_repeat(dataset): """ Utility function to find the case where repeat is used before batch. Args: dataset (Dataset): Dataset to be checked. Returns: bool, whether repeat is used before batch. """ if isinstance(dataset, RepeatDataset): return True flag = False for input_dataset in dataset.children: flag = flag | PaddedBatchDataset._is_ancestor_of_repeat(input_dataset) return flag @staticmethod def _update_batch_size_for_syncwait(dataset, batch_size): """ Utility function to notify batch size to sync_wait. Args: dataset (Dataset): Dataset to be checked. batch_size (int): batch size to notify. """ if isinstance(dataset, SyncWaitDataset): dataset.update_sync_batch_size(batch_size) for input_dataset in dataset.children: PaddedBatchDataset._update_batch_size_for_syncwait(input_dataset, batch_size) def __deepcopy__(self, memodict): return self.__safe_deepcopy__(memodict, exclude=("batch_size_func", "__transfer_dataset__")) class SyncWaitDataset(UnionBaseDataset): """ The result of adding a blocking condition to the input Dataset. Args: input_dataset (Dataset): Input dataset to apply flow control. num_batch (int): Number of batches without blocking at the start of each epoch. condition_name (str): Condition name that is used to toggle sending next row. callback (function): Callback function that will be invoked when sync_update is called. Default: ``None``. Raises: RuntimeError: If condition name already exists. """ def __init__(self, input_dataset, condition_name, num_batch, callback=None): super().__init__(children=input_dataset) # set to the default value, waiting for the batch to update it self._condition_name = condition_name if isinstance(num_batch, int) and num_batch <= 0: raise ValueError("num_batch need to be greater than 0.") self._pair = BlockReleasePair(num_batch, callback) if self._condition_name in self.children[0].get_sync_notifiers(): raise RuntimeError("Condition name is already in use.") logger.info("Please remember to add dataset.sync_update(condition=%s), otherwise hanging will result. " "If dataset.sync_update(condition=%s) has already been added, you can ignore the info.", condition_name, condition_name) def parse(self, children=None): return cde.SyncWaitNode(children[0], self._condition_name, self._pair.block_func) def get_sync_notifiers(self): return {**self.children[0].get_sync_notifiers(), **{self._condition_name: self._pair.release_func}} def is_sync(self): return True def update_sync_batch_size(self, batch_size): if isinstance(batch_size, int) and batch_size <= 0: raise ValueError("num_batch need to be greater than 0.") self._pair.update_batched_size(batch_size) def disable_sync(self): logger.info("Disabling Sync") self._pair.disable_lock() @staticmethod def _is_ancestor_of_batch(dataset): """ Utility function to find the case where sync_wait is used before batch. Args: dataset (Dataset): Dataset to be checked. Returns: bool, whether sync_wait is used before batch. """ if isinstance(dataset, (BatchDataset, PaddedBatchDataset)): return True flag = False for input_dataset in dataset.children: flag = flag | SyncWaitDataset._is_ancestor_of_batch(input_dataset) return flag def iterator_bootstrap(self): self._pair.reset() class ShuffleDataset(UnionBaseDataset): """ The result of applying Shuffle operation to the input Dataset. Args: input_dataset (Dataset): Input Dataset to be shuffled. buffer_size (int): Size of the buffer. Raises: RuntimeError: If exist sync operations before shuffle. """ def __init__(self, input_dataset, buffer_size): super().__init__(children=input_dataset) self.buffer_size = buffer_size self.reshuffle_each_epoch = True if self.is_sync(): raise RuntimeError("No shuffle after sync operators.") def parse(self, children=None): return cde.ShuffleNode(children[0], self.buffer_size, self.reshuffle_each_epoch) def is_shuffled(self): return True # Pyfunc collection for multiprocess pyfunc # This global variable will only be used within subprocesses _OP_NAME = dict() _OP_PROCESS = dict() # PythonCallable wrapper for multiprocess pyfunc class _PythonCallable: """ Internal Python function wrapper for multiprocessing pyfunc. """ def __init__(self, py_callable, idx, pool=None): # Original Python callable from user. self.py_callable = py_callable # Process pool created for current iterator. self.pool = pool # Python callable index self.idx = idx def __call__(self, *args): result = None get_data_from_worker_process = False while get_data_from_worker_process is False: if self.pool.is_running() and check_iterator_cleanup() is False: try: result = self.pool.execute(self.idx, *args) except multiprocessing.TimeoutError: continue get_data_from_worker_process = True else: # worker process is stopped logger.info("The worker process of map operation is stopped. " "So return None to main thread and break the main thread.") return None # got value from worker process if not isinstance(result, tuple) and get_data_from_worker_process is True: result = (result,) return result def to_json(self): return self.py_callable.to_json() # used when python_multiprocessing=True in map class Pipe: """ Class to handle communication between the master process and the worker processes. """ def __init__(self, warning_ctl, shared_memory=False, max_rowsize=16): self.shared_memory = shared_memory self.eof = multiprocessing.Event() if self.shared_memory: self.in_queue = _SharedQueue(1, warning_ctl, max_rowsize=max_rowsize[0]) self.res_queue = _SharedQueue(1, warning_ctl, max_rowsize=max_rowsize[1]) else: self.in_queue = _Queue(1) self.res_queue = _Queue(1) self.in_queue.cancel_join_thread() # Ensure that the process does not hung when exiting def master_send(self, func_index, data): self.in_queue.put_nowait((func_index, *data)) def master_receive(self): if self.eof is None: raise RuntimeError("EOF is none when get data from worker.") if self.eof.is_set(): return None return self.res_queue.get(timeout=1) def master_close(self): self.eof.set() self.send_finish_signal_to_worker() self.send_finish_signal() def send_finish_signal(self): self.worker_send(None) def send_finish_signal_to_worker(self): self.master_send(0, "QUIT") def worker_send(self, data): self.res_queue.put_until(data, timeout=1, exit_signal=self.eof) def worker_receive(self): result = self.in_queue.get_until(timeout=1, exit_signal=self.eof) if result is None: return result if len(result) == 1: raise RuntimeError(f"Corrupted data. Worker received {len(result)} elements, it should be more than 1.") func_index, *data = result return func_index, tuple(data) def _main_process_already_exit(): """ Judge whether main process already exit. """ ppid = os.getppid() if (platform.system().lower() != 'windows' and not _PythonMultiprocessing.is_process_alive(ppid)): return True return False def _worker_loop(operations, pipe, worker_id): """ Multiprocess worker process loop. """ # Ensure that the process does not hung when exiting pipe.res_queue.cancel_join_thread() def _ignore_sigint(): """ We need to ignore sigint signal here so subprocesses can exit normally and clear. """ signal.signal(signal.SIGINT, signal.SIG_IGN) # If the default random seed has not been changed, there is no need to fix the randomness. # Otherwise, set the random seed for each child process to "base_seed + worker_id" to ensure # that the random results of each process are different. if get_seed() != 5489: set_seed(get_seed() + worker_id) while not _main_process_already_exit(): _ignore_sigint() result = pipe.worker_receive() if result is None: return (idx, input_tensors) = result if input_tensors == "QUIT": break try: output_tensors = operations[idx](*input_tensors) pipe.worker_send(output_tensors) except Exception: pipe.worker_send(ExceptionHandler(where="in map(or batch) worker and execute Python function")) # Do not return # release the queue when stop the worker by master del pipe.in_queue del pipe.res_queue def worker_target(operations, worker_id): return lambda pipe: _worker_loop(operations, pipe, worker_id) class _MPWorker(multiprocessing.Process): """ Worker process for multiprocessing. """ def __init__(self, operations, warning_ctl, max_rowsize=16, worker_id=0): shared_memory = get_enable_shared_mem() self.pipe = Pipe(warning_ctl, shared_memory=shared_memory, max_rowsize=max_rowsize) self.check_interval = get_multiprocessing_timeout_interval() super().__init__(target=worker_target(operations, worker_id), name="MapWorker" + str(worker_id), args=(self.pipe,), daemon=True) def execute(self, idx, *args): """Acquiring data from a worker in an infinite loop""" self.pipe.master_send(idx, args) time_s = time.time() wait_count = 1 while True: cost_time = time.time() - time_s if cost_time / self.check_interval >= wait_count: wait_count += 1 logger.warning("It has been waiting for " + "%.3f" % cost_time + "s because the sub-process " "worker of the map operation is hanging. " "Check whether the user defined data transform is too slow or the " "output data is too large. You can also set the timeout interval by " "ds.config.set_multiprocessing_timeout_interval to adjust the output frequency " "of this log.") pid = self.pid logger.warning("Map worker subprocess ID {} is stuck.".format(pid)) install_status, _ = subprocess.getstatusoutput("py-spy --version") if install_status == 0: stack = subprocess.getoutput("py-spy dump -p {} -l".format(pid)) logger.warning("Map worker subprocess stack:\n{}".format(stack)) else: logger.warning("Please `pip install py-spy` to get the stacks of the stuck process.") try: res = self.pipe.master_receive() # Because there is no need to copy when creating Tensors in the C++layer, it reduces the time # from np.ndarray to C++Tensor creation. However, when using shared memory in multiple processes, # the address of the shared memory will always be passed to subsequent nodes in the dataset pipeline, # and the shared memory will also be written by the current node, causing dirty data to be accessed # by subsequent nodes in the pipeline. So make a memory copy here to solve the problem of # shared memory being contaminated. if get_enable_shared_mem(): res = copy.deepcopy(res) except queue.Empty: continue if res is None: # receive finish signal return None if isinstance(res, ExceptionHandler): res.reraise() return res def close(self): try: if self.is_alive(): # release the eager executor which is used by current process transforms.transforms.clean_unused_executors() logger.info(f"Closing worker with PID: {self.pid}") self.pipe.master_close() # del the handle which hold by master del self.pipe.in_queue del self.pipe.res_queue super().terminate() super().join() super().close() except ValueError: # Process has been closed already return return def is_alive(self): try: return super().is_alive() except ValueError: return False class _PythonMultiprocessing(cde.PythonMultiprocessingRuntime): """ A wrapper to multiprocessing.pool that performs cleanup and ensure proper termination of forked processes. """ class _ExceptHookHandler: """ Internal class ExceptionHandler """ def __init__(self): self.origin_hook = sys.excepthook sys.excepthook = self.__handler_exception @staticmethod def mp_pool_exit_preprocess(): if check_iterator_cleanup() is False: # Set the iterator_cleanup flag to True before exiting, and wait 3s for all apply_async # applied to the multiprocessing task to prevent multiprocessing from hang when exiting _set_iterator_cleanup() time.sleep(3) def __handler_exception(self, ex_type, value, tb): self.origin_hook(ex_type, value, tb) self.mp_pool_exit_preprocess() def __init__(self, op_name, num_parallel_workers, operations, max_rowsize=16): super(_PythonMultiprocessing, self).__init__() self.op_name = op_name self.num_parallel_workers = num_parallel_workers self.operations = operations self.max_rowsize = max_rowsize self.workers = None self.pids = None self.op_id = -1 self.queues_map = {} self.next_queue = 0 self.eot = None self.watch_dog = None self.ppid = os.getpid() self.hook = None self.warning_ctl = None # cache thread (get_ident()) to worker_id mapping in Python layer self.python_threads_to_workers = {} self.eof = None def __del__(self): try: self.terminate() except TypeError: pass # This wait function is for cleaning zombie subprocesses @staticmethod def wait_pid(): """ This function is used by the main process to release subprocess resources. """ try: while True: child_pid, _ = os.waitpid(-1, os.WNOHANG) if child_pid == 0: break except OSError: # waitpid may be failed for some reasons so we ignore this error pass # Dataset need watch_dog thread to monitoring fork multi-processing, # and thread can't be a member function otherwise python won't collect and release resources. @staticmethod def _watch_dog(eot, workers): """ This thread is for monitoring subprocesses forked by GeneratorDataset/map/batch """ if not isinstance(workers, list): raise TypeError("[Internal Error] The 2nd parameter of watch dog thread should be list of process, " "but got {}.".format(type(workers))) while not eot.is_set(): # Monitoring and count how many subprocesses already exit clear_subprocess_timeout = _PythonMultiprocessing._monitor_subprocess_exit(workers) # If find subprocess exit, we will wait for 30s and do some waitpid operations if clear_subprocess_timeout > 0: start = time.time() while time.time() - start < clear_subprocess_timeout: # We need to distinguishing get_dataset_size or train finished normally and hang scenario. # If get_dataset_size or train finished normally, _stop_subprocess can be execute and # self.need_abort can be set to True. If main process is hang in get(), self.need_abort # will never set to True, then we wait for 30s and kill main process if eot.is_set(): return # Sometimes subprocess may be zombie, so in 30s we can wait and do some useful tasks(waitpid). _PythonMultiprocessing.wait_pid() # multiprocessing.queue may hang in .get() forever when put() process was killed. # We have to exit main process otherwise main process will hang. _PythonMultiprocessing._terminate_processes(workers) logger.critical("The subprocess of dataset may exit unexpected or be killed, " "main process will exit. If this is not an artificial operation, you can use " "ds.config.set_enable_watchdog(False) to block this error.") os.kill(os.getpid(), signal.SIGTERM) # release the workers del workers @staticmethod def _terminate_processes(processes): """Terminate subprocesses""" for p in processes: try: if p.exitcode is None: p.terminate() except Exception: # pylint: disable=broad-except # process has been closed already pass for p in processes: if p._closed is False: # pylint: disable=W0212 # We don't use w.join because join can only used in main process or join will raise an error. p._popen.wait() # pylint: disable=W0212 # Monitor the exit number of subprocesses @staticmethod def _monitor_subprocess_exit(workers): """ To monitor whether process is exit. Args: workers (list of multiprocessing.Process): multiprocessing.Process. Returns: int, the timeout(in seconds) when process exit. """ for w in workers: try: exit_code = w.exitcode if exit_code is not None: # For kill -9, we can exit quickly if exit_code == -9: return 1 # For kill -15, we still exit after 30s if exit_code == -15: return 30 # In some cases the subprocess has been killed but the exitcode is still None. # So we use os.kill(pid, 0) to check if it is alive. subprocess_alive = _PythonMultiprocessing.is_process_alive(w.pid) if not subprocess_alive: # Like kill -15, we wait 30s before exit return 30 except ValueError: # process has been closed already return 0 return 0 @staticmethod def is_process_alive(pid): """ Check if the process is alive or not. Note: We hit a deadlock when we use psutil or w.exitcode to check whether a process is alive. Instead we use os.kill(ppid, 0). Args: pid: pid of the process to be checked Returns: True if the process is alive """ try: os.kill(pid, 0) except OSError: return False return True # When main process exit, subprocesses will be terminate @staticmethod def _clean_process(ppid, workers, quit_signal): """ This is the execute function of clean process, if we found main process exited, we will clean subprocesses. Args: ppid: The process id of main process. workers: The list of subprocesses. quit_signal: The flag of quit. """ signal.signal(signal.SIGINT, signal.SIG_IGN) while _PythonMultiprocessing.is_process_alive(ppid): if quit_signal.is_set(): return time.sleep(0.1) _PythonMultiprocessing._terminate_processes(workers) del workers os.kill(os.getpid(), signal.SIGTERM) def launch(self, op_id=-1): """ Launch Python multiprocessing pool. Args: pop_id: ID for operation to have Python multiprocessing pool launched Returns: Python multiprocssing pool is launched. """ self.python_threads_to_workers = {} self.op_id = op_id logger.info("Launching new Python Multiprocessing pool for Op:" + str(self.op_id)) if self.is_mp_enabled(): message = "Launching a new Python multiprocessing pool while a pool already exists!" + \ " The existing pool will be terminated first." logger.warning(message) self.terminate() self.reset() self.create_pool() def create_pool(self): """ Returns: """ if get_enable_shared_mem(): _check_shm_usage(self.num_parallel_workers, 1, self.max_rowsize[0], self.max_rowsize[1]) if self.workers is not None: raise Exception("Pool was already created, close it first.") # Let gc collect unreferenced memory to avoid child processes in the pool to do it gc.collect() # Construct python worker processes self.workers = [] self.warning_ctl = multiprocessing.Value('i', 0) for worker_id in range(self.num_parallel_workers): worker = _MPWorker(self.operations, self.warning_ctl, self.max_rowsize, worker_id) worker.start() self.workers.append(worker) logger.info("Op: " + str(self.op_id) + " Python multiprocessing pool workers' PIDs: " + str(self.get_pids())) self.hook = _PythonMultiprocessing._ExceptHookHandler() # The op (Map, Batch, etc) multiprocessing will launch a watch dog thread for monitoring sub processes self._launch_watch_dog() atexit.register(self.terminate) def terminate(self): # close watch dog first and then close all the workers self.abort_watchdog() self.close_all_workers() if hasattr(self, "warning_ctl"): del self.warning_ctl def get_pids(self): """ Get list of worker's PIDs Returns: list of strings """ if not self.is_mp_enabled(): return [] if not self.pids: self.pids = [] if self.workers: for w in self.workers: try: self.pids.append(w.pid) except ValueError: continue return self.pids def add_new_workers(self, num_new_workers): logger.info( "Increasing num_parallel_workers of Python Multiprocessing pool for Op:" + str(self.op_id) + ", old num_workers=" + str(self.num_parallel_workers) + " new num_workers=" + str( self.num_parallel_workers + num_new_workers) + ".") self.terminate() self.num_parallel_workers += num_new_workers self.launch(self.op_id) def remove_workers(self, num_removed_workers): logger.info( "Decreasing num_parallel_workers of Python Multiprocessing pool for Op:" + str(self.op_id) + ", old num_workers=" + str(self.num_parallel_workers) + " new num_workers=" + str( self.num_parallel_workers - num_removed_workers) + ".") self.terminate() self.num_parallel_workers -= num_removed_workers self.launch(self.op_id) def is_mp_enabled(self): return self.workers is not None def execute(self, idx, *args): """ Execute """ t_id = threading.get_ident() # get the worker_id from Python layer cache first, get from Cpp layer if not found. worker_id = self.python_threads_to_workers.setdefault(t_id, self.get_thread_to_worker()) if worker_id >= len(self.workers): raise RuntimeError("[Internal] worker_id value is greater than number of available workers!") # todo check_iterator_cleanup if self.is_running() and check_iterator_cleanup() is False: return self.workers[worker_id].execute(idx, *args) return None def _launch_watch_dog(self): """ We will launch a watchdog thread and a clean process to cleaning subprocess when there is process was killed. The watchdog thread will cleanup subprocesses and main process when one of the subprocesses was killed. The cleaning subprocess will cleanup subprocesses when main process was killed. """ if platform.system().lower() != 'windows': self.eof = multiprocessing.Event() self.cleaning_process = multiprocessing.Process(target=self._clean_process, name="MapCleanProcess", args=(self.ppid, self.workers, self.eof), daemon=True) self.cleaning_process.start() if get_enable_watchdog(): self.eot = threading.Event() self.watch_dog = threading.Thread(target=self._watch_dog, name="MapWatchDog", args=(self.eot, self.workers + [self.cleaning_process]), daemon=True) self.watch_dog.start() def _abort_watchdog(self): if not self.eot.is_set(): self.eot.set() def abort_watchdog(self): if hasattr(self, 'watch_dog') and self.watch_dog is not None and hasattr(self, 'eot') and self.eot is not None: self._abort_watchdog() if hasattr(self, 'cleaning_process') and self.cleaning_process is not None: if hasattr(self, 'eof') and self.eof is not None and not self.eof.is_set(): self.eof.set() _PythonMultiprocessing._terminate_processes([self.cleaning_process]) del self.cleaning_process def is_running(self): if hasattr(self, 'workers') and self.workers is not None: return all([w.is_alive() for w in self.workers]) return False def close_all_workers(self): """Close all the subprocess workers""" if hasattr(self, 'workers') and self.workers is not None: for w in self.workers: w.close() check_interval = get_multiprocessing_timeout_interval() for w in self.workers: try: subprocess_file_descriptor = w.sentinel st = time.time() while _PythonMultiprocessing.is_process_alive(w.pid): time.sleep(0.01) # sleep 10ms, waiting for the subprocess exit if time.time() - st > check_interval: logger.warning("Waiting for the subprocess worker [{}] to exit.".format(w.pid)) st += check_interval except ValueError as e: if "process object is closed" in str(e): continue raise e try: if w.is_alive(): os.close(subprocess_file_descriptor) except OSError as e: # Maybe the file descriptor had been released, so ignore the 'Bad file descriptor' if "Bad file descriptor" not in str(e): raise e # use clear to release the handle which is better than self.workers = None self.workers.clear() self.workers = None self.pids = None class MapDataset(UnionBaseDataset): """ The result of applying the Map operation to the input Dataset. Args: input_dataset (Dataset): Input Dataset to be mapped. operations (Union[list[TensorOperation], list[functions]]): A function mapping a nested structure of tensors to another nested structure of tensor. Default: ``None``. input_columns (Union[str, list[str]]): List of names of the input columns. Default: ``None``, the operations will be applied on the first columns in the dataset. The size of the list should match the number of inputs of the first operation. output_columns (Union[str, list[str]], optional): List of names of the output columns. The size of the list should match the number of outputs of the last operation. Default: ``None``, output columns will be the input columns, i.e., the columns will be replaced. num_parallel_workers (int, optional): Number of workers to process the dataset in parallel. Default: ``None``. python_multiprocessing (bool, optional): Parallelize Python operations with multiple worker process. This option could be beneficial if the Python operation is computational heavy. Default: ``False``. cache (DatasetCache, optional): Use tensor caching service to speed up dataset processing. Default: ``None``, which means no cache is used. callbacks (DSCallback, list[DSCallback], optional): List of Dataset callbacks to be called. Default: ``None``. max_rowsize(Union[int, list[int]], optional): Maximum size of row in MB that is used for shared memory allocation to copy data between processes, the total occupied shared memory will increase as ``num_parallel_workers`` and :func:`mindspore.dataset.config.set_prefetch_size` increase. If set to -1, shared memory will be dynamically allocated with the actual size of data. This is only used if ``python_multiprocessing`` is set to True. If it is an int value, it represents ``input_columns`` and ``output_columns`` use this value as the unit to create shared memory. If it is a list, the first element represents the ``input_columns`` use this value as the unit to create shared memory, and the second element represents ``output_columns`` use this value as the unit to create shared memory. Default: 16. offload (bool, optional): Flag to indicate whether offload is used. Default: ``None``. """ def __init__(self, input_dataset, operations=None, input_columns=None, output_columns=None, num_parallel_workers=None, python_multiprocessing=False, cache=None, callbacks=None, max_rowsize=16, offload=None): super().__init__(children=input_dataset, num_parallel_workers=num_parallel_workers, cache=cache) self.operations = to_list(operations) for op in self.operations: # user define c_vision.HWC2CHW without parentheses is error if type(op) == type: # pylint: disable=unidiomatic-typecheck raise ValueError("Parameter operations's element of method map should be a dataset processing " "operation instance, but got: {}. It may be missing parentheses for " "instantiation.".format(op)) if not isinstance(op, (c_transforms.TensorOperation, py_transforms.PyTensorOperation)) \ and not callable(op): raise ValueError("Parameter operations's element of method map should be a python function or " "class method which should be callable, but got: {}. It doesn't need parentheses " "for python function or class method.".format(op)) self.input_columns = to_list(input_columns) self.output_columns = to_list(output_columns) # If output_columns were not provided then use input_columns self.output_columns = self.input_columns if not self.output_columns else self.output_columns self.python_multiprocessing = python_multiprocessing self.process_pool = None self.callbacks = to_list(callbacks) if isinstance(max_rowsize, int): self.max_rowsize = [max_rowsize] * 2 else: self.max_rowsize = max_rowsize self.offload = offload def parse(self, children=None): operations = self.__decompose_callable_operations() count_old_transforms, count_new_transforms, count_non_data_vision_transforms = \ self.__count_transforms(operations) count_pyfunc = self.__count_pyfuncs(operations) if count_new_transforms + count_pyfunc == len(operations): prev_op = None for op in operations: # skip user added DebugHook to avoid changing to Py-implementation. if self.__is_debug_hook_op(op): if prev_op: # manually set previous_op_name prev_op_name = self.__parse_op_name(prev_op) op.set_previous_op_name(prev_op_name) continue if op.implementation is None: if prev_op and prev_op.implementation == Implementation.PY: op.implementation = Implementation.PY else: op.implementation = Implementation.C prev_op = op operations = self.__insert_debug_wrapper(operations) operations = transforms.transforms.Compose.reduce(operations) elif count_old_transforms + count_pyfunc + count_non_data_vision_transforms == len(operations): operations = self.__insert_debug_wrapper(operations) operations = transforms.py_transforms.Compose.reduce(operations) else: raise RuntimeError("Mixing old legacy c/py_transforms and new unified transforms is not allowed.") self.operations = self.__process_final_operations(operations) self.prepare_multiprocessing() callbacks = [cb.create_runtime_obj() for cb in self.callbacks] return cde.MapNode(children[0], self.operations, self.input_columns, self.output_columns, callbacks, OffloadToManualOffloadMode.get(self.offload), self.process_pool) def __deepcopy__(self, memodict): return self.__safe_deepcopy__(memodict, exclude=("operations", "callbacks", "__transfer_dataset__")) def __del__(self): if hasattr(self, "process_pool") and self.process_pool is not None: self.process_pool.terminate() del self.process_pool @staticmethod def __parse_op_name(op): """ Utility method to get operation name. """ op_name = "" if isinstance(op, transforms.py_transforms_util.FuncWrapper): try: op_name = op.transform.__name__ except (AttributeError,): op_name = op.transform.__class__.__name__ else: op_name = op.__class__.__name__ return op_name @staticmethod def __construct_debug_hook(previous_op_name=None, is_first_op=False): """ Wrap debug hook into FuncWrapper. """ inserted_functions = [] debug_hook_list = _get_debug_hook_list() if debug_hook_list: for fn in debug_hook_list: # making deep copy to allow each debug hook instance hold unique variables new_fn = copy.deepcopy(fn) new_fn.set_previous_op_name(previous_op_name) new_fn.set_is_first(is_first_op) inserted_func = transforms.py_transforms_util.FuncWrapper(new_fn) inserted_func.implementation = Implementation.PY inserted_functions.append(inserted_func) return inserted_functions @staticmethod def __is_debug_hook_op(op): """ Check if the op is user added DebugHook and skip it to avoid changing transforms implementation. """ if isinstance(op, DebugHook): if not get_debug_mode(): raise ValueError("It is not allowed to inject DebugHook object in non-debug mode.") return True return False @staticmethod def __count_pyfuncs(operations): """ Count the number of pyfuncs operations """ return sum([1 if isinstance(op, FuncWrapper) else 0 for op in operations]) @staticmethod def __count_transforms(operations): """ Count the various flavors of transforms operations """ # Count the number of old legacy data and vision c_transforms and py_transforms count_old_transforms = sum( [1 if "c_transforms" in str(op) or isinstance(op, (c_transforms.TensorOperation, py_transforms.PyTensorOperation)) or ("py_transforms" in str(op) and not isinstance(op, FuncWrapper)) else 0 for op in operations]) # Count the number of new unified data and vision transforms count_new_transforms = sum([1 if hasattr(op, "implementation") and not isinstance(op, FuncWrapper) else 0 for op in operations]) # Count the number of non-data transforms and non-vision transforms count_non_data_vision_transforms = sum( [1 if "text.transforms" in str(op) or "audio.transforms" in str(op) else 0 for op in operations]) return count_old_transforms, count_new_transforms, count_non_data_vision_transforms @staticmethod def __operation_valid_for_multiprocessing(op): if callable(op) and str(op).find("c_transform") < 0: return True return False @staticmethod def __process_final_operations(operations): """ Build final list of operations """ operations_fin = [] for op in operations: if hasattr(op, "implementation"): if op.implementation == Implementation.C and not isinstance(op, (FuncWrapper, ToNumpy)): operations_fin.append(op.parse()) elif op.implementation == Implementation.PY: operations_fin.append(op) elif isinstance(op, (FuncWrapper, ToNumpy)): operations_fin.append(op) else: raise RuntimeError("Wrong implementation") else: if op and getattr(op, 'parse', None): operations_fin.append(op.parse()) else: operations_fin.append(op) return operations_fin # Iterator bootstrap will be called on iterator construction. # A deep copy of Dataset object is created prior of iterator_bootstrap. # This method will create per iterator process pool and bind pyfunc execution to the pool. def prepare_multiprocessing(self): """ Per iterator bootstrap callback. """ if self.python_multiprocessing and platform.system().lower() == 'windows': logger.warning("Python multiprocessing is not supported on Windows platform.") return if self.python_multiprocessing and get_debug_mode(): logger.warning("Python multiprocessing is not supported in debug mode." " Ignoring Python multiprocessing for map operation.") return if self.python_multiprocessing: iter_specific_operations = [] callable_list = [] # If user didn't specify num_parallel_workers, set it to default if self.num_parallel_workers is None: self.num_parallel_workers = get_num_parallel_workers() # Pass #1, look for Python callables and build list for op in self.operations: # our c transforms is now callable and should not be run in Python multithreading if MapDataset.__operation_valid_for_multiprocessing(op): callable_list.append(op) if callable_list: self.process_pool = _PythonMultiprocessing(str(self), self.num_parallel_workers, callable_list, self.max_rowsize) # Pass #2 idx = 0 for op in self.operations: # our c transforms is now callable and should not be run in Python multithreading if MapDataset.__operation_valid_for_multiprocessing(op): # Wrap Python callable into _PythonCallable iter_specific_operations.append(_PythonCallable(op, idx, self.process_pool)) idx += 1 else: # CPP ops remain the same iter_specific_operations.append(op) self.operations = iter_specific_operations def __insert_debug_wrapper(self, operations): """ Insert DebuggerWrapper before and after each op if debug mode is on. """ if not get_debug_mode(): return operations first_op_name = self.__parse_op_name(operations[0]) inserted_operations = self.__construct_debug_hook(first_op_name, is_first_op=True) for op in operations: inserted_operations.append(op) op_name = self.__parse_op_name(op) inserted_operations.extend(self.__construct_debug_hook(op_name)) return inserted_operations def __decompose_callable_operations(self): """ Decompose operations and build list of old legacy ops which are callable """ decomposed_operations = transforms.transforms.Compose.decompose(self.operations) operations = [] for op in decomposed_operations: if callable(op) and not hasattr(op, "implementation") and str(op).find( "c_transform") < 0 and not isinstance(op, c_transforms.TensorOperation) and \ not isinstance(op, py_transforms.PyTensorOperation): op = transforms.py_transforms_util.FuncWrapper(op) operations.append(op) return operations class FilterDataset(UnionBaseDataset): """ The result of applying filter predicate to the input Dataset. Args: input_dataset (Dataset): Input Dataset to be mapped. predicate (callable): Python callable which returns a boolean value. If False then filter the element. input_columns (Union[str, list[str]], optional): List of names of the input columns. Default: ``None``, the predicate will be applied to all columns in the dataset. num_parallel_workers (int, optional): Number of workers to process the dataset in parallel. Default: ``None``. """ def __init__(self, input_dataset, predicate, input_columns=None, num_parallel_workers=None): super().__init__(children=input_dataset, num_parallel_workers=num_parallel_workers) self.predicate = lambda *args: bool(predicate(*args)) self.input_columns = to_list(input_columns) def parse(self, children=None): return cde.FilterNode(children[0], self.predicate, self.input_columns) class RepeatDataset(UnionBaseDataset): """ The result of applying Repeat operation to the input Dataset. Args: input_dataset (Dataset): Input Dataset to be repeated. count (int): Number of times the dataset will be repeated. Default: -1, repeat indefinitely. """ def __init__(self, input_dataset, count): super().__init__(children=input_dataset) self.count = replace_none(count, -1) def parse(self, children=None): return cde.RepeatNode(children[0], self.count) class SkipDataset(UnionBaseDataset): """ The result of applying Skip operation to the input Dataset. Args: input_dataset (Dataset): Input dataset to have elements skipped. count (int): Number of elements to be skipped in the dataset. """ def __init__(self, input_dataset, count): super().__init__(input_dataset) self.count = count def parse(self, children=None): return cde.SkipNode(children[0], self.count) class TakeDataset(UnionBaseDataset): """ The result of applying Take operation to the input Dataset. Args: input_dataset (Dataset): Input Dataset to have elements taken from. count (int): Number of elements to be taken from the dataset. """ def __init__(self, input_dataset, count): super().__init__(children=input_dataset) self.count = count def parse(self, children=None): return cde.TakeNode(children[0], self.count) class ZipDataset(UnionBaseDataset): """ The result of applying Zip operation to the input Dataset. Args: datasets (tuple): A tuple of datasets to be zipped together. Raises: TypeError: If dataset is not an instance of Dataset. """ def __init__(self, datasets): super().__init__(children=datasets) def parse(self, children=None): return cde.ZipNode(children) def is_sync(self): return any([c.is_sync() for c in self.children]) class ConcatDataset(UnionBaseDataset): """ The result of applying Concat operation to the input Dataset. Args: datasets (list): A list of datasets to be concatenated together. Raises: TypeError: If dataset is not an instance of Dataset. ValueError: If there is no samples in the one of the datasets. """ def __init__(self, datasets): super().__init__(children=datasets) for dataset in datasets: if not isinstance(dataset, Dataset): raise TypeError("Invalid dataset, expected Dataset object, but got %s!" % type(dataset)) self.datasets = datasets self._sampler = samplers.SequentialSampler(num_samples=None) self.children_sizes_ = [c.get_dataset_size() for c in self.children] child_index = 0 for item in self.children_sizes_: if item == 0: raise ValueError("There are no samples in the dataset number %d. Please make sure there are " "valid samples in the dataset." % child_index) child_index += 1 self._children_sizes = self.children_sizes_.copy() # _children_flag_and_nums: A list of pair<int ,int>.The first element of pair is flag that characterizes # whether the dataset is mappable. The second element of pair is length of the dataset self._children_flag_and_nums = [] # _children_start_end_index_: A list of pair<int ,int>.The elements of pair are used to characterize # the valid position of the dataset corresponding to the subscript when sampling self._children_start_end_index_ = [] for index, child in enumerate(self.children): tem_list = [-1, -1] self._children_start_end_index_.append(tem_list) dataset_len = self.children_sizes_[index] from mindspore.dataset.engine.datasets_user_defined import GeneratorDataset if isinstance(child, GeneratorDataset) and not hasattr(child.source, "__getitem__"): dataset_len = 0 self.children_sizes_[index] = 0 if isinstance(child, MappableDataset): self._children_flag_and_nums.append((0, dataset_len)) else: self._children_flag_and_nums.append((1, dataset_len)) def parse(self, children=None): return cde.ConcatNode(children, self._sampler, self._children_flag_and_nums, self._children_start_end_index_, self._children_sizes) def use_sampler(self, sampler): """ Set the distributedSampler to concat dataset Args: sampler (Sampler): The sampler to use for the current dataset. Currently supported: DistributedSampler. Raises: TypeError: If the sampler is not an instance of DistributedSampler ValueError: If the parameter shuffle of sampler is True ValueError: If the parameter NumSamples of sampler is not None. ValueError: If num_shards <=0. """ if not isinstance(sampler, (samplers.DistributedSampler, samplers.RandomSampler)): raise TypeError("The parameter %s of concat must be DistributedSampler or RandomSampler!" % sampler) if isinstance(sampler, samplers.RandomSampler): if sampler.replacement: raise ValueError("The parameter replacement of RandomSampler must be False!") if sampler.get_num_samples() is not None: raise ValueError("The parameter num_samples of RandomSampler is not support to be set!") self._sampler = sampler self._children_sizes = [c.get_dataset_size() for c in self.children] # Recursive access to other child concat nodes def set_child(node): for c in node.children: if isinstance(c, ConcatDataset): c.use_sampler(sampler) set_child(c) set_child(self) return if sampler.is_shuffled(): raise ValueError("The parameter shuffle of DistributedSampler must be False!") if sampler.num_shards <= 0: raise ValueError("The parameter num_shards of DistributedSampler must be positive int!") if sampler.get_num_samples() is not None: raise ValueError("The parameter num_samples of DistributedSampler is not support to be set!") self.dataset_size = None self._sampler = sampler cumulative_samples_nums = 0 for index, child in enumerate(self.children): if hasattr(child, 'sampler') and child.sampler.get_num_samples() is not None: raise ValueError("The parameter NumSamples of %s is not support to be set!" % child) if isinstance(child, (BatchDataset, PaddedBatchDataset)): raise TypeError("The parameter %s of concat must not be BatchDataset or PaddedBatchDataset!" % child) # if child is mappable and the length is greater than 0 if not self._children_flag_and_nums[index][0] and self._children_flag_and_nums[index][1]: tem_value = cumulative_samples_nums + self._children_flag_and_nums[index][1] if not self._children_flag_and_nums[index][1] >= sampler.num_shards: if tem_value < sampler.num_shards: self._children_start_end_index_[index][0] = cumulative_samples_nums self._children_start_end_index_[index][1] = tem_value else: self._children_start_end_index_[index][0] = cumulative_samples_nums self._children_start_end_index_[index][1] = tem_value % sampler.num_shards tem_sampler = copy.deepcopy(sampler) tem_sampler.set_offset(cumulative_samples_nums) child.use_sampler(tem_sampler) cumulative_samples_nums += self.children_sizes_[index] cumulative_samples_nums %= sampler.num_shards class RenameDataset(UnionBaseDataset): """ The result of applying Rename operation to the input Dataset. Args: input_dataset (Dataset): Input Dataset to be Renamed. input_columns (Union[str, list[str]]): List of names of the input columns. output_columns (Union[str, list[str]]): List of names of the output columns. """ def __init__(self, input_dataset, input_columns, output_columns): super().__init__(children=input_dataset) self.input_column_names = to_list(input_columns) self.output_column_names = to_list(output_columns) def parse(self, children=None): return cde.RenameNode(children[0], self.input_column_names, self.output_column_names) def to_list(items): if items is None: return [] if isinstance(items, tuple): return list(items) if not isinstance(items, list): return [items] return items class ProjectDataset(UnionBaseDataset): """ The result of applying Project operation to the input Dataset. Args: input_dataset (Dataset): Input Dataset to be Projected. columns (Union[str, list[str]]): List of names of the columns to project. """ def __init__(self, input_dataset, columns): super().__init__(children=input_dataset) self.columns = to_list(columns) def parse(self, children=None): return cde.ProjectNode(children[0], self.columns) class _ToDevice: """ Internal class to handle sending data to device. """ def __init__(self, dataset, num_epochs): if get_debug_mode(): logger.error("MindData debugger cannot be used in dataset sink mode. Please manually turn off " "sink mode and try debugger again.") ir_tree, self.api_tree = dataset.create_ir_tree() self._runtime_context = cde.PythonRuntimeContext() self._runtime_context.Init() self._to_device = cde.ToDevice(num_epochs) if dataset.get_init_step() != 0: init_step = dataset.get_init_step() dataset_size = dataset.get_dataset_size() self._to_device.Init(ir_tree, init_step, dataset_size) else: self._to_device.Init(ir_tree, 0, -1) self._runtime_context.AssignConsumer(self._to_device) ITERATORS_LIST.append(weakref.ref(self)) _unset_iterator_cleanup() def send(self): self._to_device.Send() def stop_send(self): """ send stop send signal to pipeline, it is used when end of sequence is sent at the epoch end. """ self._to_device.StopSend() def continue_send(self): """ send continue send signal to pipeline, it is used when end of sequence is sent at the epoch end. """ self._to_device.ContinueSend() def get_data_info(self): """ Get type and shape of current batch. """ return self._to_device.GetDataInfo() def get_mbuf_queue_size(self): """ Get element numbers inside mbuf. """ return self._to_device.GetMbufQueueSize() def get_send_info(self): """ In sink mode, it returns the send information of dataset at this moment. Send information includes number of send batches, time summary of fetching data on host and time summary of sending data. """ return self._to_device.GetSendInfo() def release(self): """ Manually terminate Device Queue instead of relying on out of scope destruction. """ if hasattr(self, '_runtime_context') and self._runtime_context: if hasattr(self, '_to_device') and self._to_device: self._runtime_context.Terminate() del self._to_device del self._runtime_context def __deepcopy__(self, memodict): return self def get_offload_model(self, col_names): """ Get offload model containing removed offload ops from pipeline. """ offload_model = GetOffloadModel(self._to_device, col_names) return offload_model def _reset(self, step, dataset_size): self._to_device.Reset(step, dataset_size) class TransferDataset(Dataset): """ The result of applying TDT operation to the input Dataset. Args: input_dataset (Dataset): Input Dataset to be transferred. send_epoch_end (bool, optional): Whether to send end of sequence to device or not. Default: ``True``. create_data_info_queue (bool, optional): Whether to create queue which stores types and shapes of data or not. Default: ``False``. Raises: TypeError: If device_type is empty. ValueError: If device_type is not 'Ascend', 'GPU' or 'CPU'. RuntimeError: If dataset is unknown. """ def __init__(self, input_dataset, send_epoch_end=True, create_data_info_queue=False, queue_name=""): super().__init__(children=input_dataset) if queue_name == "": self.queue_name = str(uuid.uuid1()) logger.info(f"queue_name is newly generated. value is {self.queue_name}") else: self.queue_name = queue_name logger.info(f"queue_name is read from compile cache. value is {self.queue_name}") self.device_type = context.get_context("device_target") if context else "CPU" self.device_id = context.get_context("device_id") if context else 0 self._send_epoch_end = replace_none(send_epoch_end, True) self._create_data_info_queue = create_data_info_queue self._to_device = None self.column_name = input_dataset.get_col_names() def parse(self, children=None): total_batch = 0 if hasattr(self.children[0], "__total_batch__"): total_batch = self.children[0].__total_batch__ check_total_batch(total_batch) return cde.DataQueueNode(children[0], self.queue_name, self.device_type, self.device_id, self._send_epoch_end, total_batch, self._create_data_info_queue) def create_dict_iterator(self, num_epochs=-1, output_numpy=False): raise RuntimeError("TransferDataset is not iterable.") def create_tuple_iterator(self, columns=None, num_epochs=-1, output_numpy=False, do_copy=True): raise RuntimeError("TransferDataset is not iterable.") def __iter__(self): raise RuntimeError("TransferDataset is not iterable.") def output_shapes(self): raise RuntimeError("TransferDataset does not support obtaining output_shapes.") def output_types(self): raise RuntimeError("TransferDataset does not support obtaining output_types.") @check_to_device_send def send(self, num_epochs=-1): """ Send to device """ if Dataset._noop_mode(): return if self._to_device is not None: del self._to_device self._to_device = _ToDevice(self, num_epochs) self._to_device.send() def stop_send(self): if self._to_device is not None: self._to_device.stop_send() def continue_send(self): if self._to_device is not None: self._to_device.continue_send() def get_data_info(self): """ Get type and shape of current batch """ if self._to_device is not None: return self._to_device.get_data_info() raise RuntimeError("Calling get_data_info with bad state.") def get_mbuf_queue_size(self): """ Get element numbers inside mbuf. """ if self._to_device is not None: return self._to_device.get_mbuf_queue_size() raise RuntimeError("Device queue is not init, call get_mbuf_queue_size failed.") def get_send_info(self): """ In sink mode, it returns the send information of dataset at this moment. Send information includes number of send batches, time summary of fetching data on host and time summary of sending data. """ if self._to_device is not None: return self._to_device.get_send_info() raise RuntimeError("Calling get_send_info with bad state, data queue is not initialized.") def get_offload_model(self): if self._to_device is not None: return self._to_device.get_offload_model(self.column_name) raise RuntimeError("get_offload_model, _to_device is None") def release(self): """ Manually terminate Device Queue instead of relying on out of scope destruction. """ if self._to_device is not None: self._to_device.release() def _reset(self, step, dataset_size): if self._to_device is not None: logger.info("Reset the dataset pipeline to step: " + str(step) + ", epoch: " + str(step // dataset_size)) self._to_device._reset(step, dataset_size) # pylint: disable=protected-access
[文档]class Schema: """ Class to represent a schema of a dataset. Args: schema_file (str): Path of the schema file. Default: ``None``. Raises: RuntimeError: If schema file failed to load. Examples: >>> import mindspore.dataset as ds >>> from mindspore import dtype as mstype >>> >>> # Create schema; specify column name, mindspore.dtype and shape of the column >>> schema = ds.Schema() >>> schema.add_column(name='col1', de_type=mstype.int64, shape=[2]) """ @check_schema def __init__(self, schema_file=None): self.schema_file = replace_none(schema_file, "") self.cpp_schema = cde.SchemaObj(self.schema_file)
[文档] @check_add_column def add_column(self, name, de_type, shape=None): """ Add new column to the schema. Args: name (str): The new name of the column. de_type (str): Data type of the column. shape (list[int], optional): Shape of the column. Default: ``None``, [-1] which is an unknown shape of rank 1. Raises: ValueError: If column type is unknown. Examples: >>> import mindspore.dataset as ds >>> from mindspore import dtype as mstype >>> >>> schema = ds.Schema() >>> schema.add_column('col_1d', de_type=mstype.int64, shape=[2]) """ if isinstance(de_type, typing.Type): de_type = mstype_to_detype(de_type) col_type = str(de_type) else: col_type = str(cde.DataType(de_type)) if shape is None: self.cpp_schema.add_column(name, col_type) else: self.cpp_schema.add_column(name, col_type, shape)
[文档] def parse_columns(self, columns): """ Parse the columns and add it to self. Args: columns (Union[dict, list[dict], tuple[dict]]): Dataset attribute information, decoded from schema file. - list[dict], `name` and `type` must be in keys, `shape` optional. - dict, columns.keys() as name, columns.values() is dict, and `type` inside, `shape` optional. Raises: RuntimeError: If failed to parse columns. RuntimeError: If column's name field is missing. RuntimeError: If column's type field is missing. Examples: >>> from mindspore.dataset import Schema >>> schema = Schema() >>> columns1 = [{'name': 'image', 'type': 'int8', 'shape': [3, 3]}, ... {'name': 'label', 'type': 'int8', 'shape': [1]}] >>> schema.parse_columns(columns1) >>> columns2 = {'image': {'shape': [3, 3], 'type': 'int8'}, 'label': {'shape': [1], 'type': 'int8'}} >>> schema.parse_columns(columns2) """ self.cpp_schema.parse_columns(json.dumps(columns, indent=2))
[文档] def to_json(self): """ Get a JSON string of the schema. Returns: str, JSON string of the schema. Examples: >>> from mindspore.dataset import Schema >>> from mindspore import dtype as mstype >>> >>> schema = Schema() >>> schema.add_column('col_1d', de_type=mstype.int64, shape=[2]) >>> json = schema.to_json() """ return self.cpp_schema.to_json()
[文档] def from_json(self, json_obj): """ Get schema file from JSON object. Args: json_obj(dictionary): Object of JSON parsed. Raises: RuntimeError: if there is unknown item in the object. RuntimeError: if dataset type is missing in the object. RuntimeError: if columns are missing in the object. Examples: >>> import json >>> from mindspore.dataset import Schema >>> >>> with open("/path/to/schema_file", "r") as file: ... json_obj = json.load(file) ... schema = Schema() ... schema.from_json(json_obj) """ self.cpp_schema.from_string(json.dumps(json_obj, indent=2))
def __str__(self): return self.to_json() @staticmethod def get_num_rows(schema): schema_obj = schema if not isinstance(schema_obj, Schema): schema_obj = Schema(schema_obj) return schema_obj.cpp_schema.get_num_rows()
class DeserializedDataset(Dataset): def __init__(self, input_obj): super().__init__() self.input_obj = input_obj def parse(self, children=None): if isinstance(self.input_obj, dict): json_str = json.dumps(self.input_obj) return cde.Dataset.from_json_string(json_str) return cde.Dataset.from_json_file(self.input_obj)