python/ray/data/dataset_pipeline.py
def rewindow(
        self, *, blocks_per_window: int, preserve_epoch: bool = True
    ) -> "DatasetPipeline[T]":
        """Change the windowing (blocks per dataset) of this pipeline.

        Changes the windowing of this pipeline to the specified size. For
        example, if the current pipeline has two blocks per dataset, and
        `.rewindow(blocks_per_window=4)` is requested, adjacent datasets will
        be merged until each dataset is 4 blocks. If
        `.rewindow(blocks_per_window)` was requested the datasets will be
        split into smaller windows.

        Args:
            blocks_per_window: The new target blocks per window.
            preserve_epoch: Whether to preserve epoch boundaries. If set to
                False, then windows can contain data from two adjacent epochs.
        """

        class WindowIterator:
            def __init__(self, original_iter):
                self._original_iter = original_iter
                self._buffer: Optional[Dataset[T]] = None

            def __next__(self) -> Dataset[T]:
                try:
                    # Merge windows until we meet the requested window size.
                    if self._buffer is None:
                        self._buffer = next(self._original_iter)
                    while self._buffer.num_blocks() < blocks_per_window:
                        next_ds = next(self._original_iter)
                        if (
                            preserve_epoch
                            and self._buffer._get_epoch() != next_ds._get_epoch()
                        ):
                            partial_window = self._buffer
                            self._buffer = next_ds
                            return lambda: partial_window
                        else:
                            self._buffer = self._buffer.union(next_ds)
                    # Slice off the left-most chunk and return it.
                    res, self._buffer = self._buffer._divide(blocks_per_window)
                    assert res.num_blocks() <= blocks_per_window, res
                    if self._buffer.num_blocks() == 0:
                        self._buffer = None
                    return lambda: res
                except StopIteration:
                    # Return the left-over data as a single window.
                    if self._buffer and self._buffer.num_blocks() > 0:
                        res = self._buffer
                        assert res.num_blocks() <= blocks_per_window, res
                        self._buffer = None
                        return lambda: res
                    else:
                        raise

        class WindowIterable:
            def __init__(self, original_iter):
                self._original_iter = original_iter

            def __iter__(self):
                return WindowIterator(self._original_iter)

        if self._length == float("inf"):
            length = float("inf")
        else:
            length = None

        return DatasetPipeline(WindowIterable(self.iter_datasets()), length=length)
Change the windowing (blocks per dataset) of this pipeline.

        Changes the windowing of this pipeline to the specified size. For
        example, if the current pipeline has two blocks per dataset, and
        `.rewindow(blocks_per_window=4)` is requested, adjacent datasets will
        be merged until each dataset is 4 blocks. If
        `.rewindow(blocks_per_window)` was requested the datasets will be
        split into smaller windows.

        Args:
            blocks_per_window: The new target blocks per window.
            preserve_epoch: Whether to preserve epoch boundaries. If set to
                False, then windows can contain data from two adjacent epochs.
        
Change the windowing (blocks per dataset) of this pipeline.
Changes the windowing of this pipeline to the specified size. For
example, if the current pipeline has two blocks per dataset, and
`.rewindow(blocks_per_window=4)` is requested, adjacent datasets will
be merged until each dataset is 4 blocks. If
`.rewindow(blocks_per_window)` was requested the datasets will be
split into smaller windows.
Change the windowing ( blocks per dataset ) of this pipeline . Changes the windowing of this pipeline to the specified size . For example if the current pipeline has two blocks per dataset and ` . rewindow ( blocks_per_window = 4 ) ` is requested adjacent datasets will be merged until each dataset is 4 blocks . If ` . rewindow ( blocks_per_window ) ` was requested the datasets will be split into smaller windows .
def rewindow(
        self, *, blocks_per_window: int, preserve_epoch: bool = True
    ) -> "DatasetPipeline[T]":
        class WindowIterator:
            def __init__(self, original_iter):
                self._original_iter = original_iter
                self._buffer: Optional[Dataset[T]] = None
            def __next__(self) -> Dataset[T]:
                try:
                    if self._buffer is None:
                        self._buffer = next(self._original_iter)
                    while self._buffer.num_blocks() < blocks_per_window:
                        next_ds = next(self._original_iter)
                        if (
                            preserve_epoch
                            and self._buffer._get_epoch() != next_ds._get_epoch()
                        ):
                            partial_window = self._buffer
                            self._buffer = next_ds
                            return lambda: partial_window
                        else:
                            self._buffer = self._buffer.union(next_ds)
                    res, self._buffer = self._buffer._divide(blocks_per_window)
                    assert res.num_blocks() <= blocks_per_window, res
                    if self._buffer.num_blocks() == 0:
                        self._buffer = None
                    return lambda: res
                except StopIteration:
                    if self._buffer and self._buffer.num_blocks() > 0:
                        res = self._buffer
                        assert res.num_blocks() <= blocks_per_window, res
                        self._buffer = None
                        return lambda: res
                    else:
                        raise
        class WindowIterable:
            def __init__(self, original_iter):
                self._original_iter = original_iter
            def __iter__(self):
                return WindowIterator(self._original_iter)
        if self._length == float("inf"):
            length = float("inf")
        else:
            length = None
        return DatasetPipeline(WindowIterable(self.iter_datasets()), length=length)
def rewindow ( self , * , blocks_per_window : int , preserve_epoch : bool = True ) -> "DatasetPipeline[T]" : class WindowIterator : def __init__ ( self , original_iter ) : self . _original_iter = original_iter self . _buffer : Optional [ Dataset [ T ] ] = None def __next__ ( self ) -> Dataset [ T ] : try : if self . _buffer is None : self . _buffer = next ( self . _original_iter ) while self . _buffer . num_blocks ( ) < blocks_per_window : next_ds = next ( self . _original_iter ) if ( preserve_epoch and self . _buffer . _get_epoch ( ) != next_ds . _get_epoch ( ) ) : partial_window = self . _buffer self . _buffer = next_ds return lambda : partial_window else : self . _buffer = self . _buffer . union ( next_ds ) res , self . _buffer = self . _buffer . _divide ( blocks_per_window ) assert res . num_blocks ( ) <= blocks_per_window , res if self . _buffer . num_blocks ( ) == 0 : self . _buffer = None return lambda : res except StopIteration : if self . _buffer and self . _buffer . num_blocks ( ) > 0 : res = self . _buffer assert res . num_blocks ( ) <= blocks_per_window , res self . _buffer = None return lambda : res else : raise class WindowIterable : def __init__ ( self , original_iter ) : self . _original_iter = original_iter def __iter__ ( self ) : return WindowIterator ( self . _original_iter ) if self . _length == float ( "inf" ) : length = float ( "inf" ) else : length = None return DatasetPipeline ( WindowIterable ( self . iter_datasets ( ) ) , length = length )
Change the windowing (blocks per dataset) of this pipeline.
Change the windowing ( blocks per dataset ) of this pipeline .
"""Change the windowing (blocks per dataset) of this pipeline.

        Changes the windowing of this pipeline to the specified size. For
        example, if the current pipeline has two blocks per dataset, and
        `.rewindow(blocks_per_window=4)` is requested, adjacent datasets will
        be merged until each dataset is 4 blocks. If
        `.rewindow(blocks_per_window)` was requested the datasets will be
        split into smaller windows.

        Args:
            blocks_per_window: The new target blocks per window.
            preserve_epoch: Whether to preserve epoch boundaries. If set to
                False, then windows can contain data from two adjacent epochs.
        """ # Merge windows until we meet the requested window size. # Slice off the left-most chunk and return it. # Return the left-over data as a single window.
self
blocks_per_window
int
The new target blocks per window.
The new target blocks per window .
preserve_epoch
bool
Whether to preserve epoch boundaries. If set to
False, then windows can contain data from two adjacent epochs.
Whether to preserve epoch boundaries . If set to False then windows can contain data from two adjacent epochs .