code: def save_model(
      spark_model,
      path,
      mlflow_model=None,
      conda_env=None,
      code_paths=None,
      dfs_tmpdir=None,
      sample_input=None,
      signature: ModelSignature = None,
      input_example: ModelInputExample = None,
      pip_requirements=None,
      extra_pip_requirements=None,
      metadata=None,
  ):
      """
      Save a Spark MLlib Model to a local path.
  
      By default, this function saves models using the Spark MLlib persistence mechanism.
      Additionally, if a sample input is specified using the ``sample_input`` parameter, the model
      is also serialized in MLeap format and the MLeap flavor is added.
  
      :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 path: Local path where the model is to be saved.
      :param mlflow_model: MLflow model config this flavor is being added to.
      :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 be written in this
                         destination and then copied to the requested local path. This is necessary
                         as Spark ML models read from and write to DFS if running on a cluster. All
                         temporary files created on the DFS are removed if this operation
                         completes successfully. Defaults to ``/tmp/mlflow``.
      :param sample_input: A sample input that is 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 signature: {{ signature }}
      :param input_example: {{ input_example }}
      :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.
  
      .. code-block:: python
          :caption: Example
  
          from mlflow import spark
          from pyspark.ml.pipeline import PipelineModel
  
          # your pyspark.ml.pipeline.PipelineModel type
          model = ...
          mlflow.spark.save_model(model, "spark-model")
      """
      _validate_model(spark_model)
      _validate_env_arguments(
          conda_env, pip_requirements, extra_pip_requirements
      )
  
      from pyspark.ml import PipelineModel
  
      from mlflow.utils._spark_utils import _get_active_spark_session
  
      is_spark_connect_model = _is_spark_connect_model(spark_model)
  
      if not is_spark_connect_model and not isinstance(
          spark_model, PipelineModel
      ):
          spark_model = PipelineModel([spark_model])
      if mlflow_model is None:
          mlflow_model = Model()
      if metadata is not None:
          mlflow_model.metadata = metadata
  
      # for automatic signature inference, we use an inline implementation rather than the
      # `_infer_signature_from_input_example` API because we need to convert model predictions from a
      # list into a Pandas series for signature inference.
      if signature is None and input_example is not None:
          input_ex = _Example(input_example).inference_data
          try:
              spark = _get_active_spark_session()
              if spark is not None:
                  wrapped_model = _PyFuncModelWrapper(
                      spark, spark_model
                  )
                  # We cast the predictions to a Pandas series because the Spark _PyFuncModelWrapper
                  # returns predictions as a list, which the `infer_signature` API does not support
                  # (unless it is a list of strings).
                  prediction = pd.Series(
                      wrapped_model.predict(input_ex)
                  )
                  signature = infer_signature(input_ex, prediction)
          except Exception as e:
              if environment_variables._MLFLOW_TESTING.get():
                  raise
              _logger.warning(
                  _LOG_MODEL_INFER_SIGNATURE_WARNING_TEMPLATE, repr(e)
              )
              _logger.debug("", exc_info=True)
      elif signature is False:
          signature = None
  
      sparkml_data_path = os.path.abspath(
          os.path.join(path, _SPARK_MODEL_PATH_SUB)
      )
  
      if is_spark_connect_model:
          spark_model.saveToLocal(sparkml_data_path)
      else:
          # Spark ML stores the model on DFS if running on a cluster
          # Save it to a DFS temp dir first and copy it to local path
          if dfs_tmpdir is None:
              dfs_tmpdir = MLFLOW_DFS_TMP.get()
          tmp_path = generate_tmp_dfs_path(dfs_tmpdir)
          spark_model.save(tmp_path)
          # We're copying the Spark model from DBFS to the local filesystem if (a) the temporary DFS
          # URI we saved the Spark model to is a DBFS URI ("dbfs:/my-directory"), or (b) if we're
          # running on a Databricks cluster and the URI is schemeless (e.g. looks like a filesystem
          # absolute path like "/my-directory")
          copying_from_dbfs = is_valid_dbfs_uri(tmp_path) or (
              databricks_utils.is_in_cluster()
              and posixpath.abspath(tmp_path) == tmp_path
          )
          if (
              copying_from_dbfs
              and databricks_utils.is_dbfs_fuse_available()
          ):
              tmp_path_fuse = dbfs_hdfs_uri_to_fuse_path(tmp_path)
              shutil.move(src=tmp_path_fuse, dst=sparkml_data_path)
          else:
              _HadoopFileSystem.copy_to_local_file(
                  tmp_path, sparkml_data_path, remove_src=True
              )
  
      _save_model_metadata(
          dst_dir=path,
          spark_model=spark_model,
          mlflow_model=mlflow_model,
          sample_input=sample_input,
          conda_env=conda_env,
          code_paths=code_paths,
          signature=signature,
          input_example=input_example,
          pip_requirements=pip_requirements,
          extra_pip_requirements=extra_pip_requirements,
      )
  
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 10 for calling a global function with example usage: save_model(spark_model, path, mlflow_model, conda_env, code_paths, dfs_tmpdir, sample_input, signature, input_example, pip_requirements, extra_pip_requirements, metadata) without return types