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,
      store_license=False,
  ):
      """
      Log a ``Johnsnowlabs NLUPipeline`` created via `nlp.load()
      <https://nlp.johnsnowlabs.com/docs/en/jsl/load_api>`_, as an MLflow artifact for the current
      run. This uses the MLlib persistence format and produces an MLflow Model with the
      ``johnsnowlabs`` flavor.
  
      Note: If no run is active, it will instantiate a run to obtain a run_id.
  
      Args:
          spark_model: NLUPipeline obtained via `nlp.load()
              <https://nlp.johnsnowlabs.com/docs/en/jsl/load_api>`_
          store_license: If True, the license will be stored with the model and used and re-loading
              it.
          artifact_path: Run relative artifact path.
          conda_env: Either a dictionary representation of a Conda environment or the path to a
              Conda environment yaml file. If provided, this describes 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',
                          'johnsnowlabs'
                      ]
                  }
          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``.
          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.
          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.
          signature: :py:class:`ModelSignature <mlflow.models.ModelSignature>`
              describes model input and output :py:class:`Schema <mlflow.types.Schema>`.
              The model signature can be :py:func:`inferred <mlflow.models.infer_signature>`
              from datasets with valid model input (e.g. the training dataset with target
              column omitted) and valid model output (e.g. model predictions generated on
              the training dataset), for example:
  
              .. code-block:: python
  
                  from mlflow.models.signature import infer_signature
  
                  train = df.drop_column("target_label")
                  predictions = ...  # compute model predictions
                  signature = infer_signature(train, predictions)
  
          input_example: {{ input_example }}
          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.
          pip_requirements: {{ pip_requirements }}
          extra_pip_requirements: {{ extra_pip_requirements }}
          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.
  
      Returns:
          A :py:class:`ModelInfo <mlflow.models.model.ModelInfo>` instance that contains the
          metadata of the logged model.
  
      .. code-block:: python
          :caption: Example
  
          import os
          import json
          import pandas as pd
          import mlflow
          from johnsnowlabs import nlp
  
          # Write your raw license.json string into the 'JOHNSNOWLABS_LICENSE_JSON' env variable
          creds = {
              "AWS_ACCESS_KEY_ID": "...",
              "AWS_SECRET_ACCESS_KEY": "...",
              "SPARK_NLP_LICENSE": "...",
              "SECRET": "...",
          }
          os.environ["JOHNSNOWLABS_LICENSE_JSON"] = json.dumps(creds)
  
          # Download & Install Jars/Wheels if missing and Start a spark Session
          nlp.start()
  
          # For more details on trainable models and parameterization like embedding choice see
          # https://nlp.johnsnowlabs.com/docs/en/jsl/training
          trainable_classifier = nlp.load("train.classifier")
  
          # Create a sample training dataset
          data = pd.DataFrame(
              {"text": ["I hate covid ", "I love covid"], "y": ["negative", "positive"]}
          )
  
          # Fit and get a trained classifier
          trained_classifier = trainable_classifier.fit(data)
          trained_classifier.predict("He hates covid")
  
          # Log it
          mlflow.johnsnowlabs.log_model(trained_classifier, "my_trained_model")
      """
  
      _validate_env_vars()
      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 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().
      if 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.johnsnowlabs,
              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,
              store_license=store_license,
          )
          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 json
  import logging
  import os
  import posixpath
  import shutil
  import sys
  from pathlib import Path
  from typing import Any, Dict, Optional
  import yaml
  import mlflow
  from mlflow import mleap, pyfunc
  from mlflow.environment_variables import MLFLOW_DFS_TMP
  from mlflow.models import Model
  from mlflow.models.model import MLMODEL_FILE_NAME
  from mlflow.models.signature import ModelSignature
  from mlflow.models.utils import ModelInputExample, _save_example
  from mlflow.spark import _HadoopFileSystem, _maybe_save_model, _mlflowdbfs_path, _should_use_mlflowdbfs
  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 databricks_utils
  from mlflow.utils.annotations import experimental
  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
  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_local_uri, is_valid_dbfs_uri
label_desc: Use image node_id 5 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, store_license) and returns: Model, mlflow_model