code: def log_model(
      spark_model,
      artifact_path,
      conda_env=None,
      code_paths=None,
      dfs_tmpdir=None,
      sample_input=None,
      registered_model_name=None,
      signature: ModelSignature = None,
      input_example: ModelInputExample = None,
      await_registration_for=DEFAULT_AWAIT_MAX_SLEEP_SECONDS,
      pip_requirements=None,
      extra_pip_requirements=None,
      metadata=None,
  ):
      """
      Log a Spark MLlib model as an MLflow artifact for the current run. This uses the
      MLlib persistence format and produces an MLflow Model with the Spark flavor.
  
      Note: If no run is active, it will instantiate a run to obtain a run_id.
  
      :param spark_model: Spark model to be saved - MLflow can only save descendants of
                          pyspark.ml.Model or pyspark.ml.Transformer which implement
                          MLReadable and MLWritable.
      :param artifact_path: Run relative artifact path.
      :param conda_env: Either a dictionary representation of a Conda environment or the path to a
                        Conda environment yaml file. If provided, this decsribes the environment
                        this model should be run in. At minimum, it should specify the dependencies
                        contained in :func:`get_default_conda_env()`. If `None`, the default
                        :func:`get_default_conda_env()` environment is added to the model.
                        The following is an *example* dictionary representation of a Conda
                        environment::
  
                          {
                              'name': 'mlflow-env',
                              'channels': ['defaults'],
                              'dependencies': [
                                  'python=3.8.15',
                                  'pyspark=2.3.0'
                              ]
                          }
      :param dfs_tmpdir: Temporary directory path on Distributed (Hadoop) File System (DFS) or local
                         filesystem if running in local mode. The model is written in this
                         destination and then copied into the model's artifact directory. This is
                         necessary as Spark ML models read from and write to DFS if running on a
                         cluster. If this operation completes successfully, all temporary files
                         created on the DFS are removed. Defaults to ``/tmp/mlflow``.
                         For models defined in `pyspark.ml.connect` module, this param is ignored.
      :param sample_input: A sample input used to add the MLeap flavor to the model.
                           This must be a PySpark DataFrame that the model can evaluate. If
                           ``sample_input`` is ``None``, the MLeap flavor is not added.
      :param registered_model_name: If given, create a model version under
                                    ``registered_model_name``, also creating a registered model if one
                                    with the given name does not exist.
      :param signature: {{ signature }}
      :param input_example: {{ input_example }}
      :param await_registration_for: Number of seconds to wait for the model version to finish
                              being created and is in ``READY`` status. By default, the function
                              waits for five minutes. Specify 0 or None to skip waiting.
      :param pip_requirements: {{ pip_requirements }}
      :param extra_pip_requirements: {{ extra_pip_requirements }}
      :param metadata: Custom metadata dictionary passed to the model and stored in the MLmodel file.
  
                       .. Note:: Experimental: This parameter may change or be removed in a future
                                               release without warning.
      :return: A :py:class:`ModelInfo <mlflow.models.model.ModelInfo>` instance that contains the
               metadata of the logged model.
  
      .. code-block:: python
          :caption: Example
  
          from pyspark.ml import Pipeline
          from pyspark.ml.classification import LogisticRegression
          from pyspark.ml.feature import HashingTF, Tokenizer
  
          training = spark.createDataFrame(
              [
                  (0, "a b c d e spark", 1.0),
                  (1, "b d", 0.0),
                  (2, "spark f g h", 1.0),
                  (3, "hadoop mapreduce", 0.0),
              ],
              ["id", "text", "label"],
          )
          tokenizer = Tokenizer(inputCol="text", outputCol="words")
          hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="features")
          lr = LogisticRegression(maxIter=10, regParam=0.001)
          pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])
          model = pipeline.fit(training)
          mlflow.spark.log_model(model, "spark-model")
      """
      _validate_model(spark_model)
      from pyspark.ml import PipelineModel
  
      if _is_spark_connect_model(spark_model):
          return Model.log(
              artifact_path=artifact_path,
              flavor=mlflow.spark,
              spark_model=spark_model,
              conda_env=conda_env,
              code_paths=code_paths,
              sample_input=sample_input,
              registered_model_name=registered_model_name,
              signature=signature,
              input_example=input_example,
              await_registration_for=await_registration_for,
              pip_requirements=pip_requirements,
              extra_pip_requirements=extra_pip_requirements,
              metadata=metadata,
          )
  
      if not isinstance(spark_model, PipelineModel):
          spark_model = PipelineModel([spark_model])
      run_id = mlflow.tracking.fluent._get_or_start_run().info.run_id
      run_root_artifact_uri = mlflow.get_artifact_uri()
      remote_model_path = None
      if _should_use_mlflowdbfs(run_root_artifact_uri):
          remote_model_path = append_to_uri_path(
              run_root_artifact_uri,
              artifact_path,
              _SPARK_MODEL_PATH_SUB,
          )
          mlflowdbfs_path = _mlflowdbfs_path(run_id, artifact_path)
          with databricks_utils.MlflowCredentialContext(
              get_databricks_profile_uri_from_artifact_uri(
                  run_root_artifact_uri
              )
          ):
              try:
                  spark_model.save(mlflowdbfs_path)
              except Exception as e:
                  raise MlflowException(
                      "failed to save spark model via mlflowdbfs"
                  ) from e
  
      # If the artifact URI is a local filesystem path, defer to Model.log() to persist the model,
      # since Spark may not be able to write directly to the driver's filesystem. For example,
      # writing to `file:/uri` will write to the local filesystem from each executor, which will
      # be incorrect on multi-node clusters.
      # If the artifact URI is not a local filesystem path we attempt to write directly to the
      # artifact repo via Spark. If this fails, we defer to Model.log().
      elif is_local_uri(run_root_artifact_uri) or not _maybe_save_model(
          spark_model,
          append_to_uri_path(run_root_artifact_uri, artifact_path),
      ):
          return Model.log(
              artifact_path=artifact_path,
              flavor=mlflow.spark,
              spark_model=spark_model,
              conda_env=conda_env,
              code_paths=code_paths,
              dfs_tmpdir=dfs_tmpdir,
              sample_input=sample_input,
              registered_model_name=registered_model_name,
              signature=signature,
              input_example=input_example,
              await_registration_for=await_registration_for,
              pip_requirements=pip_requirements,
              extra_pip_requirements=extra_pip_requirements,
              metadata=metadata,
          )
      # Otherwise, override the default model log behavior and save model directly to artifact repo
      mlflow_model = Model(artifact_path=artifact_path, run_id=run_id)
      with TempDir() as tmp:
          tmp_model_metadata_dir = tmp.path()
          _save_model_metadata(
              tmp_model_metadata_dir,
              spark_model,
              mlflow_model,
              sample_input,
              conda_env,
              code_paths,
              signature=signature,
              input_example=input_example,
              pip_requirements=pip_requirements,
              extra_pip_requirements=extra_pip_requirements,
              remote_model_path=remote_model_path,
          )
          mlflow.tracking.fluent.log_artifacts(
              tmp_model_metadata_dir, artifact_path
          )
          mlflow.tracking.fluent._record_logged_model(mlflow_model)
          if registered_model_name is not None:
              mlflow.register_model(
                  f"runs:/{run_id}/{artifact_path}",
                  registered_model_name,
                  await_registration_for,
              )
          return mlflow_model.get_model_info()
  
imports: import logging
  import os
  import posixpath
  import re
  import shutil
  from typing import Any, Dict, Optional
  import pandas
  import yaml
  from packaging.version import Version
  import mlflow
  from mlflow import environment_variables, mleap, pyfunc
  from mlflow.environment_variables import MLFLOW_DFS_TMP
  from mlflow.exceptions import MlflowException
  from mlflow.models import Model, ModelInputExample, ModelSignature, infer_signature
  from mlflow.models.model import MLMODEL_FILE_NAME
  from mlflow.models.signature import _LOG_MODEL_INFER_SIGNATURE_WARNING_TEMPLATE
  from mlflow.models.utils import _Example, _save_example
  from mlflow.protos.databricks_pb2 import INVALID_PARAMETER_VALUE
  from mlflow.store.artifact.databricks_artifact_repo import DatabricksArtifactRepository
  from mlflow.tracking._model_registry import DEFAULT_AWAIT_MAX_SLEEP_SECONDS
  from mlflow.tracking.artifact_utils import _download_artifact_from_uri, _get_root_uri_and_artifact_path
  from mlflow.utils import _get_fully_qualified_class_name, databricks_utils
  from mlflow.utils.autologging_utils import autologging_integration, safe_patch
  from mlflow.utils.class_utils import _get_class_from_string
  from mlflow.utils.docstring_utils import LOG_MODEL_PARAM_DOCS, format_docstring
  from mlflow.utils.environment import _CONDA_ENV_FILE_NAME, _CONSTRAINTS_FILE_NAME, _PYTHON_ENV_FILE_NAME, _REQUIREMENTS_FILE_NAME, _mlflow_conda_env, _process_conda_env, _process_pip_requirements, _PythonEnv, _validate_env_arguments
  from mlflow.utils.file_utils import TempDir, get_total_file_size, shutil_copytree_without_file_permissions, write_to
  from mlflow.utils.model_utils import _add_code_from_conf_to_system_path, _get_flavor_configuration_from_uri, _validate_and_copy_code_paths
  from mlflow.utils.requirements_utils import _get_pinned_requirement
  from mlflow.utils.uri import append_to_uri_path, dbfs_hdfs_uri_to_fuse_path, generate_tmp_dfs_path, get_databricks_profile_uri_from_artifact_uri, is_databricks_acled_artifacts_uri, is_local_uri, is_valid_dbfs_uri
label_desc: Use image node_id 3 for calling a global function with example usage: log_model(spark_model, artifact_path, conda_env, code_paths, dfs_tmpdir, sample_input, registered_model_name, signature, input_example, await_registration_for, pip_requirements, extra_pip_requirements, metadata) and returns: Model, mlflow_model, Model