python/ray/data/dataset.py
def map(
        self,
        fn: Union[CallableClass, Callable[[T], U]],
        *,
        compute: Optional[str] = None,
        **ray_remote_args,
    ) -> "Dataset[U]":
        """Apply the given function to each record of this dataset.

        This is a blocking operation. Note that mapping individual records
        can be quite slow. Consider using `.map_batches()` for performance.

        Examples:
            >>> import ray
            >>> # Transform python objects.
            >>> ds = ray.data.range(1000) # doctest: +SKIP
            >>> ds.map(lambda x: x * 2) # doctest: +SKIP
            >>> # Transform Arrow records.
            >>> ds = ray.data.from_items( # doctest: +SKIP
            ...     [{"value": i} for i in range(1000)])
            >>> ds.map(lambda record: {"v2": record["value"] * 2}) # doctest: +SKIP
            >>> # Define a callable class that persists state across
            >>> # function invocations for efficiency.
            >>> init_model = ... # doctest: +SKIP
            >>> class CachedModel:
            ...    def __init__(self):
            ...        self.model = init_model()
            ...    def __call__(self, batch):
            ...        return self.model(batch)
            >>> # Apply the transform in parallel on GPUs. Since
            >>> # compute=ActorPoolStrategy(2, 8) the transform will be applied on an
            >>> # autoscaling pool of 2-8 Ray actors, each allocated 1 GPU by Ray.
            >>> from ray.data.impl.compute import ActorPoolStrategy
            >>> ds.map(CachedModel, # doctest: +SKIP
            ...        compute=ActorPoolStrategy(2, 8),
            ...        num_gpus=1)

        Time complexity: O(dataset size / parallelism)

        Args:
            fn: The function to apply to each record, or a class type
                that can be instantiated to create such a callable. Callable classes are
                only supported for the actor compute strategy.
            compute: The compute strategy, either "tasks" (default) to use Ray
                tasks, or ActorPoolStrategy(min, max) to use an autoscaling actor pool.
            ray_remote_args: Additional resource requirements to request from
                ray (e.g., num_gpus=1 to request GPUs for the map tasks).
        """

        self._warn_slow()
        fn = cache_wrapper(fn, compute)
        context = DatasetContext.get_current()

        def transform(block: Block) -> Iterable[Block]:
            DatasetContext._set_current(context)
            block = BlockAccessor.for_block(block)
            output_buffer = BlockOutputBuffer(None, context.target_max_block_size)
            for row in block.iter_rows():
                output_buffer.add(fn(row))
                if output_buffer.has_next():
                    yield output_buffer.next()
            output_buffer.finalize()
            if output_buffer.has_next():
                yield output_buffer.next()

        plan = self._plan.with_stage(
            OneToOneStage("map", transform, compute, ray_remote_args)
        )
        return Dataset(plan, self._epoch, self._lazy)
Apply the given function to each record of this dataset.

        This is a blocking operation. Note that mapping individual records
        can be quite slow. Consider using `.map_batches()` for performance.

        Examples:
            >>> import ray
            >>> # Transform python objects.
            >>> ds = ray.data.range(1000) # doctest: +SKIP
            >>> ds.map(lambda x: x * 2) # doctest: +SKIP
            >>> # Transform Arrow records.
            >>> ds = ray.data.from_items( # doctest: +SKIP
            ...     [{"value": i} for i in range(1000)])
            >>> ds.map(lambda record: {"v2": record["value"] * 2}) # doctest: +SKIP
            >>> # Define a callable class that persists state across
            >>> # function invocations for efficiency.
            >>> init_model = ... # doctest: +SKIP
            >>> class CachedModel:
            ...    def __init__(self):
            ...        self.model = init_model()
            ...    def __call__(self, batch):
            ...        return self.model(batch)
            >>> # Apply the transform in parallel on GPUs. Since
            >>> # compute=ActorPoolStrategy(2, 8) the transform will be applied on an
            >>> # autoscaling pool of 2-8 Ray actors, each allocated 1 GPU by Ray.
            >>> from ray.data.impl.compute import ActorPoolStrategy
            >>> ds.map(CachedModel, # doctest: +SKIP
            ...        compute=ActorPoolStrategy(2, 8),
            ...        num_gpus=1)

        Time complexity: O(dataset size / parallelism)

        Args:
            fn: The function to apply to each record, or a class type
                that can be instantiated to create such a callable. Callable classes are
                only supported for the actor compute strategy.
            compute: The compute strategy, either "tasks" (default) to use Ray
                tasks, or ActorPoolStrategy(min, max) to use an autoscaling actor pool.
            ray_remote_args: Additional resource requirements to request from
                ray (e.g., num_gpus=1 to request GPUs for the map tasks).
        
Apply the given function to each record of this dataset.
This is a blocking operation. Note that mapping individual records
can be quite slow. Consider using `.map_batches()` for performance.
Apply the given function to each record of this dataset . This is a blocking operation . Note that mapping individual records can be quite slow . Consider using ` . map_batches () ` for performance .
def map(
        self,
        fn: Union[CallableClass, Callable[[T], U]],
        *,
        compute: Optional[str] = None,
        **ray_remote_args,
    ) -> "Dataset[U]":
        self._warn_slow()
        fn = cache_wrapper(fn, compute)
        context = DatasetContext.get_current()
        def transform(block: Block) -> Iterable[Block]:
            DatasetContext._set_current(context)
            block = BlockAccessor.for_block(block)
            output_buffer = BlockOutputBuffer(None, context.target_max_block_size)
            for row in block.iter_rows():
                output_buffer.add(fn(row))
                if output_buffer.has_next():
                    yield output_buffer.next()
            output_buffer.finalize()
            if output_buffer.has_next():
                yield output_buffer.next()
        plan = self._plan.with_stage(
            OneToOneStage("map", transform, compute, ray_remote_args)
        )
        return Dataset(plan, self._epoch, self._lazy)
def map ( self , fn : Union [ CallableClass , Callable [ [ T ] , U ] ] , * , compute : Optional [ str ] = None , ** ray_remote_args , ) -> "Dataset[U]" : self . _warn_slow ( ) fn = cache_wrapper ( fn , compute ) context = DatasetContext . get_current ( ) def transform ( block : Block ) -> Iterable [ Block ] : DatasetContext . _set_current ( context ) block = BlockAccessor . for_block ( block ) output_buffer = BlockOutputBuffer ( None , context . target_max_block_size ) for row in block . iter_rows ( ) : output_buffer . add ( fn ( row ) ) if output_buffer . has_next ( ) : yield output_buffer . next ( ) output_buffer . finalize ( ) if output_buffer . has_next ( ) : yield output_buffer . next ( ) plan = self . _plan . with_stage ( OneToOneStage ( "map" , transform , compute , ray_remote_args ) ) return Dataset ( plan , self . _epoch , self . _lazy )
Apply the given function to each record of this dataset.
Apply the given function to each record of this dataset .
"""Apply the given function to each record of this dataset.

        This is a blocking operation. Note that mapping individual records
        can be quite slow. Consider using `.map_batches()` for performance.

        Examples:
            >>> import ray
            >>> # Transform python objects.
            >>> ds = ray.data.range(1000) # doctest: +SKIP
            >>> ds.map(lambda x: x * 2) # doctest: +SKIP
            >>> # Transform Arrow records.
            >>> ds = ray.data.from_items( # doctest: +SKIP
            ...     [{"value": i} for i in range(1000)])
            >>> ds.map(lambda record: {"v2": record["value"] * 2}) # doctest: +SKIP
            >>> # Define a callable class that persists state across
            >>> # function invocations for efficiency.
            >>> init_model = ... # doctest: +SKIP
            >>> class CachedModel:
            ...    def __init__(self):
            ...        self.model = init_model()
            ...    def __call__(self, batch):
            ...        return self.model(batch)
            >>> # Apply the transform in parallel on GPUs. Since
            >>> # compute=ActorPoolStrategy(2, 8) the transform will be applied on an
            >>> # autoscaling pool of 2-8 Ray actors, each allocated 1 GPU by Ray.
            >>> from ray.data.impl.compute import ActorPoolStrategy
            >>> ds.map(CachedModel, # doctest: +SKIP
            ...        compute=ActorPoolStrategy(2, 8),
            ...        num_gpus=1)

        Time complexity: O(dataset size / parallelism)

        Args:
            fn: The function to apply to each record, or a class type
                that can be instantiated to create such a callable. Callable classes are
                only supported for the actor compute strategy.
            compute: The compute strategy, either "tasks" (default) to use Ray
                tasks, or ActorPoolStrategy(min, max) to use an autoscaling actor pool.
            ray_remote_args: Additional resource requirements to request from
                ray (e.g., num_gpus=1 to request GPUs for the map tasks).
        """
self
fn
Union[CallableClass, Callable[[T], U]]
The function to apply to each record, or a class type
that can be instantiated to create such a callable. Callable classes are
only supported for the actor compute strategy.
The function to apply to each record or a class type that can be instantiated to create such a callable . Callable classes are only supported for the actor compute strategy .
compute
Optional[str]
The compute strategy, either "tasks" (default) to use Ray
tasks, or ActorPoolStrategy(min, max) to use an autoscaling actor pool.
The compute strategy either " tasks " ( default ) to use Ray tasks or ActorPoolStrategy ( min max ) to use an autoscaling actor pool .
ray_remote_args
Additional resource requirements to request from
ray .
Additional resource requirements to request from ray .
examples
>>> import ray
>>> # Transform python objects.
>>> ds = ray.data.range(1000) # doctest: +SKIP
>>> ds.map(lambda x: x * 2) # doctest: +SKIP
>>> # Transform Arrow records.
>>> ds = ray.data.from_items( # doctest: +SKIP
[{"value": i} for i in range(1000)])
>>> ds.map(lambda record: {"v2": record["value"] * 2}) # doctest: +SKIP
>>> # Define a callable class that persists state across
>>> # function invocations for efficiency.
>>> import ray >>> # Transform python objects . >>> ds = ray . data . range ( 1000 ) # doctest : + SKIP >>> ds . map ( lambda x : x * 2 ) # doctest : + SKIP >>> # Transform Arrow records . >>> ds = ray . data . from_items ( # doctest : + SKIP [ { " value " : i } for i in range ( 1000 ) ] ) >>> ds . map ( lambda record : { " v2 " : record [ " value " ] * 2 } ) # doctest : + SKIP >>> # Define a callable class that persists state across >>> # function invocations for efficiency .