code: def convert_predict_rdd_to_dataframe(
      df, prediction_rdd, output_cols=None
  ):
      from pyspark.sql import Row
      from pyspark.ml.linalg import Vectors
  
      def convert_elem(elem):
          # list of np array
          if isinstance(elem, (list, tuple)):
              return [convert_elem(i) for i in elem]
          # dict of np array as values
          elif isinstance(elem, dict):
              return {k: convert_elem(v) for k, v in elem.items()}
          # scalar in basic type
          elif isinstance(elem, np.ScalarType):
              return float(elem)
          # np ndarray
          else:
              dim = len(elem.shape)
              if dim in [0, 1]:
                  # np 1-D array
                  return Vectors.dense(elem)
              else:
                  # multi-dimensional array
                  return elem.tolist()
  
      def combine(pair):
          if not output_cols:
              # a singleton list in pair[1] and stacked like [f1, f2] + [output1]
              if (
                  isinstance(pair[1], (list, tuple))
                  and len(pair[1]) == 1
              ):
                  return Row(
                      *(
                          [pair[0][col] for col in pair[0].__fields__]
                          + convert_elem(pair[1])
                      )
                  )
              else:
                  # a multiple list in pair[1] and stacked like [f1, f2] + [[output1], [output2]]
                  return Row(
                      *(
                          [pair[0][col] for col in pair[0].__fields__]
                          + [convert_elem(pair[1])]
                      )
                  )
          elif not isinstance(pair[1], (list, tuple)):
              # if pair[1] is not iterable, don't split them into list
              return Row(
                  *(
                      [pair[0][col] for col in pair[0].__fields__]
                      + [convert_elem(pair[1])]
                  )
              )
          else:
              # a multiple columns in pair[1] and merged like [f1, f2] + [output1, output2]
              return Row(
                  *(
                      [pair[0][col] for col in pair[0].__fields__]
                      + [convert_elem(item) for item in pair[1]]
                  )
              )
  
      combined_rdd = df.rdd.zip(prediction_rdd).map(combine)
      if output_cols is None:
          columns = df.columns + ["prediction"]
      else:
          columns = df.columns + output_cols
      # Converting to DataFrame will trigger the computation
      # to infer the schema of the prediction column.
      result_df = combined_rdd.toDF(columns)
      return result_df
  
imports: from contextlib import closing
  import socket
  import sys
  import tempfile
  import shutil
  from bigdl.dllib.utils import log4Error
  from bigdl.dllib.utils.common import callBigDlFunc
  from bigdl.dllib.utils.file_utils import get_file_list, is_local_path
  from bigdl.orca.data import SparkXShards
  from bigdl.orca.data.utils import get_size
  from bigdl.orca.data.file import put_local_dir_tree_to_remote, put_local_file_to_remote, get_remote_file_to_local, get_remote_dir_to_local
  from bigdl.dllib.utils.utils import convert_row_to_numpy
  from functools import partial
  import numpy
  from bigdl.orca.common import SafePickle
  import os
  import subprocess
  from bigdl.dllib.utils.log4Error import *
label_desc: Use image node_id 8 for calling a global function with example usage: convert_predict_rdd_to_dataframe(df, prediction_rdd, output_cols) and returns: result_df, unknown, unknown, Row, Row, Row, Row, float, Vectors, elem