code: def _apply(
      self,
      block_fn: BlockTransform,
      remote_args: dict,
      blocks: BlockList,
      clear_input_blocks: bool,
  ) -> BlockList:
      raise NotImplementedError
  
imports: import collections
  import logging
  import math
  from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, TypeVar, Union
  import ray
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata, BlockPartition, CallableClass, UserDefinedFunction
  from ray.data.context import DataContext
  from ray.types import ObjectRef
  from ray.util.annotations import DeveloperAPI, PublicAPI
label_desc: Use image node_id 1 for calling the ComputeStrategy obj's underlying member method code with example usage: obj._apply(block_fn, remote_args, blocks, clear_input_blocks) without return types