python/ray/data/dataset.py
def window(
        self,
        *,
        blocks_per_window: Optional[int] = None,
        bytes_per_window: Optional[int] = None,
    ) -> "DatasetPipeline[T]":
        """Convert this into a DatasetPipeline by windowing over data blocks.

        Transformations prior to the call to ``window()`` are evaluated in
        bulk on the entire dataset. Transformations done on the returned
        pipeline are evaluated incrementally per window of blocks as data is
        read from the output of the pipeline.

        Windowing execution allows for output to be read sooner without
        waiting for all transformations to fully execute, and can also improve
        efficiency if transforms use different resources (e.g., GPUs).

        Without windowing::

            [preprocessing......]
                                  [inference.......]
                                                     [write........]
            Time ----------------------------------------------------------->

        With windowing::

            [prep1] [prep2] [prep3]
                    [infer1] [infer2] [infer3]
                             [write1] [write2] [write3]
            Time ----------------------------------------------------------->

        Examples:
            >>> import ray
            >>> # Create an inference pipeline.
            >>> ds = ray.data.read_binary_files(dir) # doctest: +SKIP
            >>> infer = ... # doctest: +SKIP
            >>> pipe = ds.window(blocks_per_window=10).map(infer) # doctest: +SKIP
            DatasetPipeline(num_windows=40, num_stages=2)
            >>> # The higher the stage parallelism, the shorter the pipeline.
            >>> pipe = ds.window(blocks_per_window=20).map(infer) # doctest: +SKIP
            DatasetPipeline(num_windows=20, num_stages=2)
            >>> # Outputs can be incrementally read from the pipeline.
            >>> for item in pipe.iter_rows(): # doctest: +SKIP
            ...    print(item) # doctest: +SKIP

        Args:
            blocks_per_window: The window size (parallelism) in blocks.
                Increasing window size increases pipeline throughput, but also
                increases the latency to initial output, since it decreases the
                length of the pipeline. Setting this to infinity effectively
                disables pipelining.
            bytes_per_window: Specify the window size in bytes instead of blocks.
                This will be treated as an upper bound for the window size, but each
                window will still include at least one block. This is mutually
                exclusive with ``blocks_per_window``.
        """
        from ray.data.dataset_pipeline import DatasetPipeline
        from ray.data.impl.plan import _rewrite_read_stage

        if blocks_per_window is not None and bytes_per_window is not None:
            raise ValueError("Only one windowing scheme can be specified.")

        if blocks_per_window is None:
            blocks_per_window = 10

        ctx = DatasetContext.get_current()
        if self._plan.is_read_stage() and ctx.optimize_fuse_read_stages:
            blocks, _ = self._plan._get_source_blocks()
            blocks.clear()
            blocks, outer_stats, read_stage = _rewrite_read_stage(blocks)
        else:
            blocks = self._plan.execute()
            outer_stats = self._plan.stats()
            read_stage = None

        class Iterator:
            def __init__(self, splits, epoch):
                self._splits = splits.copy()
                self._epoch = epoch

            def __next__(self) -> "Dataset[T]":
                if not self._splits:
                    raise StopIteration

                blocks = self._splits.pop(0)

                def gen():
                    ds = Dataset(
                        ExecutionPlan(blocks, outer_stats), self._epoch, lazy=False
                    )
                    return ds

                return gen

        class Iterable:
            def __init__(self, blocks, epoch):
                if bytes_per_window:
                    self._splits = blocks.split_by_bytes(bytes_per_window)
                else:
                    self._splits = blocks.split(split_size=blocks_per_window)
                try:
                    sizes = [s.size_bytes() for s in self._splits]
                    assert [s > 0 for s in sizes], sizes

                    def fmt(size_bytes):
                        if size_bytes > 10 * 1024:
                            return "{}MiB".format(round(size_bytes / (1024 * 1024), 2))
                        else:
                            return "{}b".format(size_bytes)

                    logger.info(
                        "Created DatasetPipeline with {} windows: "
                        "{} min, {} max, {} mean".format(
                            len(self._splits),
                            fmt(min(sizes)),
                            fmt(max(sizes)),
                            fmt(int(np.mean(sizes))),
                        )
                    )
                except Exception as e:
                    logger.info(
                        "Created DatasetPipeline with {} windows; "
                        "error getting sizes: {}".format(
                            len(self._splits),
                            e,
                        )
                    )
                self._epoch = epoch

            def __iter__(self):
                return Iterator(self._splits, self._epoch)

        it = Iterable(blocks, self._epoch)
        pipe = DatasetPipeline(it, length=len(it._splits))
        if read_stage:
            pipe = pipe.foreach_window(
                lambda ds, read_stage=read_stage: Dataset(
                    ds._plan.with_stage(read_stage), ds._epoch, True
                )
            )
        return pipe
Convert this into a DatasetPipeline by windowing over data blocks.

        Transformations prior to the call to ``window()`` are evaluated in
        bulk on the entire dataset. Transformations done on the returned
        pipeline are evaluated incrementally per window of blocks as data is
        read from the output of the pipeline.

        Windowing execution allows for output to be read sooner without
        waiting for all transformations to fully execute, and can also improve
        efficiency if transforms use different resources (e.g., GPUs).

        Without windowing::

            [preprocessing......]
                                  [inference.......]
                                                     [write........]
            Time ----------------------------------------------------------->

        With windowing::

            [prep1] [prep2] [prep3]
                    [infer1] [infer2] [infer3]
                             [write1] [write2] [write3]
            Time ----------------------------------------------------------->

        Examples:
            >>> import ray
            >>> # Create an inference pipeline.
            >>> ds = ray.data.read_binary_files(dir) # doctest: +SKIP
            >>> infer = ... # doctest: +SKIP
            >>> pipe = ds.window(blocks_per_window=10).map(infer) # doctest: +SKIP
            DatasetPipeline(num_windows=40, num_stages=2)
            >>> # The higher the stage parallelism, the shorter the pipeline.
            >>> pipe = ds.window(blocks_per_window=20).map(infer) # doctest: +SKIP
            DatasetPipeline(num_windows=20, num_stages=2)
            >>> # Outputs can be incrementally read from the pipeline.
            >>> for item in pipe.iter_rows(): # doctest: +SKIP
            ...    print(item) # doctest: +SKIP

        Args:
            blocks_per_window: The window size (parallelism) in blocks.
                Increasing window size increases pipeline throughput, but also
                increases the latency to initial output, since it decreases the
                length of the pipeline. Setting this to infinity effectively
                disables pipelining.
            bytes_per_window: Specify the window size in bytes instead of blocks.
                This will be treated as an upper bound for the window size, but each
                window will still include at least one block. This is mutually
                exclusive with ``blocks_per_window``.
        
Convert this into a DatasetPipeline by windowing over data blocks.
Transformations prior to the call to ``window()`` are evaluated in
bulk on the entire dataset. Transformations done on the returned
pipeline are evaluated incrementally per window of blocks as data is
read from the output of the pipeline.

Windowing execution allows for output to be read sooner without
waiting for all transformations to fully execute, and can also improve
efficiency if transforms use different resources .

Without windowing:.



With windowing:.


Convert this into a DatasetPipeline by windowing over data blocks . Transformations prior to the call to ` ` window () ` ` are evaluated in bulk on the entire dataset . Transformations done on the returned pipeline are evaluated incrementally per window of blocks as data is read from the output of the pipeline . Windowing execution allows for output to be read sooner without waiting for all transformations to fully execute and can also improve efficiency if transforms use different resources . Without windowing : . With windowing : .
def window(
        self,
        *,
        blocks_per_window: Optional[int] = None,
        bytes_per_window: Optional[int] = None,
    ) -> "DatasetPipeline[T]":
        from ray.data.dataset_pipeline import DatasetPipeline
        from ray.data.impl.plan import _rewrite_read_stage
        if blocks_per_window is not None and bytes_per_window is not None:
            raise ValueError("Only one windowing scheme can be specified.")
        if blocks_per_window is None:
            blocks_per_window = 10
        ctx = DatasetContext.get_current()
        if self._plan.is_read_stage() and ctx.optimize_fuse_read_stages:
            blocks, _ = self._plan._get_source_blocks()
            blocks.clear()
            blocks, outer_stats, read_stage = _rewrite_read_stage(blocks)
        else:
            blocks = self._plan.execute()
            outer_stats = self._plan.stats()
            read_stage = None
        class Iterator:
            def __init__(self, splits, epoch):
                self._splits = splits.copy()
                self._epoch = epoch
            def __next__(self) -> "Dataset[T]":
                if not self._splits:
                    raise StopIteration
                blocks = self._splits.pop(0)
                def gen():
                    ds = Dataset(
                        ExecutionPlan(blocks, outer_stats), self._epoch, lazy=False
                    )
                    return ds
                return gen
        class Iterable:
            def __init__(self, blocks, epoch):
                if bytes_per_window:
                    self._splits = blocks.split_by_bytes(bytes_per_window)
                else:
                    self._splits = blocks.split(split_size=blocks_per_window)
                try:
                    sizes = [s.size_bytes() for s in self._splits]
                    assert [s > 0 for s in sizes], sizes
                    def fmt(size_bytes):
                        if size_bytes > 10 * 1024:
                            return "{}MiB".format(round(size_bytes / (1024 * 1024), 2))
                        else:
                            return "{}b".format(size_bytes)
                    logger.info(
                        "Created DatasetPipeline with {} windows: "
                        "{} min, {} max, {} mean".format(
                            len(self._splits),
                            fmt(min(sizes)),
                            fmt(max(sizes)),
                            fmt(int(np.mean(sizes))),
                        )
                    )
                except Exception as e:
                    logger.info(
                        "Created DatasetPipeline with {} windows; "
                        "error getting sizes: {}".format(
                            len(self._splits),
                            e,
                        )
                    )
                self._epoch = epoch
            def __iter__(self):
                return Iterator(self._splits, self._epoch)
        it = Iterable(blocks, self._epoch)
        pipe = DatasetPipeline(it, length=len(it._splits))
        if read_stage:
            pipe = pipe.foreach_window(
                lambda ds, read_stage=read_stage: Dataset(
                    ds._plan.with_stage(read_stage), ds._epoch, True
                )
            )
        return pipe
def window ( self , * , blocks_per_window : Optional [ int ] = None , bytes_per_window : Optional [ int ] = None , ) -> "DatasetPipeline[T]" : from ray . data . dataset_pipeline import DatasetPipeline from ray . data . impl . plan import _rewrite_read_stage if blocks_per_window is not None and bytes_per_window is not None : raise ValueError ( "Only one windowing scheme can be specified." ) if blocks_per_window is None : blocks_per_window = 10 ctx = DatasetContext . get_current ( ) if self . _plan . is_read_stage ( ) and ctx . optimize_fuse_read_stages : blocks , _ = self . _plan . _get_source_blocks ( ) blocks . clear ( ) blocks , outer_stats , read_stage = _rewrite_read_stage ( blocks ) else : blocks = self . _plan . execute ( ) outer_stats = self . _plan . stats ( ) read_stage = None class Iterator : def __init__ ( self , splits , epoch ) : self . _splits = splits . copy ( ) self . _epoch = epoch def __next__ ( self ) -> "Dataset[T]" : if not self . _splits : raise StopIteration blocks = self . _splits . pop ( 0 ) def gen ( ) : ds = Dataset ( ExecutionPlan ( blocks , outer_stats ) , self . _epoch , lazy = False ) return ds return gen class Iterable : def __init__ ( self , blocks , epoch ) : if bytes_per_window : self . _splits = blocks . split_by_bytes ( bytes_per_window ) else : self . _splits = blocks . split ( split_size = blocks_per_window ) try : sizes = [ s . size_bytes ( ) for s in self . _splits ] assert [ s > 0 for s in sizes ] , sizes def fmt ( size_bytes ) : if size_bytes > 10 * 1024 : return "{}MiB" . format ( round ( size_bytes / ( 1024 * 1024 ) , 2 ) ) else : return "{}b" . format ( size_bytes ) logger . info ( "Created DatasetPipeline with {} windows: " "{} min, {} max, {} mean" . format ( len ( self . _splits ) , fmt ( min ( sizes ) ) , fmt ( max ( sizes ) ) , fmt ( int ( np . mean ( sizes ) ) ) , ) ) except Exception as e : logger . info ( "Created DatasetPipeline with {} windows; " "error getting sizes: {}" . format ( len ( self . _splits ) , e , ) ) self . _epoch = epoch def __iter__ ( self ) : return Iterator ( self . _splits , self . _epoch ) it = Iterable ( blocks , self . _epoch ) pipe = DatasetPipeline ( it , length = len ( it . _splits ) ) if read_stage : pipe = pipe . foreach_window ( lambda ds , read_stage = read_stage : Dataset ( ds . _plan . with_stage ( read_stage ) , ds . _epoch , True ) ) return pipe
Convert this into a DatasetPipeline by windowing over data blocks.
Convert this into a DatasetPipeline by windowing over data blocks .
"""Convert this into a DatasetPipeline by windowing over data blocks.

        Transformations prior to the call to ``window()`` are evaluated in
        bulk on the entire dataset. Transformations done on the returned
        pipeline are evaluated incrementally per window of blocks as data is
        read from the output of the pipeline.

        Windowing execution allows for output to be read sooner without
        waiting for all transformations to fully execute, and can also improve
        efficiency if transforms use different resources (e.g., GPUs).

        Without windowing::

            [preprocessing......]
                                  [inference.......]
                                                     [write........]
            Time ----------------------------------------------------------->

        With windowing::

            [prep1] [prep2] [prep3]
                    [infer1] [infer2] [infer3]
                             [write1] [write2] [write3]
            Time ----------------------------------------------------------->

        Examples:
            >>> import ray
            >>> # Create an inference pipeline.
            >>> ds = ray.data.read_binary_files(dir) # doctest: +SKIP
            >>> infer = ... # doctest: +SKIP
            >>> pipe = ds.window(blocks_per_window=10).map(infer) # doctest: +SKIP
            DatasetPipeline(num_windows=40, num_stages=2)
            >>> # The higher the stage parallelism, the shorter the pipeline.
            >>> pipe = ds.window(blocks_per_window=20).map(infer) # doctest: +SKIP
            DatasetPipeline(num_windows=20, num_stages=2)
            >>> # Outputs can be incrementally read from the pipeline.
            >>> for item in pipe.iter_rows(): # doctest: +SKIP
            ...    print(item) # doctest: +SKIP

        Args:
            blocks_per_window: The window size (parallelism) in blocks.
                Increasing window size increases pipeline throughput, but also
                increases the latency to initial output, since it decreases the
                length of the pipeline. Setting this to infinity effectively
                disables pipelining.
            bytes_per_window: Specify the window size in bytes instead of blocks.
                This will be treated as an upper bound for the window size, but each
                window will still include at least one block. This is mutually
                exclusive with ``blocks_per_window``.
        """
self
blocks_per_window
Optional[int]
The window size (parallelism) in blocks.
Increasing window size increases pipeline throughput, but also
increases the latency to initial output, since it decreases the
length of the pipeline. Setting this to infinity effectively
disables pipelining.
The window size ( parallelism ) in blocks . Increasing window size increases pipeline throughput but also increases the latency to initial output since it decreases the length of the pipeline . Setting this to infinity effectively disables pipelining .
bytes_per_window
Optional[int]
Specify the window size in bytes instead of blocks.
This will be treated as an upper bound for the window size, but each
window will still include at least one block. This is mutually
exclusive with ``blocks_per_window``.
Specify the window size in bytes instead of blocks . This will be treated as an upper bound for the window size but each window will still include at least one block . This is mutually exclusive with ` ` blocks_per_window ` ` .
examples
>>> import ray
>>> # Create an inference pipeline.
>>> ds = ray.data.read_binary_files(dir) # doctest: +SKIP
>>> infer = ... # doctest: +SKIP
>>> pipe = ds.window(blocks_per_window=10).map(infer) # doctest: +SKIP
DatasetPipeline(num_windows=40, num_stages=2)
>>> # The higher the stage parallelism, the shorter the pipeline.
>>> pipe = ds.window(blocks_per_window=20).map(infer) # doctest: +SKIP
DatasetPipeline(num_windows=20, num_stages=2)
>>> # Outputs can be incrementally read from the pipeline.
>>> for item in pipe.iter_rows(): # doctest: +SKIP
print(item) # doctest: +SKIP
>>> import ray >>> # Create an inference pipeline . >>> ds = ray . data . read_binary_files ( dir ) # doctest : + SKIP >>> infer = ... # doctest : + SKIP >>> pipe = ds . window ( blocks_per_window = 10 ) . map ( infer ) # doctest : + SKIP DatasetPipeline ( num_windows = 40 num_stages = 2 ) >>> # The higher the stage parallelism the shorter the pipeline . >>> pipe = ds . window ( blocks_per_window = 20 ) . map ( infer ) # doctest : + SKIP DatasetPipeline ( num_windows = 20 num_stages = 2 ) >>> # Outputs can be incrementally read from the pipeline . >>> for item in pipe . iter_rows () : # doctest : + SKIP print ( item ) # doctest : + SKIP