code: def postprocess(
      predictions,
      output_features,
      training_set_metadata,
      output_directory="",
      backend=LOCAL_BACKEND,
      skip_save_unprocessed_output=False,
  ) -> DataFrame:
      if not backend.is_coordinator():
          # Only save unprocessed output on the coordinator
          skip_save_unprocessed_output = True
  
      saved_keys = set()
      if not skip_save_unprocessed_output:
          _save_as_numpy(
              predictions, output_directory, saved_keys, backend
          )
  
      def postprocess_batch(df):
          for of_name, output_feature in output_features.items():
              df = output_feature.postprocess_predictions(
                  df,
                  training_set_metadata[of_name],
              )
          return df
  
      # We disable tensor extension casting here because this step is the final data processing step and
      # we do not expect return to Ray Datasets after this point. The dtype of the predictions will be
      # whatever they would be if we did all postprocessing in Dask.
      predictions = backend.df_engine.map_batches(
          predictions,
          postprocess_batch,
          enable_tensor_extension_casting=False,
      )
  
      # Save any new columns but do not save the original columns again
      if not skip_save_unprocessed_output:
          _save_as_numpy(
              predictions, output_directory, saved_keys, backend
          )
  
      return predictions
  
file_path: ludwig/ludwig/data/postprocessing.py
imports: import os
  from typing import Any, Dict, List, Optional, Union
  import numpy
  import pandas
  import torch
  from ludwig.backend import LOCAL_BACKEND
  from ludwig.data.utils import convert_to_dict
  from ludwig.utils.data_utils import DATAFRAME_FORMATS, DICT_FORMATS
  from ludwig.utils.dataframe_utils import to_numpy_dataset
  from ludwig.utils.fs_utils import has_remote_protocol, open_file
  from ludwig.utils.misc_utils import get_from_registry
  from ludwig.utils.strings_utils import make_safe_filename
  from ludwig.utils.types import DataFrame
label_desc: Use image node_id 1 for calling a global function with example usage: postprocess(predictions, output_features, training_set_metadata, output_directory, backend, skip_save_unprocessed_output) and returns: predictions, df