code: def _run(
      uri,
      experiment_id,
      entry_point,
      version,
      parameters,
      docker_args,
      backend_name,
      backend_config,
      storage_dir,
      env_manager,
      synchronous,
      run_name,
      build_image,
      docker_auth,
  ):
      """
      Helper that delegates to the project-running method corresponding to the passed-in backend.
      Returns a ``SubmittedRun`` corresponding to the project run.
      """
      tracking_store_uri = tracking.get_tracking_uri()
      backend_config[PROJECT_ENV_MANAGER] = env_manager
      backend_config[PROJECT_SYNCHRONOUS] = synchronous
      backend_config[PROJECT_DOCKER_ARGS] = docker_args
      backend_config[PROJECT_STORAGE_DIR] = storage_dir
      backend_config[PROJECT_BUILD_IMAGE] = build_image
      backend_config[PROJECT_DOCKER_AUTH] = docker_auth
      # TODO: remove this check once kubernetes execution has been refactored
      if backend_name not in {"databricks", "kubernetes"}:
          backend = loader.load_backend(backend_name)
          if backend:
              submitted_run = backend.run(
                  uri,
                  entry_point,
                  parameters,
                  version,
                  backend_config,
                  tracking_store_uri,
                  experiment_id,
              )
              tracking.MlflowClient().set_tag(
                  submitted_run.run_id,
                  MLFLOW_PROJECT_BACKEND,
                  backend_name,
              )
              if run_name is not None:
                  tracking.MlflowClient().set_tag(
                      submitted_run.run_id, MLFLOW_RUN_NAME, run_name
                  )
              return submitted_run
  
      work_dir = fetch_and_validate_project(
          uri, version, entry_point, parameters
      )
      project = load_project(work_dir)
      _validate_execution_environment(project, backend_name)
  
      active_run = get_or_create_run(
          None,
          uri,
          experiment_id,
          work_dir,
          version,
          entry_point,
          parameters,
      )
  
      if run_name is not None:
          tracking.MlflowClient().set_tag(
              active_run.info.run_id, MLFLOW_RUN_NAME, run_name
          )
  
      if backend_name == "databricks":
          tracking.MlflowClient().set_tag(
              active_run.info.run_id,
              MLFLOW_PROJECT_BACKEND,
              "databricks",
          )
          from mlflow.projects.databricks import run_databricks
  
          return run_databricks(
              remote_run=active_run,
              uri=uri,
              entry_point=entry_point,
              work_dir=work_dir,
              parameters=parameters,
              experiment_id=experiment_id,
              cluster_spec=backend_config,
              env_manager=env_manager,
          )
  
      elif backend_name == "kubernetes":
          from mlflow.projects import kubernetes as kb
          from mlflow.projects.docker import (
              build_docker_image,
              validate_docker_env,
              validate_docker_installation,
          )
  
          tracking.MlflowClient().set_tag(
              active_run.info.run_id, MLFLOW_PROJECT_ENV, "docker"
          )
          tracking.MlflowClient().set_tag(
              active_run.info.run_id,
              MLFLOW_PROJECT_BACKEND,
              "kubernetes",
          )
          validate_docker_env(project)
          validate_docker_installation()
          kube_config = _parse_kubernetes_config(backend_config)
          image = build_docker_image(
              work_dir=work_dir,
              repository_uri=kube_config["repository-uri"],
              base_image=project.docker_env.get("image"),
              run_id=active_run.info.run_id,
              build_image=build_image,
              docker_auth=docker_auth,
          )
          image_digest = kb.push_image_to_registry(image.tags[0])
          tracking.MlflowClient().set_tag(
              active_run.info.run_id,
              MLFLOW_DOCKER_IMAGE_ID,
              image_digest,
          )
          return kb.run_kubernetes_job(
              project.name,
              active_run,
              image.tags[0],
              image_digest,
              get_entry_point_command(
                  project, entry_point, parameters, storage_dir
              ),
              get_run_env_vars(
                  run_id=active_run.info.run_uuid,
                  experiment_id=active_run.info.experiment_id,
              ),
              kube_config.get("kube-context", None),
              kube_config["kube-job-template"],
          )
  
      supported_backends = ["databricks", "kubernetes"] + list(
          loader.MLFLOW_BACKENDS.keys()
      )
      raise ExecutionException(
          f"Got unsupported execution mode {backend_name}. Supported values: {supported_backends}"
      )
  
imports: import json
  import logging
  import os
  import yaml
  import mlflow.projects.databricks
  import mlflow.utils.uri
  from mlflow import tracking
  from mlflow.entities import RunStatus
  from mlflow.exceptions import ExecutionException, MlflowException
  from mlflow.projects.backend import loader
  from mlflow.projects.submitted_run import SubmittedRun
  from mlflow.projects.utils import MLFLOW_LOCAL_BACKEND_RUN_ID_CONFIG, PROJECT_BUILD_IMAGE, PROJECT_DOCKER_ARGS, PROJECT_DOCKER_AUTH, PROJECT_ENV_MANAGER, PROJECT_STORAGE_DIR, PROJECT_SYNCHRONOUS, fetch_and_validate_project, get_entry_point_command, get_or_create_run, get_run_env_vars, load_project
  from mlflow.tracking.fluent import _get_experiment_id
  from mlflow.utils import env_manager
  from mlflow.utils.mlflow_tags import MLFLOW_DOCKER_IMAGE_ID, MLFLOW_PROJECT_BACKEND, MLFLOW_PROJECT_ENV, MLFLOW_RUN_NAME
label_desc: Use image node_id 2 for calling a global function with example usage: _run(uri, experiment_id, entry_point, version, parameters, docker_args, backend_name, backend_config, storage_dir, env_manager, synchronous, run_name, build_image, docker_auth) and returns: run_databricks, submitted_run, kb