code: def _validate_and_transform_prediction_result(
      preds: np.ndarray
      | Mapping[str, np.ndarray]
      | List[Mapping[str, Any]],
      num_input_rows: int,
      return_type: DataType,
  ) -> pd.DataFrame | pd.Series:
      """Validate numpy-based model predictions against the expected pandas_udf return_type and
      transforms the predictions into an equivalent pandas DataFrame or Series.
      """
      if isinstance(return_type, StructType):
          struct_rtype: StructType = return_type
          fieldNames = struct_rtype.names
          if isinstance(preds, dict):
              # dictionary of columns
              predNames = list(preds.keys())
              for field in struct_rtype.fields:
                  if isinstance(field.dataType, ArrayType):
                      if len(preds[field.name].shape) == 2:
                          preds[field.name] = list(preds[field.name])
                      else:
                          raise ValueError(
                              "Prediction results for ArrayType must be two-dimensional."
                          )
                  elif isinstance(
                      field.dataType, supported_scalar_types
                  ):
                      if len(preds[field.name].shape) != 1:
                          raise ValueError(
                              "Prediction results for scalar types must be one-dimensional."
                          )
                  else:
                      raise ValueError(
                          "Unsupported field type in return struct type."
                      )
  
                  if len(preds[field.name]) != num_input_rows:
                      raise ValueError(
                          "Prediction results must have same length as input data"
                      )
  
          elif isinstance(preds, list) and isinstance(preds[0], dict):
              # rows of dictionaries
              predNames = list(preds[0].keys())
              if len(preds) != num_input_rows:
                  raise ValueError(
                      "Prediction results must have same length as input data."
                  )
              for field in struct_rtype.fields:
                  if isinstance(field.dataType, ArrayType):
                      if len(preds[0][field.name].shape) != 1:
                          raise ValueError(
                              "Prediction results for ArrayType must be one-dimensional."
                          )
                  elif isinstance(
                      field.dataType, supported_scalar_types
                  ):
                      if not np.isscalar(preds[0][field.name]):
                          raise ValueError(
                              "Invalid scalar prediction result."
                          )
                  else:
                      raise ValueError(
                          "Unsupported field type in return struct type."
                      )
          else:
              raise ValueError(
                  "Prediction results for StructType must be a dictionary or "
                  "a list of dictionary, got: {}".format(type(preds))
              )
  
          # check column names
          if set(predNames) != set(fieldNames):
              raise ValueError(
                  "Prediction result columns did not match expected return_type "
                  "columns: expected {}, got: {}".format(
                      fieldNames, predNames
                  )
              )
  
          return pd.DataFrame(preds)
      elif isinstance(return_type, ArrayType):
          if isinstance(preds, np.ndarray):
              if len(preds) != num_input_rows:
                  raise ValueError(
                      "Prediction results must have same length as input data."
                  )
              if len(preds.shape) != 2:
                  raise ValueError(
                      "Prediction results for ArrayType must be two-dimensional."
                  )
          else:
              raise ValueError(
                  "Prediction results for ArrayType must be an ndarray."
              )
  
          return pd.Series(list(preds))
      elif isinstance(return_type, supported_scalar_types):
          preds_array: np.ndarray = preds  # type: ignore
          if len(preds_array) != num_input_rows:
              raise ValueError(
                  "Prediction results must have same length as input data."
              )
          if not (
              (
                  len(preds_array.shape) == 2
                  and preds_array.shape[1] == 1
              )
              or len(preds_array.shape) == 1
          ):
              raise ValueError(
                  "Invalid shape for scalar prediction result."
              )
  
          output = (
              np.squeeze(preds_array, -1)
              if len(preds_array.shape) > 1
              else preds_array
          )
          return pd.Series(output).astype(output.dtype)
      else:
          raise ValueError("Unsupported return type")
  
file_path: spark/python/pyspark/ml/functions.py
imports: from __future__ import annotations
  import inspect
  import uuid
  from typing import Any, Callable, Iterator, List, Mapping, TYPE_CHECKING, Tuple, Union, Optional
  import numpy
  from pyspark import SparkContext
  from pyspark.sql.functions import pandas_udf
  from pyspark.sql.column import Column, _to_java_column
  from pyspark.sql.types import ArrayType, ByteType, DataType, DoubleType, FloatType, IntegerType, LongType, ShortType, StringType, StructType
  from pyspark.ml.util import try_remote_functions
label_desc: Use image node_id 8 for calling a global function with example usage: _validate_and_transform_prediction_result(preds, num_input_rows, return_type) and returns: pd, pd, pd