code: def write(
      self,
      blocks: Iterable[Block],
      ctx: TaskContext,
  ) -> Any:
      num_rows_written = 0
  
      block_index = 0
      for block in blocks:
          block = BlockAccessor.for_block(block)
          if block.num_rows() == 0:
              continue
  
          self.write_block(block, block_index, ctx)
  
          num_rows_written += block.num_rows()
          block_index += 1
  
      if num_rows_written == 0:
          logger.get_logger().warning(
              f"Skipped writing empty dataset with UUID {self.dataset_uuid} at "
              f"{self.path}.",
          )
          return "skip"
  
      # TODO: decide if we want to return richer object when the task
      # succeeds.
      return "ok"
  
imports: import posixpath
  import warnings
  from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional
  from ray._private.utils import _add_creatable_buckets_param_if_s3_uri
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.util import _is_local_scheme, call_with_retry
  from ray.data.block import Block, BlockAccessor
  from ray.data.context import DataContext
  from ray.data.datasource.block_path_provider import BlockWritePathProvider
  from ray.data.datasource.datasink import Datasink
  from ray.data.datasource.filename_provider import FilenameProvider, _DefaultFilenameProvider
  from ray.data.datasource.path_util import _resolve_paths_and_filesystem
  from ray.util.annotations import DeveloperAPI
label_desc: Use image node_id 4 for calling the _FileDatasink obj's underlying member method code with example usage: obj.write(blocks, ctx) and returns: str, str