python/ray/data/dataset_pipeline.py
def _optimize_stages(self):
        """Optimize this pipeline, fusing stages together as possible."""
        context = DatasetContext.get_current()

        if not context.optimize_fuse_stages:
            self._optimized_stages = self._stages
            return

        # This dummy dataset will be used to get a set of optimized stages.
        dummy_ds = Dataset(
            ExecutionPlan(BlockList([], []), DatasetStats(stages={}, parent=None)),
            0,
            True,
        )
        # Apply all pipeline operations to the dummy dataset.
        for stage in self._stages:
            dummy_ds = stage(dummy_ds)
        # Get the optimized stages.
        _, _, stages = dummy_ds._plan._optimize()
        # Apply these optimized stages to the datasets underlying the pipeline.
        # These optimized stages will be executed by the PipelineExecutor.
        optimized_stages = []
        for stage in stages:
            optimized_stages.append(
                lambda ds, stage=stage: Dataset(
                    ds._plan.with_stage(stage), ds._epoch, True
                )
            )
        self._optimized_stages = optimized_stages
Optimize this pipeline, fusing stages together as possible.
Optimize this pipeline, fusing stages together as possible.
Optimize this pipeline fusing stages together as possible .
def _optimize_stages(self):
        context = DatasetContext.get_current()
        if not context.optimize_fuse_stages:
            self._optimized_stages = self._stages
            return
        dummy_ds = Dataset(
            ExecutionPlan(BlockList([], []), DatasetStats(stages={}, parent=None)),
            0,
            True,
        )
        for stage in self._stages:
            dummy_ds = stage(dummy_ds)
        _, _, stages = dummy_ds._plan._optimize()
        optimized_stages = []
        for stage in stages:
            optimized_stages.append(
                lambda ds, stage=stage: Dataset(
                    ds._plan.with_stage(stage), ds._epoch, True
                )
            )
        self._optimized_stages = optimized_stages
def _optimize_stages ( self ) : context = DatasetContext . get_current ( ) if not context . optimize_fuse_stages : self . _optimized_stages = self . _stages return dummy_ds = Dataset ( ExecutionPlan ( BlockList ( [ ] , [ ] ) , DatasetStats ( stages = { } , parent = None ) ) , 0 , True , ) for stage in self . _stages : dummy_ds = stage ( dummy_ds ) _ , _ , stages = dummy_ds . _plan . _optimize ( ) optimized_stages = [ ] for stage in stages : optimized_stages . append ( lambda ds , stage = stage : Dataset ( ds . _plan . with_stage ( stage ) , ds . _epoch , True ) ) self . _optimized_stages = optimized_stages
Optimize this pipeline, fusing stages together as possible.
Optimize this pipeline fusing stages together as possible .
"""Optimize this pipeline, fusing stages together as possible.""" # This dummy dataset will be used to get a set of optimized stages. # Apply all pipeline operations to the dummy dataset. # Get the optimized stages. # Apply these optimized stages to the datasets underlying the pipeline. # These optimized stages will be executed by the PipelineExecutor.
self