opennmt/data/dataset.py
def training_pipeline(
    batch_size,
    batch_type="examples",
    batch_multiplier=1,
    batch_size_multiple=1,
    process_fn=None,
    transform_fns=None,
    length_bucket_width=None,
    features_length_fn=None,
    labels_length_fn=None,
    maximum_features_length=None,
    maximum_labels_length=None,
    single_pass=False,
    num_shards=1,
    shard_index=0,
    num_threads=None,
    dataset_size=None,
    shuffle_buffer_size=None,
    prefetch_buffer_size=None,
    cardinality_multiple=1,
):
    """Transformation that applies most of the dataset operations commonly used
    for training on sequence data:

    * sharding
    * shuffling
    * processing
    * filtering
    * bucketization
    * batching
    * prefetching

    Example:

      >>> dataset = dataset.apply(opennmt.data.training_pipeline(...))

    Args:
      batch_size: The batch size to use.
      batch_type: The training batching strategy to use: can be "examples" or
        "tokens".
      batch_multiplier: The batch size multiplier.
      batch_size_multiple: When :obj:`batch_type` is "tokens", ensure that the
        resulting batch size is a multiple of this value.
      process_fn: The processing function to apply on each element.
      transform_fns: List of dataset transformation functions (applied after
        :obj:`process_fn` if defined).
      length_bucket_width: The width of the length buckets to select batch
        candidates from. ``None`` to not constrain batch formation.
      features_length_fn: A function mapping features to a sequence length.
      labels_length_fn: A function mapping labels to a sequence length.
      maximum_features_length: The maximum length or list of maximum lengths of
        the features sequence(s). ``None`` to not constrain the length.
      maximum_labels_length: The maximum length of the labels sequence.
        ``None`` to not constrain the length.
      single_pass: If ``True``, makes a single pass over the training data.
      num_shards: The number of data shards (usually the number of workers in a
        distributed setting).
      shard_index: The shard index this data pipeline should read from.
      num_threads: The number of elements processed in parallel.
      dataset_size: If the dataset size is already known, it can be passed here to
        avoid a slower generic computation of the dataset size later.
      shuffle_buffer_size: The number of elements from which to sample.
      prefetch_buffer_size: The number of batches to prefetch asynchronously. If
        ``None``, use an automatically tuned value.
      cardinality_multiple: Ensure that the dataset cardinality is a multiple of
        this value when :obj:`single_pass` is ``True``.

    Returns:
      A ``tf.data.Dataset`` transformation.

    See Also:
      - :func:`opennmt.data.batch_sequence_dataset`
      - :func:`opennmt.data.make_cardinality_multiple_of`
      - :func:`opennmt.data.filter_examples_by_length`
      - :func:`opennmt.data.filter_irregular_batches`
      - :func:`opennmt.data.shuffle_dataset`
    """
    if dataset_size is not None and num_shards > 1:
        # Update dataset_size based on the shard size.
        if isinstance(dataset_size, list):
            dataset_size = [size // num_shards for size in dataset_size]
        else:
            dataset_size //= num_shards

    def _make_weighted_dataset(datasets, weights):
        if single_pass:
            raise ValueError(
                "single_pass parameter is not compatible with weighted datasets"
            )
        if not datasets:
            raise ValueError("At least one dataset is required")
        if weights is not None and len(weights) != len(datasets):
            raise ValueError(
                "%d dataset weights were provided, but %d were expected to match the "
                "number of data files" % (len(weights), len(datasets))
            )
        if num_shards > 1:
            datasets = [dataset.shard(num_shards, shard_index) for dataset in datasets]
        weights = normalize_weights(datasets, weights=weights, sizes=dataset_size)
        datasets = [dataset.repeat() for dataset in datasets]
        dataset = tf.data.experimental.sample_from_datasets(datasets, weights=weights)
        if shuffle_buffer_size is not None and shuffle_buffer_size != 0:
            if shuffle_buffer_size < 0:
                raise ValueError(
                    "shuffle_buffer_size < 0 is not compatible with weighted datasets"
                )
            dataset = dataset.shuffle(shuffle_buffer_size)
        return dataset

    def _make_single_dataset(dataset):
        if num_shards > 1:
            dataset = dataset.shard(num_shards, shard_index)
        if shuffle_buffer_size is not None and shuffle_buffer_size != 0:
            dataset = dataset.apply(
                shuffle_dataset(shuffle_buffer_size, dataset_size=dataset_size)
            )
        return dataset

    def _pipeline(dataset):
        if isinstance(dataset, tuple):
            dataset, weights = dataset
        else:
            weights = None
        is_weighted_dataset = isinstance(dataset, list)
        if is_weighted_dataset:
            dataset = _make_weighted_dataset(dataset, weights)
        else:
            dataset = _make_single_dataset(dataset)
        if process_fn is not None:
            dataset = dataset.map(process_fn, num_parallel_calls=num_threads or 4)
        if transform_fns is not None:
            for transform_fn in transform_fns:
                dataset = dataset.apply(transform_fn)
        else:
            dataset = dataset.apply(
                filter_examples_by_length(
                    maximum_features_length=maximum_features_length,
                    maximum_labels_length=maximum_labels_length,
                    features_length_fn=features_length_fn,
                    labels_length_fn=labels_length_fn,
                )
            )
        dataset = dataset.apply(
            batch_sequence_dataset(
                batch_size,
                batch_type=batch_type,
                batch_multiplier=batch_multiplier,
                batch_size_multiple=batch_size_multiple,
                length_bucket_width=length_bucket_width,
                length_fn=[features_length_fn, labels_length_fn],
            )
        )
        dataset = dataset.apply(filter_irregular_batches(batch_multiplier))
        if not single_pass:
            if not is_weighted_dataset:  # Weighted dataset is repeated before sampling.
                dataset = dataset.repeat()
        else:
            dataset = dataset.apply(make_cardinality_multiple_of(cardinality_multiple))
        dataset = dataset.prefetch(prefetch_buffer_size)
        return dataset

    return _pipeline
Transformation that applies most of the dataset operations commonly used
    for training on sequence data:

    * sharding
    * shuffling
    * processing
    * filtering
    * bucketization
    * batching
    * prefetching

    Example:

      >>> dataset = dataset.apply(opennmt.data.training_pipeline(...))

    Args:
      batch_size: The batch size to use.
      batch_type: The training batching strategy to use: can be "examples" or
        "tokens".
      batch_multiplier: The batch size multiplier.
      batch_size_multiple: When :obj:`batch_type` is "tokens", ensure that the
        resulting batch size is a multiple of this value.
      process_fn: The processing function to apply on each element.
      transform_fns: List of dataset transformation functions (applied after
        :obj:`process_fn` if defined).
      length_bucket_width: The width of the length buckets to select batch
        candidates from. ``None`` to not constrain batch formation.
      features_length_fn: A function mapping features to a sequence length.
      labels_length_fn: A function mapping labels to a sequence length.
      maximum_features_length: The maximum length or list of maximum lengths of
        the features sequence(s). ``None`` to not constrain the length.
      maximum_labels_length: The maximum length of the labels sequence.
        ``None`` to not constrain the length.
      single_pass: If ``True``, makes a single pass over the training data.
      num_shards: The number of data shards (usually the number of workers in a
        distributed setting).
      shard_index: The shard index this data pipeline should read from.
      num_threads: The number of elements processed in parallel.
      dataset_size: If the dataset size is already known, it can be passed here to
        avoid a slower generic computation of the dataset size later.
      shuffle_buffer_size: The number of elements from which to sample.
      prefetch_buffer_size: The number of batches to prefetch asynchronously. If
        ``None``, use an automatically tuned value.
      cardinality_multiple: Ensure that the dataset cardinality is a multiple of
        this value when :obj:`single_pass` is ``True``.

    Returns:
      A ``tf.data.Dataset`` transformation.

    See Also:
      - :func:`opennmt.data.batch_sequence_dataset`
      - :func:`opennmt.data.make_cardinality_multiple_of`
      - :func:`opennmt.data.filter_examples_by_length`
      - :func:`opennmt.data.filter_irregular_batches`
      - :func:`opennmt.data.shuffle_dataset`
    
Transformation that applies most of the dataset operations commonly used
for training on sequence data.

sharding
shuffling
processing
filtering
bucketization
batching
prefetching
Transformation that applies most of the dataset operations commonly used for training on sequence data . sharding shuffling processing filtering bucketization batching prefetching
def training_pipeline(
    batch_size,
    batch_type="examples",
    batch_multiplier=1,
    batch_size_multiple=1,
    process_fn=None,
    transform_fns=None,
    length_bucket_width=None,
    features_length_fn=None,
    labels_length_fn=None,
    maximum_features_length=None,
    maximum_labels_length=None,
    single_pass=False,
    num_shards=1,
    shard_index=0,
    num_threads=None,
    dataset_size=None,
    shuffle_buffer_size=None,
    prefetch_buffer_size=None,
    cardinality_multiple=1,
):
    if dataset_size is not None and num_shards > 1:
        if isinstance(dataset_size, list):
            dataset_size = [size // num_shards for size in dataset_size]
        else:
            dataset_size //= num_shards
    def _make_weighted_dataset(datasets, weights):
        if single_pass:
            raise ValueError(
                "single_pass parameter is not compatible with weighted datasets"
            )
        if not datasets:
            raise ValueError("At least one dataset is required")
        if weights is not None and len(weights) != len(datasets):
            raise ValueError(
                "%d dataset weights were provided, but %d were expected to match the "
                "number of data files" % (len(weights), len(datasets))
            )
        if num_shards > 1:
            datasets = [dataset.shard(num_shards, shard_index) for dataset in datasets]
        weights = normalize_weights(datasets, weights=weights, sizes=dataset_size)
        datasets = [dataset.repeat() for dataset in datasets]
        dataset = tf.data.experimental.sample_from_datasets(datasets, weights=weights)
        if shuffle_buffer_size is not None and shuffle_buffer_size != 0:
            if shuffle_buffer_size < 0:
                raise ValueError(
                    "shuffle_buffer_size < 0 is not compatible with weighted datasets"
                )
            dataset = dataset.shuffle(shuffle_buffer_size)
        return dataset
    def _make_single_dataset(dataset):
        if num_shards > 1:
            dataset = dataset.shard(num_shards, shard_index)
        if shuffle_buffer_size is not None and shuffle_buffer_size != 0:
            dataset = dataset.apply(
                shuffle_dataset(shuffle_buffer_size, dataset_size=dataset_size)
            )
        return dataset
    def _pipeline(dataset):
        if isinstance(dataset, tuple):
            dataset, weights = dataset
        else:
            weights = None
        is_weighted_dataset = isinstance(dataset, list)
        if is_weighted_dataset:
            dataset = _make_weighted_dataset(dataset, weights)
        else:
            dataset = _make_single_dataset(dataset)
        if process_fn is not None:
            dataset = dataset.map(process_fn, num_parallel_calls=num_threads or 4)
        if transform_fns is not None:
            for transform_fn in transform_fns:
                dataset = dataset.apply(transform_fn)
        else:
            dataset = dataset.apply(
                filter_examples_by_length(
                    maximum_features_length=maximum_features_length,
                    maximum_labels_length=maximum_labels_length,
                    features_length_fn=features_length_fn,
                    labels_length_fn=labels_length_fn,
                )
            )
        dataset = dataset.apply(
            batch_sequence_dataset(
                batch_size,
                batch_type=batch_type,
                batch_multiplier=batch_multiplier,
                batch_size_multiple=batch_size_multiple,
                length_bucket_width=length_bucket_width,
                length_fn=[features_length_fn, labels_length_fn],
            )
        )
        dataset = dataset.apply(filter_irregular_batches(batch_multiplier))
        if not single_pass:
            if not is_weighted_dataset:  
                dataset = dataset.repeat()
        else:
            dataset = dataset.apply(make_cardinality_multiple_of(cardinality_multiple))
        dataset = dataset.prefetch(prefetch_buffer_size)
        return dataset
    return _pipeline
def training_pipeline ( batch_size , batch_type = "examples" , batch_multiplier = 1 , batch_size_multiple = 1 , process_fn = None , transform_fns = None , length_bucket_width = None , features_length_fn = None , labels_length_fn = None , maximum_features_length = None , maximum_labels_length = None , single_pass = False , num_shards = 1 , shard_index = 0 , num_threads = None , dataset_size = None , shuffle_buffer_size = None , prefetch_buffer_size = None , cardinality_multiple = 1 , ) : if dataset_size is not None and num_shards > 1 : if isinstance ( dataset_size , list ) : dataset_size = [ size // num_shards for size in dataset_size ] else : dataset_size //= num_shards def _make_weighted_dataset ( datasets , weights ) : if single_pass : raise ValueError ( "single_pass parameter is not compatible with weighted datasets" ) if not datasets : raise ValueError ( "At least one dataset is required" ) if weights is not None and len ( weights ) != len ( datasets ) : raise ValueError ( "%d dataset weights were provided, but %d were expected to match the " "number of data files" % ( len ( weights ) , len ( datasets ) ) ) if num_shards > 1 : datasets = [ dataset . shard ( num_shards , shard_index ) for dataset in datasets ] weights = normalize_weights ( datasets , weights = weights , sizes = dataset_size ) datasets = [ dataset . repeat ( ) for dataset in datasets ] dataset = tf . data . experimental . sample_from_datasets ( datasets , weights = weights ) if shuffle_buffer_size is not None and shuffle_buffer_size != 0 : if shuffle_buffer_size < 0 : raise ValueError ( "shuffle_buffer_size < 0 is not compatible with weighted datasets" ) dataset = dataset . shuffle ( shuffle_buffer_size ) return dataset def _make_single_dataset ( dataset ) : if num_shards > 1 : dataset = dataset . shard ( num_shards , shard_index ) if shuffle_buffer_size is not None and shuffle_buffer_size != 0 : dataset = dataset . apply ( shuffle_dataset ( shuffle_buffer_size , dataset_size = dataset_size ) ) return dataset def _pipeline ( dataset ) : if isinstance ( dataset , tuple ) : dataset , weights = dataset else : weights = None is_weighted_dataset = isinstance ( dataset , list ) if is_weighted_dataset : dataset = _make_weighted_dataset ( dataset , weights ) else : dataset = _make_single_dataset ( dataset ) if process_fn is not None : dataset = dataset . map ( process_fn , num_parallel_calls = num_threads or 4 ) if transform_fns is not None : for transform_fn in transform_fns : dataset = dataset . apply ( transform_fn ) else : dataset = dataset . apply ( filter_examples_by_length ( maximum_features_length = maximum_features_length , maximum_labels_length = maximum_labels_length , features_length_fn = features_length_fn , labels_length_fn = labels_length_fn , ) ) dataset = dataset . apply ( batch_sequence_dataset ( batch_size , batch_type = batch_type , batch_multiplier = batch_multiplier , batch_size_multiple = batch_size_multiple , length_bucket_width = length_bucket_width , length_fn = [ features_length_fn , labels_length_fn ] , ) ) dataset = dataset . apply ( filter_irregular_batches ( batch_multiplier ) ) if not single_pass : if not is_weighted_dataset : dataset = dataset . repeat ( ) else : dataset = dataset . apply ( make_cardinality_multiple_of ( cardinality_multiple ) ) dataset = dataset . prefetch ( prefetch_buffer_size ) return dataset return _pipeline
Transformation that applies most of the dataset operations commonly used
for training on sequence data:
Transformation that applies most of the dataset operations commonly used for training on sequence data :
"""Transformation that applies most of the dataset operations commonly used
    for training on sequence data:

    * sharding
    * shuffling
    * processing
    * filtering
    * bucketization
    * batching
    * prefetching

    Example:

      >>> dataset = dataset.apply(opennmt.data.training_pipeline(...))

    Args:
      batch_size: The batch size to use.
      batch_type: The training batching strategy to use: can be "examples" or
        "tokens".
      batch_multiplier: The batch size multiplier.
      batch_size_multiple: When :obj:`batch_type` is "tokens", ensure that the
        resulting batch size is a multiple of this value.
      process_fn: The processing function to apply on each element.
      transform_fns: List of dataset transformation functions (applied after
        :obj:`process_fn` if defined).
      length_bucket_width: The width of the length buckets to select batch
        candidates from. ``None`` to not constrain batch formation.
      features_length_fn: A function mapping features to a sequence length.
      labels_length_fn: A function mapping labels to a sequence length.
      maximum_features_length: The maximum length or list of maximum lengths of
        the features sequence(s). ``None`` to not constrain the length.
      maximum_labels_length: The maximum length of the labels sequence.
        ``None`` to not constrain the length.
      single_pass: If ``True``, makes a single pass over the training data.
      num_shards: The number of data shards (usually the number of workers in a
        distributed setting).
      shard_index: The shard index this data pipeline should read from.
      num_threads: The number of elements processed in parallel.
      dataset_size: If the dataset size is already known, it can be passed here to
        avoid a slower generic computation of the dataset size later.
      shuffle_buffer_size: The number of elements from which to sample.
      prefetch_buffer_size: The number of batches to prefetch asynchronously. If
        ``None``, use an automatically tuned value.
      cardinality_multiple: Ensure that the dataset cardinality is a multiple of
        this value when :obj:`single_pass` is ``True``.

    Returns:
      A ``tf.data.Dataset`` transformation.

    See Also:
      - :func:`opennmt.data.batch_sequence_dataset`
      - :func:`opennmt.data.make_cardinality_multiple_of`
      - :func:`opennmt.data.filter_examples_by_length`
      - :func:`opennmt.data.filter_irregular_batches`
      - :func:`opennmt.data.shuffle_dataset`
    """ # Update dataset_size based on the shard size. # Weighted dataset is repeated before sampling.
A ``tf.data.Dataset`` transformation.
A ` ` tf . data . Dataset ` ` transformation .
batch_size
The batch size to use.
The batch size to use .
batch_type
The training batching strategy to use: can be "examples" or
"tokens".
The training batching strategy to use : can be " examples " or " tokens " .
batch_multiplier
The batch size multiplier.
The batch size multiplier .
batch_size_multiple
When :obj:`batch_type` is "tokens", ensure that the
resulting batch size is a multiple of this value.
When : obj : ` batch_type ` is " tokens " ensure that the resulting batch size is a multiple of this value .
process_fn
The processing function to apply on each element.
The processing function to apply on each element .
transform_fns
List of dataset transformation functions (applied after
:obj:`process_fn` if defined).
List of dataset transformation functions ( applied after : obj : ` process_fn ` if defined ) .
length_bucket_width
The width of the length buckets to select batch
candidates from. ``None`` to not constrain batch formation.
The width of the length buckets to select batch candidates from . ` ` None ` ` to not constrain batch formation .
features_length_fn
A function mapping features to a sequence length.
A function mapping features to a sequence length .
labels_length_fn
A function mapping labels to a sequence length.
A function mapping labels to a sequence length .
maximum_features_length
The maximum length or list of maximum lengths of
the features sequence(s). ``None`` to not constrain the length.
The maximum length or list of maximum lengths of the features sequence ( s ) . ` ` None ` ` to not constrain the length .
maximum_labels_length
The maximum length of the labels sequence.
``None`` to not constrain the length.
The maximum length of the labels sequence . ` ` None ` ` to not constrain the length .
single_pass
If ``True``, makes a single pass over the training data.
If ` ` True ` ` makes a single pass over the training data .
num_shards
The number of data shards (usually the number of workers in a
distributed setting).
The number of data shards ( usually the number of workers in a distributed setting ) .
shard_index
The shard index this data pipeline should read from.
The shard index this data pipeline should read from .
num_threads
The number of elements processed in parallel.
The number of elements processed in parallel .
dataset_size
If the dataset size is already known, it can be passed here to
avoid a slower generic computation of the dataset size later.
If the dataset size is already known it can be passed here to avoid a slower generic computation of the dataset size later .
shuffle_buffer_size
The number of elements from which to sample.
The number of elements from which to sample .
prefetch_buffer_size
The number of batches to prefetch asynchronously. If
``None``, use an automatically tuned value.
The number of batches to prefetch asynchronously . If ` ` None ` ` use an automatically tuned value .
cardinality_multiple
Ensure that the dataset cardinality is a multiple of
this value when :obj:`single_pass` is ``True``.
Ensure that the dataset cardinality is a multiple of this value when : obj : ` single_pass ` is ` ` True ` ` .
examples
>>> dataset = dataset.apply(opennmt.data.training_pipeline(...))
>>> dataset = dataset . apply ( opennmt . data . training_pipeline ( ... ))