code: def _run(self, output_directory):
      import pandas as pd
      from pyspark.sql.functions import struct
  
      run_start_time = time.time()
  
      apply_recipe_tracking_config(self.tracking_config)
      if self.registry_uri:
          mlflow.set_registry_uri(self.registry_uri)
  
      # Get or create spark session
      try:
          spark = _get_active_spark_session()
          if spark:
              _logger.info("Found active spark session")
          else:
              _logger.info("Creating new spark session")
              spark = _create_local_spark_session_for_recipes()
      except Exception as e:
          raise MlflowException(
              message=(
                  "Encountered an error while getting or creating an active Spark session to"
                  " score dataset with spark UDF."
              ),
              error_code=BAD_REQUEST,
          ) from e
  
      # read cleaned dataset
      ingested_data_path = get_step_output_path(
          recipe_root_path=self.recipe_root,
          step_name="ingest_scoring",
          relative_path=_INPUT_FILE_NAME,
      )
      # Because the cached parquet file is not on DBFS, we have to first load it as a pandas df
      input_pdf = pd.read_parquet(ingested_data_path)
      input_sdf = spark.createDataFrame(input_pdf)
      if _PREDICTION_COLUMN_NAME in input_sdf.columns:
          _logger.warning(
              f"Input scoring dataframe already contains a column '{_PREDICTION_COLUMN_NAME}'. "
              f"This column will be dropped in favor of the predict output column name."
          )
  
      # get model uri
      model_uri = self.step_config["model_uri"]
      registered_model_file_path = get_step_output_path(
          recipe_root_path=self.recipe_root,
          step_name="register",
          relative_path=_REGISTERED_MV_INFO_FILE,
      )
      if os.path.exists(registered_model_file_path):
          rmi = RegisteredModelVersionInfo.from_json(
              path=registered_model_file_path
          )
          model_uri = f"models:/{rmi.name}/{rmi.version}"
  
      # scored dataset
      result_type = self.step_config.get("result_type", "double")
      predict = mlflow.pyfunc.spark_udf(
          spark,
          model_uri,
          result_type=result_type,
          env_manager=_ENV_MANAGER,
      )
      scored_sdf = input_sdf.withColumn(
          _PREDICTION_COLUMN_NAME, predict(struct(*input_sdf.columns))
      )
  
      # check if output location is already populated for non-delta output formats
      output_format = self.step_config["using"]
      output_location = self.step_config["location"]
      output_populated = False
      if self.save_mode in ["default", "error", "errorifexists"]:
          if output_format == "parquet" or output_format == "delta":
              output_populated = os.path.exists(output_location)
          else:
              try:
                  output_populated = (
                      spark._jsparkSession.catalog().tableExists(
                          output_location
                      )
                  )
              except Exception:
                  # swallow spark failures
                  pass
      if output_populated:
          raise MlflowException(
              message=(
                  f"Output location `{output_location}` using format `{output_format}` is "
                  "already populated. To overwrite, please change the spark `save_mode` in "
                  "the predict step configuration."
              ),
              error_code=BAD_REQUEST,
          )
  
      if output_format == "table":
          try:
              from delta.tables import DeltaTable
  
              output_populated = DeltaTable.forName(
                  spark, output_location
              )
          except Exception:
              # swallow spark failures
              pass
  
          if output_populated:
              _logger.info(f"Table already exists at {output_location}")
              # If the table already exists, we are just setting up the table properties to
              # ensure that the table can be written with column names with spaces.
              spark.sql(
                  f"ALTER TABLE {output_location} SET TBLPROPERTIES "
                  "('delta.columnMapping.mode'='name','delta.minReaderVersion'='2',"
                  "'delta.minWriterVersion'='5')"
              )
          else:
              _logger.info(f"Creating a new table at {output_location}")
              from delta.tables import DeltaTable
  
              # If the table location specified doesn't exist, we are creating a new table
              # with properties required to ensure that column names can have spaces.
              DeltaTable.create().addColumns(
                  scored_sdf.schema
              ).property("delta.minReaderVersion", "2").property(
                  "delta.minWriterVersion", "5"
              ).property(
                  "delta.columnMapping.mode", "name"
              ).tableName(
                  output_location
              ).execute()
              # We are overriding the save_mode to append for the create case, since the table
              # is already created above, so adding any record to the empty table can be
              # appended to the table
              self.save_mode = "append"
  
      # save predictions
      if output_format in ["parquet", "delta"]:
          scored_sdf.coalesce(1).write.format(output_format).mode(
              self.save_mode
          ).save(output_location)
      else:
          scored_sdf.write.format("delta").mode(
              self.save_mode
          ).saveAsTable(output_location)
  
      # predict step artifacts
      write_spark_dataframe_to_parquet_on_local_disk(
          scored_sdf,
          os.path.join(output_directory, _SCORED_OUTPUT_FILE_NAME),
      )
  
      self.run_end_time = time.time()
      self.execution_duration = self.run_end_time - run_start_time
      return self._build_profiles_and_card(scored_sdf)
  
imports: import logging
  import os
  import time
  from typing import Any, Dict
  import mlflow
  from mlflow.exceptions import BAD_REQUEST, INVALID_PARAMETER_VALUE, MlflowException
  from mlflow.recipes.artifacts import DataframeArtifact, RegisteredModelVersionInfo
  from mlflow.recipes.cards import BaseCard
  from mlflow.recipes.step import BaseStep, StepClass
  from mlflow.recipes.steps.register import _REGISTERED_MV_INFO_FILE
  from mlflow.recipes.utils.execution import get_step_output_path
  from mlflow.recipes.utils.step import get_pandas_data_profiles
  from mlflow.recipes.utils.tracking import TrackingConfig, apply_recipe_tracking_config, get_recipe_tracking_config
  from mlflow.utils._spark_utils import _create_local_spark_session_for_recipes, _get_active_spark_session
  from mlflow.utils.databricks_utils import get_databricks_env_vars
  from mlflow.utils.file_utils import write_spark_dataframe_to_parquet_on_local_disk
label_desc: Use image node_id 4 for calling the PredictStep obj's underlying member method code with example usage: obj._run(output_directory) and returns: self