code: def to_petastorm_fn(schema_cols, metadata):
      ARRAY = constants.ARRAY
      CUSTOM_SPARSE = constants.CUSTOM_SPARSE
  
      # Convert Spark Vectors into arrays so Petastorm can read them
      def to_petastorm(row):
          import numpy as np
          from pyspark import Row
  
          converted = {}
          for col in schema_cols:
              col_data = row[col]
              if isinstance(col_data, Vector):
                  intermediate_format = (
                      metadata[col]["intermediate_format"]
                      if metadata
                      else ARRAY
                  )
                  if intermediate_format == ARRAY:
                      converted[col] = col_data.toArray().tolist()
                  elif intermediate_format == CUSTOM_SPARSE:
                      # Currently petastorm does not support reading pyspark sparse vector. We put
                      # the indices and values into one array. when consuming the data, we re-create
                      # the vector from this format.
                      size = len(col_data.indices)
                      padding_zeros = 2 * (
                          metadata[col]["max_size"]
                          - len(col_data.indices)
                      )
  
                      converted[col] = np.concatenate(
                          (
                              np.array([size]),
                              col_data.indices,
                              col_data.values,
                              np.zeros(padding_zeros),
                          )
                      ).tolist()
  
          if converted:
              row = row.asDict().copy()
              row.update(converted)
          return Row(**row)
  
      return to_petastorm
  
imports: import horovod.spark.common._namedtuple_fix
  import contextlib
  import os
  import time
  from multiprocessing.pool import ThreadPool
  import pyarrow
  import numpy
  import pyspark.sql.functions
  from pyspark.ml.linalg import DenseVector, SparseVector, Vector, VectorUDT
  from pyspark.sql.types import ArrayType, BinaryType, BooleanType, FloatType, DoubleType, IntegerType, LongType, NullType, StringType
  from pyspark.sql import SparkSession
  from horovod.runner.common.util import codec, host_hash
  from horovod.spark.common import cache, constants
label_desc: Use image node_id 10 for calling a global function with example usage: to_petastorm_fn(schema_cols, metadata) and returns: to_petastorm, Row