python/ray/data/dataset_pipeline.py
def iter_epochs(self) -> Iterator["DatasetPipeline[T]"]:
        """Split this pipeline up by epoch.

        This allows reading of data per-epoch for repeated Datasets, which is
        useful for ML training. For example, ``ray.data.range(10).repeat(50)``
        generates a pipeline with 500 rows total split across 50 epochs. This
        method allows iterating over the data individually per epoch
        (repetition) of the original data.

        Examples:
            >>> import ray
            >>> epochs = ray.data.range(10).repeat(50).iter_epochs()  # doctest: +SKIP
            >>> for i, epoch in enumerate(epochs): # doctest: +SKIP
            ...     print("Epoch", i) # doctest: +SKIP
            ...     for row in epoch.iter_rows(): # doctest: +SKIP
            ...         print(row) # doctest: +SKIP

        Returns:
            Iterator over epoch objects, where each epoch is a DatasetPipeline
            containing data from that epoch only.
        """

        class Peekable:
            def __init__(self, base_iter: Iterator[T]):
                self._iter = base_iter
                self._buffer = None

            def _fill_buffer_if_possible(self):
                if self._buffer is None:
                    try:
                        self._buffer = next(self._iter)
                        assert self._buffer is not None
                    except StopIteration:
                        pass

            def peek(self) -> T:
                self._fill_buffer_if_possible()
                if self._buffer is None:
                    raise StopIteration
                return self._buffer

            def __next__(self) -> T:
                self._fill_buffer_if_possible()
                if self._buffer is None:
                    raise StopIteration
                item = self._buffer
                self._buffer = None
                return item

        class SingleEpochIterator:
            def __init__(self, peekable_iter: Iterator[Dataset[T]], epoch: int):
                self._iter = peekable_iter
                self._epoch = epoch

            def __next__(self) -> Dataset[T]:
                if self._iter.peek()._get_epoch() > self._epoch:
                    raise StopIteration
                ds = next(self._iter)
                return lambda: ds

            def __iter__(self):
                return self

        class EpochDelimitedIterator:
            def __init__(self, pipe):
                self._iter = Peekable(pipe.iter_datasets())
                self._cur_epoch = None

            def __next__(self) -> "DatasetPipeline[T]":
                if self._cur_epoch is None:
                    self._cur_epoch = self._iter.peek()._get_epoch()
                else:
                    self._cur_epoch += 1
                warned = False
                while self._iter.peek()._get_epoch() < self._cur_epoch:
                    if not warned:
                        warned = True
                        logger.warn(
                            "Data from epoch {} was not fully read, "
                            "skipping to next epoch.".format(self._cur_epoch - 1)
                        )
                    next(self._iter)
                epoch_pipe = DatasetPipeline.from_iterable(
                    SingleEpochIterator(self._iter, epoch=self._cur_epoch)
                )
                return epoch_pipe

            def __iter__(self):
                return self

        return EpochDelimitedIterator(self)
Split this pipeline up by epoch.

        This allows reading of data per-epoch for repeated Datasets, which is
        useful for ML training. For example, ``ray.data.range(10).repeat(50)``
        generates a pipeline with 500 rows total split across 50 epochs. This
        method allows iterating over the data individually per epoch
        (repetition) of the original data.

        Examples:
            >>> import ray
            >>> epochs = ray.data.range(10).repeat(50).iter_epochs()  # doctest: +SKIP
            >>> for i, epoch in enumerate(epochs): # doctest: +SKIP
            ...     print("Epoch", i) # doctest: +SKIP
            ...     for row in epoch.iter_rows(): # doctest: +SKIP
            ...         print(row) # doctest: +SKIP

        Returns:
            Iterator over epoch objects, where each epoch is a DatasetPipeline
            containing data from that epoch only.
        
Split this pipeline up by epoch.
This allows reading of data per-epoch for repeated Datasets, which is
useful for ML training. For example, ``ray.data.range(10).repeat(50)``
generates a pipeline with 500 rows total split across 50 epochs. This
method allows iterating over the data individually per epoch
(repetition) of the original data.
Split this pipeline up by epoch . This allows reading of data per - epoch for repeated Datasets which is useful for ML training . For example ` ` ray . data . range ( 10 ) . repeat ( 50 ) ` ` generates a pipeline with 500 rows total split across 50 epochs . This method allows iterating over the data individually per epoch ( repetition ) of the original data .
def iter_epochs(self) -> Iterator["DatasetPipeline[T]"]:
        class Peekable:
            def __init__(self, base_iter: Iterator[T]):
                self._iter = base_iter
                self._buffer = None
            def _fill_buffer_if_possible(self):
                if self._buffer is None:
                    try:
                        self._buffer = next(self._iter)
                        assert self._buffer is not None
                    except StopIteration:
                        pass
            def peek(self) -> T:
                self._fill_buffer_if_possible()
                if self._buffer is None:
                    raise StopIteration
                return self._buffer
            def __next__(self) -> T:
                self._fill_buffer_if_possible()
                if self._buffer is None:
                    raise StopIteration
                item = self._buffer
                self._buffer = None
                return item
        class SingleEpochIterator:
            def __init__(self, peekable_iter: Iterator[Dataset[T]], epoch: int):
                self._iter = peekable_iter
                self._epoch = epoch
            def __next__(self) -> Dataset[T]:
                if self._iter.peek()._get_epoch() > self._epoch:
                    raise StopIteration
                ds = next(self._iter)
                return lambda: ds
            def __iter__(self):
                return self
        class EpochDelimitedIterator:
            def __init__(self, pipe):
                self._iter = Peekable(pipe.iter_datasets())
                self._cur_epoch = None
            def __next__(self) -> "DatasetPipeline[T]":
                if self._cur_epoch is None:
                    self._cur_epoch = self._iter.peek()._get_epoch()
                else:
                    self._cur_epoch += 1
                warned = False
                while self._iter.peek()._get_epoch() < self._cur_epoch:
                    if not warned:
                        warned = True
                        logger.warn(
                            "Data from epoch {} was not fully read, "
                            "skipping to next epoch.".format(self._cur_epoch - 1)
                        )
                    next(self._iter)
                epoch_pipe = DatasetPipeline.from_iterable(
                    SingleEpochIterator(self._iter, epoch=self._cur_epoch)
                )
                return epoch_pipe
            def __iter__(self):
                return self
        return EpochDelimitedIterator(self)
def iter_epochs ( self ) -> Iterator [ "DatasetPipeline[T]" ] : class Peekable : def __init__ ( self , base_iter : Iterator [ T ] ) : self . _iter = base_iter self . _buffer = None def _fill_buffer_if_possible ( self ) : if self . _buffer is None : try : self . _buffer = next ( self . _iter ) assert self . _buffer is not None except StopIteration : pass def peek ( self ) -> T : self . _fill_buffer_if_possible ( ) if self . _buffer is None : raise StopIteration return self . _buffer def __next__ ( self ) -> T : self . _fill_buffer_if_possible ( ) if self . _buffer is None : raise StopIteration item = self . _buffer self . _buffer = None return item class SingleEpochIterator : def __init__ ( self , peekable_iter : Iterator [ Dataset [ T ] ] , epoch : int ) : self . _iter = peekable_iter self . _epoch = epoch def __next__ ( self ) -> Dataset [ T ] : if self . _iter . peek ( ) . _get_epoch ( ) > self . _epoch : raise StopIteration ds = next ( self . _iter ) return lambda : ds def __iter__ ( self ) : return self class EpochDelimitedIterator : def __init__ ( self , pipe ) : self . _iter = Peekable ( pipe . iter_datasets ( ) ) self . _cur_epoch = None def __next__ ( self ) -> "DatasetPipeline[T]" : if self . _cur_epoch is None : self . _cur_epoch = self . _iter . peek ( ) . _get_epoch ( ) else : self . _cur_epoch += 1 warned = False while self . _iter . peek ( ) . _get_epoch ( ) < self . _cur_epoch : if not warned : warned = True logger . warn ( "Data from epoch {} was not fully read, " "skipping to next epoch." . format ( self . _cur_epoch - 1 ) ) next ( self . _iter ) epoch_pipe = DatasetPipeline . from_iterable ( SingleEpochIterator ( self . _iter , epoch = self . _cur_epoch ) ) return epoch_pipe def __iter__ ( self ) : return self return EpochDelimitedIterator ( self )
Split this pipeline up by epoch.
Split this pipeline up by epoch .
"""Split this pipeline up by epoch.

        This allows reading of data per-epoch for repeated Datasets, which is
        useful for ML training. For example, ``ray.data.range(10).repeat(50)``
        generates a pipeline with 500 rows total split across 50 epochs. This
        method allows iterating over the data individually per epoch
        (repetition) of the original data.

        Examples:
            >>> import ray
            >>> epochs = ray.data.range(10).repeat(50).iter_epochs()  # doctest: +SKIP
            >>> for i, epoch in enumerate(epochs): # doctest: +SKIP
            ...     print("Epoch", i) # doctest: +SKIP
            ...     for row in epoch.iter_rows(): # doctest: +SKIP
            ...         print(row) # doctest: +SKIP

        Returns:
            Iterator over epoch objects, where each epoch is a DatasetPipeline
            containing data from that epoch only.
        """
Iterator over epoch objects, where each epoch is a DatasetPipeline
containing data from that epoch only.
Iterator over epoch objects where each epoch is a DatasetPipeline containing data from that epoch only .
self
examples
>>> import ray
>>> epochs = ray.data.range(10).repeat(50).iter_epochs()  # doctest: +SKIP
>>> for i, epoch in enumerate(epochs): # doctest: +SKIP
print("Epoch", i) # doctest: +SKIP
for row in epoch.iter_rows(): # doctest: +SKIP
print(row) # doctest: +SKIP
>>> import ray >>> epochs = ray . data . range ( 10 ) . repeat ( 50 ) . iter_epochs () # doctest : + SKIP >>> for i epoch in enumerate ( epochs ) : # doctest : + SKIP print ( " Epoch " i ) # doctest : + SKIP for row in epoch . iter_rows () : # doctest : + SKIP print ( row ) # doctest : + SKIP