code: def convert_dataset(
      dataset_dir: str,
      out_dir: str,
      out_file_format: str | file_adapters.FileFormat,
      use_beam: bool,
      overwrite: bool = False,
  ) -> None:
      """Convert a dataset from one file format to another format."""
      dataset_dir = epath.Path(dataset_dir)
      out_path = epath.Path(out_dir)
  
      if dataset_dir == out_path:
          raise ValueError(
              f"The dataset dir ({dataset_dir}) is the same as the specified out"
              f" directory ({out_dir})"
          )
  
      if overwrite:
          out_path.unlink(missing_ok=True)
      out_path.mkdir(parents=True, exist_ok=False)
  
      if isinstance(out_file_format, str):
          out_file_format = file_adapters.file_format_from_suffix(
              out_file_format
          )
      out_file_adapter = file_adapters.ADAPTER_FOR_FORMAT[
          out_file_format
      ]
  
      builder = read_only_builder_lib.builder_from_directory(
          dataset_dir
      )
      if out_file_format == builder.info.file_format:
          raise ValueError(
              f"The file format of the dataset ({builder.info.file_format}) is the"
              f" same as the specified out file format! ({out_file_format})"
          )
      in_file_adapter = file_adapters.ADAPTER_FOR_FORMAT[
          builder.info.file_format
      ]
  
      logging.info(
          "Converting dataset in %s from %s to %s., storing in %s",
          dataset_dir,
          builder.info.file_format,
          out_file_format,
          out_path,
      )
      with py_utils.incomplete_dir(out_path) as tmp_dir:
          tmp_dir = epath.Path(tmp_dir)
          shard_instructions = get_all_shard_instructions(
              info=builder.info,
              out_file_format=out_file_format,
              out_path=tmp_dir,
              in_file_adapter=in_file_adapter,
              out_file_adapter=out_file_adapter,
          )
          if use_beam:
              runner = None
              with beam.Pipeline(runner=runner) as pipeline:
                  _ = (
                      pipeline
                      | beam.Create(shard_instructions)
                      | beam.Map(
                          lambda shard_instruction: shard_instruction.convert()
                      )
                  )
  
          else:
              for shard_instruction in shard_instructions:
                  shard_instruction.convert()
  
          logging.info("Converting metadata in %s.", dataset_dir)
          convert_metadata(
              info=builder.info,
              out_file_format=out_file_format,
              out_path=tmp_dir,
          )
  
          logging.info("Removing incomplete files in %s.", dataset_dir)
          num_incomplete_files = 0
          for incomplete_file in tmp_dir.glob(
              f"*{constants.INCOMPLETE_SUFFIX}.*"
          ):
              incomplete_file.unlink()
              num_incomplete_files += 1
          logging.info(
              "Removed %d incomplete files.", num_incomplete_files
          )
  
      logging.info(
          "Dataset in %s successfully converted to %s.",
          dataset_dir,
          out_path,
      )
  
imports: from collections.abc import Iterator
  import dataclasses
  from typing import Type
  from absl import logging
  from etils import epath
  from tensorflow_datasets.core import constants
  from tensorflow_datasets.core import dataset_info
  from tensorflow_datasets.core import file_adapters
  from tensorflow_datasets.core import naming
  from tensorflow_datasets.core import read_only_builder
  from tensorflow_datasets.core import splits
  from tensorflow_datasets.core.utils import py_utils
  from tensorflow_datasets.core.utils import type_utils
  from tensorflow_datasets.core.utils.lazy_imports_utils import apache_beam
label_desc: Use image node_id 5 for calling a global function with example usage: convert_dataset(dataset_dir, out_dir, out_file_format, use_beam, overwrite) without return types