code: def optimize(
      fn: Callable[[Any], Any],
      inputs: Sequence[Any],
      output_dir: str,
      chunk_size: Optional[int] = None,
      chunk_bytes: Optional[Union[int, str]] = None,
      compression: Optional[str] = None,
      num_workers: Optional[int] = None,
      fast_dev_run: bool = False,
      num_nodes: Optional[int] = None,
      machine: Optional[str] = None,
      num_downloaders: Optional[int] = None,
      reorder_files: bool = True,
  ) -> None:
      """This function converts a dataset into chunks possibly in a distributed way.
  
      Arguments:
          fn: A function to be executed over each input element
          inputs: A sequence of input to be processed by the `fn` function.
              Each input should contain at least a valid filepath.
          output_dir: The folder where the processed data should be stored.
          chunk_size: The maximum number of elements to hold within a chunk.
          chunk_bytes: The maximum number of bytes to hold within a chunk.
          compression: The compression algorithm to use over the chunks.
          num_workers: The number of workers to use during processing
          fast_dev_run: Whether to use process only a sub part of the inputs
          num_nodes: When doing remote execution, the number of nodes to use.
          machine: When doing remote execution, the machine to use.
          num_downloaders: The number of downloaders per worker.
          reorder_files: By default, reorders the files by file size to distribute work equally among all workers.
              Set this to ``False`` if the order in which samples are processed should be preserved.
  
      """
      if not isinstance(inputs, Sequence):
          raise ValueError(
              f"The provided inputs should be non empty sequence. Found {inputs}."
          )
  
      if len(inputs) == 0:
          raise ValueError(
              f"The provided inputs should be non empty. Found {inputs}."
          )
  
      if chunk_size is None and chunk_bytes is None:
          raise ValueError(
              "Either `chunk_size` or `chunk_bytes` needs to be defined."
          )
  
      if (
          num_nodes is None
          or int(os.getenv("DATA_OPTIMIZER_NUM_NODES", 0)) > 0
      ):
          output_dir = _resolve_dir(output_dir)
  
          if (
              output_dir.url is not None
              and "cloudspaces" in output_dir.url
          ):
              raise ValueError(
                  f"The provided `output_dir` isn't valid. Found {output_dir.path}."
                  " HINT: You can either use `/teamspace/s3_connections/...` or `/teamspace/datasets/...`."
              )
  
          _assert_dir_has_index_file(output_dir)
  
          input_dir = _resolve_dir(_get_input_dir(inputs))
  
          data_processor = DataProcessor(
              input_dir=input_dir,
              output_dir=output_dir,
              num_workers=num_workers or os.cpu_count(),
              fast_dev_run=fast_dev_run,
              num_downloaders=num_downloaders,
              reorder_files=reorder_files,
          )
          return data_processor.run(
              LambdaDataChunkRecipe(
                  fn,
                  inputs,
                  chunk_size=chunk_size,
                  chunk_bytes=chunk_bytes,
                  compression=compression,
              )
          )
      return _execute(
          f"data-prep-optimize-{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}",
          num_nodes,
          machine,
      )
  
imports: import inspect
  import os
  from datetime import datetime
  from pathlib import Path
  from types import FunctionType
  from typing import Any, Callable, Dict, Optional, Sequence, Union
  import torch
  from lightning.data.streaming.constants import _LIGHTNING_CLOUD_LATEST, _TORCH_GREATER_EQUAL_2_1_0
  from lightning.data.streaming.data_processor import DataChunkRecipe, DataProcessor, DataTransformRecipe
label_desc: Use image node_id 4 for calling a global function with example usage: optimize(fn, inputs, output_dir, chunk_size, chunk_bytes, compression, num_workers, fast_dev_run, num_nodes, machine, num_downloaders, reorder_files) and returns: _execute, data_processor