code: def score_model(spark, data_path, model_uri):
      if os.path.isdir(data_path):
          filenames = [
              os.path.abspath(os.path.join(data_path, x))
              for x in os.listdir(data_path)
              if os.path.isfile(os.path.join(data_path, x))
          ]
      else:
          filenames = [data_path]
  
      image_classifier_udf = mlflow.pyfunc.spark_udf(
          spark=spark,
          model_uri=model_uri,
          result_type=ArrayType(StringType()),
      )
  
      image_df = read_images(spark, filenames)
  
      raw_preds = (
          image_df.withColumn(
              "prediction", image_classifier_udf("image")
          )
          .select(["filename", "prediction"])
          .toPandas()
      )
      # load the pyfunc model to get our domain
      pyfunc_model = mlflow.pyfunc.load_model(model_uri=model_uri)
      preds = pd.DataFrame(raw_preds["filename"], index=raw_preds.index)
      preds[pyfunc_model._column_names] = pd.DataFrame(
          raw_preds["prediction"].values.tolist(),
          columns=pyfunc_model._column_names,
          index=raw_preds.index,
      )
  
      preds = pd.DataFrame(raw_preds["filename"], index=raw_preds.index)
  
      preds[pyfunc_model._column_names] = pd.DataFrame(
          raw_preds["prediction"].values.tolist(),
          columns=pyfunc_model._column_names,
          index=raw_preds.index,
      )
      return preds.to_json(orient="records")
  
file_path: mlflow/examples/flower_classifier/score_images_spark.py
imports: import base64
  import os
  import click
  import pandas
  import pyspark
  from pyspark.sql.types import ArrayType, Row, StringType, StructField, StructType
  import mlflow
  import mlflow.pyfunc
  from mlflow.utils import cli_args
label_desc: Use image node_id 3 for calling a global function with example usage: score_model(spark, data_path, model_uri) and returns: preds