Repository: mlflow/mlflow

Issue:
[FR] Currently unable to specify variant names when deploying models to Sagemaker
### Willingness to contribute

Yes. I would be willing to contribute this feature with guidance from the MLflow community.

### Proposal Summary

I would like to add the ability to pass in a specific name for the model variant when deploying to Sagmaker as an additional argument to the deploy method in sagemaker/__init__.py.  Currently, variant names are autogenerated by _get_sagemaker_model_name using the endpoint name, model name and uid.  

### Motivation

> #### What is the use case for this feature?
To supply a specific name to each model variant hosted on Sagemaker.

> #### Why is this use case valuable to support for MLflow users in general?
The name would be more meaningful than the current default.

> #### Why is this use case valuable to support for your project(s) or organization?
This would be especially important for endpoints that host multiple models or different version of the same model (ex: for testing purposes).  It would make it much easier for the user to better understand the purpose of a given model based on its name and distinguish between models.  

> #### Why is it currently difficult to achieve this use case?
Currently, there is not way to specify the variant name when deploying, they are just generated automatically


### Details

I propose adding an argument to the deploy method where a specific name can be entered.  This will throw an error if that name currently exists or if it violates any naming conventions.  Use the current default method if no name is specified when calling this method.

### What component(s) does this bug affect?

- [ ] `area/artifacts`: Artifact stores and artifact logging
- [ ] `area/build`: Build and test infrastructure for MLflow
- [ ] `area/docs`: MLflow documentation pages
- [ ] `area/examples`: Example code
- [ ] `area/model-registry`: Model Registry service, APIs, and the fluent client calls for Model Registry
- [ ] `area/models`: MLmodel format, model serialization/deserialization, flavors
- [ ] `area/pipelines`: Pipelines, Pipeline APIs, Pipeline configs, Pipeline Templates
- [ ] `area/projects`: MLproject format, project running backends
- [X] `area/scoring`: MLflow Model server, model deployment tools, Spark UDFs
- [ ] `area/server-infra`: MLflow Tracking server backend
- [ ] `area/tracking`: Tracking Service, tracking client APIs, autologging

### What interface(s) does this bug affect?

- [ ] `area/uiux`: Front-end, user experience, plotting, JavaScript, JavaScript dev server
- [ ] `area/docker`: Docker use across MLflow's components, such as MLflow Projects and MLflow Models
- [ ] `area/sqlalchemy`: Use of SQLAlchemy in the Tracking Service or Model Registry
- [ ] `area/windows`: Windows support

### What language(s) does this bug affect?

- [ ] `language/r`: R APIs and clients
- [ ] `language/java`: Java APIs and clients
- [ ] `language/new`: Proposals for new client languages

### What integration(s) does this bug affect?

- [ ] `integrations/azure`: Azure and Azure ML integrations
- [X] `integrations/sagemaker`: SageMaker integrations
- [ ] `integrations/databricks`: Databricks integrations

Discussion:
github-actions[bot]: @nfarley-soaren Thanks for the contribution! The DCO check failed. Please sign off your commits by following the instructions here: https://github.com/mlflow/mlflow/runs/7869201785. See https://github.com/mlflow/mlflow/blob/master/CONTRIBUTING.rst#sign-your-work for more details.
nfarley-soaren: I'm not sure where to update the docs with examples.

Patch:
diff --git a/docs/source/models.rst b/docs/source/models.rst
index 1ee817b3f..20b1d9d9a 100644
--- a/docs/source/models.rst
+++ b/docs/source/models.rst
@@ -30,7 +30,8 @@ function" flavor that describes how to run the model as a Python function. Howev
 also define and use other flavors. For example, MLflow's :py:mod:`mlflow.sklearn` library allows
 loading models back as a scikit-learn ``Pipeline`` object for use in code that is aware of
 scikit-learn, or as a generic Python function for use in tools that just need to apply the model
-(for example, the ``mlflow sagemaker`` tool for deploying models to Amazon SageMaker).
+(for example, the ``mlflow deployments`` tool with the option ``-t sagemaker`` for deploying models
+to Amazon SageMaker).
 
 All of the flavors that a particular model supports are defined in its ``MLmodel`` file in YAML
 format. For example, :py:mod:`mlflow.sklearn` outputs models as follows:
@@ -67,12 +68,12 @@ can serve a model with the ``python_function`` or the ``crate`` (R Function) fla
 
     mlflow models serve -m my_model
 
-In addition, the ``mlflow sagemaker`` command-line tool can package and deploy models to AWS
+In addition, the ``mlflow deployments`` command-line tool can package and deploy models to AWS
 SageMaker as long as they support the ``python_function`` flavor:
 
 .. code-block:: bash
 
-    mlflow sagemaker deploy -m my_model [other options]
+    mlflow deployments create -t sagemaker -m my_model [other options]
 
 Fields in the MLmodel Format
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -1920,12 +1921,14 @@ For more info, see:
 Deploy a ``python_function`` model on Amazon SageMaker
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-The :py:mod:`mlflow.sagemaker` module can deploy ``python_function`` models locally in a Docker
-container with SageMaker compatible environment and remotely on SageMaker.
-To deploy remotely to SageMaker you need to set up your environment and user accounts.
-To export a custom model to SageMaker, you need a MLflow-compatible Docker image to be available on Amazon ECR.
-MLflow provides a default Docker image definition; however, it is up to you to build the image and upload it to ECR.
-MLflow includes the utility function ``build_and_push_container`` to perform this step. Once built and uploaded, you can use the MLflow container for all MLflow Models. Model webservers deployed using the :py:mod:`mlflow.sagemaker`
+The :py:mod:`mlflow.deployments` and :py:mod:`mlflow.sagemaker` modules can deploy
+``python_function`` models locally in a Docker container with SageMaker compatible environment and
+remotely on SageMaker. To deploy remotely to SageMaker you need to set up your environment and user
+accounts. To export a custom model to SageMaker, you need a MLflow-compatible Docker image to be
+available on Amazon ECR. MLflow provides a default Docker image definition; however, it is up to you
+to build the image and upload it to ECR. MLflow includes the utility function
+``build_and_push_container`` to perform this step. Once built and uploaded, you can use the MLflow
+container for all MLflow Models. Model webservers deployed using the :py:mod:`mlflow.deployments`
 module accept the following data formats as input, depending on the deployment flavor:
 
 * ``python_function``: For this deployment flavor, the endpoint accepts the same formats described
@@ -1939,25 +1942,26 @@ module accept the following data formats as input, depending on the deployment f
 Commands
 ~~~~~~~~~
 
-* :py:func:`run-local <mlflow.sagemaker.run_local>` deploys the model locally in a Docker
-  container. The image and the environment should be identical to how the model would be run
+* :py:func:`mlflow sagemaker run-local <mlflow.sagemaker.run_local>` deploys the model locally in a
+  Docker container. The image and the environment should be identical to how the model would be run
   remotely and it is therefore useful for testing the model prior to deployment.
 
-* `build-and-push-container <cli.html#mlflow-sagemaker-build-and-push-container>`_ builds an MLfLow
-  Docker image and uploads it to ECR. The caller must have the correct permissions set up. The image
-  is built locally and requires Docker to be present on the machine that performs this step.
+* `mlflow sagemaker build-and-push-container <cli.html#mlflow-sagemaker-build-and-push-container>`_
+  builds an MLfLow Docker image and uploads it to ECR. The caller must have the correct permissions
+  set up. The image is built locally and requires Docker to be present on the machine that performs
+  this step.
 
-* :py:func:`deploy <mlflow.sagemaker.deploy>` deploys the model on Amazon SageMaker. MLflow
-  uploads the Python Function model into S3 and starts an Amazon SageMaker endpoint serving
-  the model.
+* :py:func:`mlflow deployments create -t sagemaker <mlflow.sagemaker.SageMakerDeploymentClient.create_deployment>`
+  deploys the model on Amazon SageMaker. MLflow uploads the Python Function model into S3 and starts
+  an Amazon SageMaker endpoint serving the model.
 
 .. rubric:: Example workflow using the MLflow CLI
 
 .. code-block:: bash
 
-    mlflow sagemaker build-and-push-container  - build the container (only needs to be called once)
-    mlflow sagemaker run-local -m <path-to-model>  - test the model locally
-    mlflow sagemaker deploy <parameters> - deploy the model remotely
+    mlflow sagemaker build-and-push-container  # build the container (only needs to be called once)
+    mlflow sagemaker run-local -m <path-to-model>  # test the model locally
+    mlflow deployments sagemaker create -t  # deploy the model remotely
 
 
 For more info, see:
@@ -1965,10 +1969,8 @@ For more info, see:
 .. code-block:: bash
 
     mlflow sagemaker --help
-    mlflow sagemaker build-and-push-container --help
     mlflow sagemaker run-local --help
-    mlflow sagemaker deploy --help
-
+    mlflow deployments help -t sagemaker
 
 Export a ``python_function`` model as an Apache Spark UDF
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
diff --git a/mlflow/models/__init__.py b/mlflow/models/__init__.py
index 550c397b5..8d8b1b0f8 100644
--- a/mlflow/models/__init__.py
+++ b/mlflow/models/__init__.py
@@ -21,7 +21,7 @@ The built-in flavors are:
 For details, see `MLflow Models <../models.html>`_.
 """
 
-from .model import Model
+from .model import Model, get_model_info
 from .flavor_backend import FlavorBackend
 from ..utils.environment import infer_pip_requirements
 from .evaluation import (
@@ -39,6 +39,7 @@ __all__ = [
     "evaluate",
     "EvaluationArtifact",
     "EvaluationResult",
+    "get_model_info",
     "list_evaluators",
     "MetricThreshold",
 ]
@@ -48,12 +49,13 @@ __all__ = [
 # because of lack of numpy/pandas library, so wrap them with try...except block
 try:
     from .signature import ModelSignature, infer_signature  # pylint: disable=unused-import
-    from .utils import ModelInputExample  # pylint: disable=unused-import
+    from .utils import ModelInputExample, validate_schema  # pylint: disable=unused-import
 
     __all__ += [
         "ModelSignature",
         "ModelInputExample",
         "infer_signature",
+        "validate_schema",
     ]
 except ImportError:
     pass
diff --git a/mlflow/models/model.py b/mlflow/models/model.py
index 30c127cb8..b1d98ef3b 100644
--- a/mlflow/models/model.py
+++ b/mlflow/models/model.py
@@ -1,18 +1,19 @@
 from datetime import datetime
 import json
 import logging
+import warnings
 
 import yaml
 import os
 import uuid
 
-from typing import Any, Dict, Optional, Union, Callable, NamedTuple
+from typing import Any, Dict, Optional, Union, Callable
 
 import mlflow
 from mlflow.exceptions import MlflowException
+from mlflow.tracking._model_registry import DEFAULT_AWAIT_MAX_SLEEP_SECONDS
 from mlflow.utils.file_utils import TempDir
 from mlflow.utils.databricks_utils import get_databricks_runtime
-from mlflow.tracking._model_registry import DEFAULT_AWAIT_MAX_SLEEP_SECONDS
 
 _logger = logging.getLogger(__name__)
 
@@ -29,55 +30,131 @@ _LOG_MODEL_METADATA_WARNING_TEMPLATE = (
 _MLFLOW_VERSION_KEY = "mlflow_version"
 
 
-class ModelInfo(NamedTuple):
+class ModelInfo:
     """
     The metadata of a logged MLflow Model.
     """
 
-    #: Run relative path identifying the logged model.
-    artifact_path: str
+    def __init__(
+        self,
+        artifact_path: str,
+        flavors: Dict[str, Any],
+        model_uri: str,
+        model_uuid: str,
+        run_id: str,
+        saved_input_example_info: Optional[Dict[str, Any]],
+        signature,  # Optional[ModelSignature]
+        utc_time_created: str,
+        mlflow_version: str,
+        signature_dict: Optional[Dict[str, Any]] = None,
+    ):
+        self._artifact_path = artifact_path
+        self._flavors = flavors
+        self._model_uri = model_uri
+        self._model_uuid = model_uuid
+        self._run_id = run_id
+        self._saved_input_example_info = saved_input_example_info
+        self._signature_dict = signature_dict
+        self._signature = signature
+        self._utc_time_created = utc_time_created
+        self._mlflow_version = mlflow_version
 
-    #: A dictionary mapping the flavor name to how to serve the model as that flavor. For example:
-    #:
-    #: .. code-block:: python
-    #:
-    #:     {
-    #:         "python_function": {
-    #:             "model_path": "model.pkl",
-    #:             "loader_module": "mlflow.sklearn",
-    #:             "python_version": "3.8.10",
-    #:             "env": "conda.yaml",
-    #:         },
-    #:         "sklearn": {
-    #:             "pickled_model": "model.pkl",
-    #:             "sklearn_version": "0.24.1",
-    #:             "serialization_format": "cloudpickle",
-    #:         },
-    #:     }
-    flavors: Dict[str, Any]
+    @property
+    def artifact_path(self):
+        """
+        Run relative path identifying the logged model.
+        """
+        return self._artifact_path
 
-    #: The ``model_uri`` of the logged model in the format ``'runs:/<run_id>/<artifact_path>'``.
-    model_uri: str
+    @property
+    def flavors(self):
+        """
+        A dictionary mapping the flavor name to how to serve
+        the model as that flavor. For example:
+
+            .. code-block:: python
+
+                {
+                    "python_function": {
+                        "model_path": "model.pkl",
+                        "loader_module": "mlflow.sklearn",
+                        "python_version": "3.8.10",
+                        "env": "conda.yaml",
+                    },
+                    "sklearn": {
+                        "pickled_model": "model.pkl",
+                        "sklearn_version": "0.24.1",
+                        "serialization_format": "cloudpickle",
+                    },
+                }
+        """
+        return self._flavors
 
-    #: The ``model_uuid`` of the logged model, e.g., ``'39ca11813cfc46b09ab83972740b80ca'``.
-    model_uuid: str
+    @property
+    def model_uri(self):
+        """
+        The ``model_uri`` of the logged model in the format ``'runs:/<run_id>/<artifact_path>'``.
+        """
+        return self._model_uri
 
-    #: The ``run_id`` associated with the logged model, e.g., ``'8ede7df408dd42ed9fc39019ef7df309'``
-    run_id: str
+    @property
+    def model_uuid(self):
+        """
+        The ``model_uuid`` of the logged model,
+        e.g., ``'39ca11813cfc46b09ab83972740b80ca'``.
+        """
+        return self._model_uuid
 
-    #: A dictionary that contains the metadata of the saved input example, e.g.,
-    #: ``{"artifact_path": "input_example.json", "type": "dataframe", "pandas_orient": "split"}``.
-    saved_input_example_info: Optional[Dict[str, Any]]
+    @property
+    def run_id(self):
+        """
+        The ``run_id`` associated with the logged model,
+        e.g., ``'8ede7df408dd42ed9fc39019ef7df309'``
+        """
+        return self._run_id
 
-    #: A dictionary that describes the model input and output generated by
-    #: :py:meth:`ModelSignature.to_dict() <mlflow.models.ModelSignature.to_dict>`.
-    signature_dict: Optional[Dict[str, Any]]
+    @property
+    def saved_input_example_info(self):
+        """
+        A dictionary that contains the metadata of the saved input example, e.g.,
+        ``{"artifact_path": "input_example.json", "type": "dataframe", "pandas_orient": "split"}``.
+        """
+        return self._saved_input_example_info
 
-    #: The UTC time that the logged model is created, e.g., ``'2022-01-12 05:17:31.634689'``.
-    utc_time_created: str
+    @property
+    def signature_dict(self):
+        """
+        A dictionary that describes the model input and output generated by
+        :py:meth:`ModelSignature.to_dict() <mlflow.models.ModelSignature.to_dict>`.
+        """
+        warnings.warn(
+            "Field signature_dict is deprecated since v1.28.1. Use signature instead.",
+            category=FutureWarning,
+            stacklevel=2,
+        )
+        return self._signature_dict
 
-    #: Version of MLFlow used to log the model
-    mlflow_version: str
+    @property
+    def signature(self):  # -> Optional[ModelSignature]
+        """
+        A :py:class:`ModelSignature <mlflow.models.ModelSignature>` that describes the
+        model input and output.
+        """
+        return self._signature
+
+    @property
+    def utc_time_created(self):
+        """
+        The UTC time that the logged model is created, e.g., ``'2022-01-12 05:17:31.634689'``.
+        """
+        return self._utc_time_created
+
+    @property
+    def mlflow_version(self):
+        """
+        Version of MLflow used to log the model
+        """
+        return self._mlflow_version
 
 
 class Model:
@@ -180,6 +257,7 @@ class Model:
             run_id=self.run_id,
             saved_input_example_info=self.saved_input_example_info,
             signature_dict=self.signature.to_dict() if self.signature else None,
+            signature=self.signature,
             utc_time_created=self.utc_time_created,
             mlflow_version=self.mlflow_version,
         )
@@ -307,3 +385,64 @@ class Model:
                     await_registration_for=await_registration_for,
                 )
         return mlflow_model.get_model_info()
+
+
+def get_model_info(model_uri: str) -> ModelInfo:
+    """
+    Get metadata for the specified model, such as its input/output signature.
+
+    :param model_uri: The location, in URI format, of the MLflow model. For example:
+
+                      - ``/Users/me/path/to/local/model``
+                      - ``relative/path/to/local/model``
+                      - ``s3://my_bucket/path/to/model``
+                      - ``runs:/<mlflow_run_id>/run-relative/path/to/model``
+                      - ``models:/<model_name>/<model_version>``
+                      - ``models:/<model_name>/<stage>``
+                      - ``mlflow-artifacts:/path/to/model``
+
+                      For more information about supported URI schemes, see
+                      `Referencing Artifacts <https://www.mlflow.org/docs/latest/concepts.html#
+                      artifact-locations>`_.
+
+    :return: A :py:class:`ModelInfo <mlflow.models.model.ModelInfo>` instance that contains the
+            metadata of the logged model.
+
+    .. code-block:: python
+            :caption: Example usage of get_model_info
+
+            import mlflow.models
+            import mlflow.sklearn
+            from sklearn.ensemble import RandomForestRegressor
+
+            with mlflow.start_run() as run:
+                params = {"n_estimators": 3, "random_state": 42}
+                X, y = [[0, 1]], [1]
+                signature = mlflow.models.infer_signature(X, y)
+                rfr = RandomForestRegressor(**params).fit(X, y)
+                mlflow.log_params(params)
+                mlflow.sklearn.log_model(rfr, artifact_path="sklearn-model", signature=signature)
+
+            model_uri = "runs:/{}/sklearn-model".format(run.info.run_id)
+            # Get model info with model_uri
+            model_info = mlflow.models.get_model_info(model_uri)
+            # Get model signature directly
+            model_signature = model_info.signature
+            assert model_signature == signature
+    """
+    from mlflow.pyfunc import _download_artifact_from_uri
+
+    local_path = _download_artifact_from_uri(artifact_uri=model_uri, output_path=None)
+    model_meta = Model.load(os.path.join(local_path, MLMODEL_FILE_NAME))
+    return ModelInfo(
+        artifact_path=model_meta.artifact_path,
+        flavors=model_meta.flavors,
+        model_uri=model_uri,
+        model_uuid=model_meta.model_uuid,
+        run_id=model_meta.run_id,
+        saved_input_example_info=model_meta.saved_input_example_info,
+        signature_dict=model_meta.signature.to_dict() if model_meta.signature else None,
+        signature=model_meta.signature,
+        utc_time_created=model_meta.utc_time_created,
+        mlflow_version=model_meta.mlflow_version,
+    )
diff --git a/mlflow/models/utils.py b/mlflow/models/utils.py
index bae0cf02a..6f61d0e68 100644
--- a/mlflow/models/utils.py
+++ b/mlflow/models/utils.py
@@ -1,13 +1,14 @@
 import json
 import os
-from typing import Union
+from typing import Union, Any, Dict, List
 
 import numpy as np
 import pandas as pd
 
 from mlflow.exceptions import MlflowException
 from mlflow.models import Model
-from mlflow.types.utils import TensorsNotSupportedException
+from mlflow.types import DataType, Schema, TensorSpec
+from mlflow.types.utils import TensorsNotSupportedException, clean_tensor_type
 from mlflow.utils.proto_json_utils import NumpyEncoder, _dataframe_from_json, parse_tf_serving_input
 
 try:
@@ -20,6 +21,17 @@ except ImportError:
 
 ModelInputExample = Union[pd.DataFrame, np.ndarray, dict, list, "csr_matrix", "csc_matrix"]
 
+PyFuncInput = Union[
+    pd.DataFrame,
+    pd.Series,
+    np.ndarray,
+    "csc_matrix",
+    "csr_matrix",
+    List[Any],
+    Dict[str, Any],
+]
+PyFuncOutput = Union[pd.DataFrame, pd.Series, np.ndarray, list]
+
 
 class _Example:
     """
@@ -264,3 +276,300 @@ def plot_lines(data_series, xlabel, ylabel, legend_loc=None, line_kwargs=None):
     ax.set(xlabel=xlabel, ylabel=ylabel)
 
     return fig, ax
+
+
+def _enforce_tensor_spec(
+    values: Union[np.ndarray, "csc_matrix", "csr_matrix"],
+    tensor_spec: TensorSpec,
+):
+    """
+    Enforce the input tensor shape and type matches the provided tensor spec.
+    """
+    expected_shape = tensor_spec.shape
+    actual_shape = values.shape
+
+    actual_type = values.dtype if isinstance(values, np.ndarray) else values.data.dtype
+
+    if len(expected_shape) != len(actual_shape):
+        raise MlflowException(
+            "Shape of input {0} does not match expected shape {1}.".format(
+                actual_shape, expected_shape
+            )
+        )
+    for expected, actual in zip(expected_shape, actual_shape):
+        if expected == -1:
+            continue
+        if expected != actual:
+            raise MlflowException(
+                "Shape of input {0} does not match expected shape {1}.".format(
+                    actual_shape, expected_shape
+                )
+            )
+    if clean_tensor_type(actual_type) != tensor_spec.type:
+        raise MlflowException(
+            "dtype of input {0} does not match expected dtype {1}".format(
+                values.dtype, tensor_spec.type
+            )
+        )
+    return values
+
+
+def _enforce_mlflow_datatype(name, values: pd.Series, t: DataType):
+    """
+    Enforce the input column type matches the declared in model input schema.
+
+    The following type conversions are allowed:
+
+    1. object -> string
+    2. int -> long (upcast)
+    3. float -> double (upcast)
+    4. int -> double (safe conversion)
+    5. np.datetime64[x] -> datetime (any precision)
+    6. object -> datetime
+
+    Any other type mismatch will raise error.
+    """
+    if values.dtype == object and t not in (DataType.binary, DataType.string):
+        values = values.infer_objects()
+
+    if t == DataType.string and values.dtype == object:
+        # NB: the object can contain any type and we currently cannot cast to pandas Strings
+        # due to how None is cast
+        return values
+
+    # NB: Comparison of pandas and numpy data type fails when numpy data type is on the left hand
+    # side of the comparison operator. It works, however, if pandas type is on the left hand side.
+    # That is because pandas is aware of numpy.
+    if t.to_pandas() == values.dtype or t.to_numpy() == values.dtype:
+        # The types are already compatible => conversion is not necessary.
+        return values
+
+    if t == DataType.binary and values.dtype.kind == t.binary.to_numpy().kind:
+        # NB: bytes in numpy have variable itemsize depending on the length of the longest
+        # element in the array (column). Since MLflow binary type is length agnostic, we ignore
+        # itemsize when matching binary columns.
+        return values
+
+    if t == DataType.datetime and values.dtype.kind == t.to_numpy().kind:
+        # NB: datetime values have variable precision denoted by brackets, e.g. datetime64[ns]
+        # denotes nanosecond precision. Since MLflow datetime type is precision agnostic, we
+        # ignore precision when matching datetime columns.
+        return values
+
+    if t == DataType.datetime and values.dtype == object:
+        # NB: Pyspark date columns get converted to object when converted to a pandas
+        # DataFrame. To respect the original typing, we convert the column to datetime.
+        try:
+            return values.astype(np.datetime64, errors="raise")
+        except ValueError:
+            raise MlflowException(
+                "Failed to convert column {0} from type {1} to {2}.".format(name, values.dtype, t)
+            )
+
+    numpy_type = t.to_numpy()
+    if values.dtype.kind == numpy_type.kind:
+        is_upcast = values.dtype.itemsize <= numpy_type.itemsize
+    elif values.dtype.kind == "u" and numpy_type.kind == "i":
+        is_upcast = values.dtype.itemsize < numpy_type.itemsize
+    elif values.dtype.kind in ("i", "u") and numpy_type == np.float64:
+        # allow (u)int => double conversion
+        is_upcast = values.dtype.itemsize <= 6
+    else:
+        is_upcast = False
+
+    if is_upcast:
+        return values.astype(numpy_type, errors="raise")
+    else:
+        # NB: conversion between incompatible types (e.g. floats -> ints or
+        # double -> float) are not allowed. While supported by pandas and numpy,
+        # these conversions alter the values significantly.
+        def all_ints(xs):
+            return all(pd.isnull(x) or int(x) == x for x in xs)
+
+        hint = ""
+        if (
+            values.dtype == np.float64
+            and numpy_type.kind in ("i", "u")
+            and values.hasnans
+            and all_ints(values)
+        ):
+            hint = (
+                " Hint: the type mismatch is likely caused by missing values. "
+                "Integer columns in python can not represent missing values and are therefore "
+                "encoded as floats. The best way to avoid this problem is to infer the model "
+                "schema based on a realistic data sample (training dataset) that includes missing "
+                "values. Alternatively, you can declare integer columns as doubles (float64) "
+                "whenever these columns may have missing values. See `Handling Integers With "
+                "Missing Values <https://www.mlflow.org/docs/latest/models.html#"
+                "handling-integers-with-missing-values>`_ for more details."
+            )
+
+        raise MlflowException(
+            "Incompatible input types for column {0}. "
+            "Can not safely convert {1} to {2}.{3}".format(name, values.dtype, numpy_type, hint)
+        )
+
+
+def _enforce_col_schema(pfInput: PyFuncInput, input_schema: Schema):
+    """Enforce the input columns conform to the model's column-based signature."""
+    if input_schema.has_input_names():
+        input_names = input_schema.input_names()
+    else:
+        input_names = pfInput.columns[: len(input_schema.inputs)]
+    input_types = input_schema.input_types()
+    new_pfInput = pd.DataFrame()
+    for i, x in enumerate(input_names):
+        new_pfInput[x] = _enforce_mlflow_datatype(x, pfInput[x], input_types[i])
+    return new_pfInput
+
+
+def _enforce_tensor_schema(pfInput: PyFuncInput, input_schema: Schema):
+    """Enforce the input tensor(s) conforms to the model's tensor-based signature."""
+
+    def _is_sparse_matrix(x):
+        if not HAS_SCIPY:
+            # we can safely assume that it's not a sparse matrix if scipy is not installed
+            return False
+        return isinstance(x, (csr_matrix, csc_matrix))
+
+    if input_schema.has_input_names():
+        if isinstance(pfInput, dict):
+            new_pfInput = dict()
+            for col_name, tensor_spec in zip(input_schema.input_names(), input_schema.inputs):
+                if not isinstance(pfInput[col_name], np.ndarray):
+                    raise MlflowException(
+                        "This model contains a tensor-based model signature with input names,"
+                        " which suggests a dictionary input mapping input name to a numpy"
+                        " array, but a dict with value type {0} was found.".format(
+                            type(pfInput[col_name])
+                        )
+                    )
+                new_pfInput[col_name] = _enforce_tensor_spec(pfInput[col_name], tensor_spec)
+        elif isinstance(pfInput, pd.DataFrame):
+            new_pfInput = dict()
+            for col_name, tensor_spec in zip(input_schema.input_names(), input_schema.inputs):
+                new_pfInput[col_name] = _enforce_tensor_spec(
+                    np.array(pfInput[col_name], dtype=tensor_spec.type), tensor_spec
+                )
+        else:
+            raise MlflowException(
+                "This model contains a tensor-based model signature with input names, which"
+                " suggests a dictionary input mapping input name to tensor, but an input of"
+                " type {0} was found.".format(type(pfInput))
+            )
+    else:
+        if isinstance(pfInput, pd.DataFrame):
+            new_pfInput = _enforce_tensor_spec(pfInput.to_numpy(), input_schema.inputs[0])
+        elif isinstance(pfInput, np.ndarray) or _is_sparse_matrix(pfInput):
+            new_pfInput = _enforce_tensor_spec(pfInput, input_schema.inputs[0])
+        else:
+            raise MlflowException(
+                "This model contains a tensor-based model signature with no input names,"
+                " which suggests a numpy array input, but an input of type {0} was"
+                " found.".format(type(pfInput))
+            )
+    return new_pfInput
+
+
+def _enforce_schema(pfInput: PyFuncInput, input_schema: Schema):
+    """
+    Enforces the provided input matches the model's input schema,
+
+    For signatures with input names, we check there are no missing inputs and reorder the inputs to
+    match the ordering declared in schema if necessary. Any extra columns are ignored.
+
+    For column-based signatures, we make sure the types of the input match the type specified in
+    the schema or if it can be safely converted to match the input schema.
+
+    For tensor-based signatures, we make sure the shape and type of the input matches the shape
+    and type specified in model's input schema.
+    """
+    if isinstance(pfInput, pd.Series):
+        pfInput = pd.DataFrame(pfInput)
+    if not input_schema.is_tensor_spec():
+        if isinstance(pfInput, (list, np.ndarray, dict, pd.Series)):
+            try:
+                pfInput = pd.DataFrame(pfInput)
+            except Exception as e:
+                raise MlflowException(
+                    "This model contains a column-based signature, which suggests a DataFrame"
+                    " input. There was an error casting the input data to a DataFrame:"
+                    " {0}".format(str(e))
+                )
+        if not isinstance(pfInput, pd.DataFrame):
+            raise MlflowException(
+                "Expected input to be DataFrame or list. Found: %s" % type(pfInput).__name__
+            )
+
+    if input_schema.has_input_names():
+        # make sure there are no missing columns
+        input_names = input_schema.input_names()
+        expected_cols = set(input_names)
+        actual_cols = set()
+        if len(expected_cols) == 1 and isinstance(pfInput, np.ndarray):
+            # for schemas with a single column, match input with column
+            pfInput = {input_names[0]: pfInput}
+            actual_cols = expected_cols
+        elif isinstance(pfInput, pd.DataFrame):
+            actual_cols = set(pfInput.columns)
+        elif isinstance(pfInput, dict):
+            actual_cols = set(pfInput.keys())
+        missing_cols = expected_cols - actual_cols
+        extra_cols = actual_cols - expected_cols
+        # Preserve order from the original columns, since missing/extra columns are likely to
+        # be in same order.
+        missing_cols = [c for c in input_names if c in missing_cols]
+        extra_cols = [c for c in actual_cols if c in extra_cols]
+        if missing_cols:
+            message = "Model is missing inputs {0}.".format(missing_cols)
+            if extra_cols:
+                message += " Note that there were extra inputs: {0}".format(extra_cols)
+            raise MlflowException(message)
+    elif not input_schema.is_tensor_spec():
+        # The model signature does not specify column names => we can only verify column count.
+        num_actual_columns = len(pfInput.columns)
+        if num_actual_columns < len(input_schema.inputs):
+            raise MlflowException(
+                "Model inference is missing inputs. The model signature declares "
+                "{0} inputs  but the provided value only has "
+                "{1} inputs. Note: the inputs were not named in the signature so we can "
+                "only verify their count.".format(len(input_schema.inputs), num_actual_columns)
+            )
+
+    return (
+        _enforce_tensor_schema(pfInput, input_schema)
+        if input_schema.is_tensor_spec()
+        else _enforce_col_schema(pfInput, input_schema)
+    )
+
+
+def validate_schema(data: PyFuncInput, expected_schema: Schema) -> None:
+    """
+    Validate that the input data has the expected schema.
+
+    :param data: Input data to be validated. Supported types are:
+
+                 - pandas.DataFrame
+                 - pandas.Series
+                 - numpy.ndarray
+                 - scipy.sparse.csc_matrix
+                 - scipy.sparse.csr_matrix
+                 - List[Any]
+                 - Dict[str, Any]
+    :param expected_schema: Expected :py:class:`Schema <mlflow.types.Schema>` of the input data.
+    :raises: A :py:class:`mlflow.exceptions.MlflowException`. when the input data does
+             not match the schema.
+
+    .. code-block:: python
+        :caption: Example usage of validate_schema
+
+        import mlflow.models
+
+        # Suppose you've already got a model_uri
+        model_info = mlflow.models.get_model_info(model_uri)
+        # Get model signature directly
+        model_signature = model_info.signature
+        # validate schema
+        mlflow.models.validate_schema(input_data, model_signature.inputs)
+    """
+    _enforce_schema(data, expected_schema)
diff --git a/mlflow/pipelines/steps/evaluate.py b/mlflow/pipelines/steps/evaluate.py
index ea2dea04b..98d091853 100644
--- a/mlflow/pipelines/steps/evaluate.py
+++ b/mlflow/pipelines/steps/evaluate.py
@@ -81,8 +81,11 @@ class EvaluateStep(BaseStep):
             metric_name = val_criterion["metric"]
             metric_val = metrics.get(metric_name)
             if metric_val is None:
-                summary[metric_name] = False
-                continue
+                raise MlflowException(
+                    f"The metric {metric_name} is defined in the pipeline's validation criteria"
+                    " but was not returned from mlflow evaluation.",
+                    error_code=INVALID_PARAMETER_VALUE,
+                )
             greater_is_better = self.evaluation_metrics[metric_name].greater_is_better
             comp_func = operator.ge if greater_is_better else operator.le
             threshold = val_criterion["threshold"]
diff --git a/mlflow/pipelines/utils/__init__.py b/mlflow/pipelines/utils/__init__.py
index 9a6ebb47e..7df9a025d 100644
--- a/mlflow/pipelines/utils/__init__.py
+++ b/mlflow/pipelines/utils/__init__.py
@@ -1,5 +1,6 @@
 import logging
 import os
+import posixpath
 import pathlib
 from typing import Dict, Any
 
@@ -40,6 +41,8 @@ def get_pipeline_config(pipeline_root_path: str = None, profile: str = None) ->
     :param pipeline_root_path: The absolute path of the pipeline root directory on the local
                                filesystem. If unspecified, the pipeline root directory is
                                resolved from the current working directory, and an
+    :param profile: The name of the profile under the `profiles` directory to use,
+                    e.g. "dev" to use configs from "profiles/dev.yaml"
     :raises MlflowException: If the specified ``pipeline_root_path`` is not a pipeline root
                              directory or if ``pipeline_root_path`` is ``None`` and the current
                              working directory does not correspond to a pipeline.
@@ -49,6 +52,9 @@ def get_pipeline_config(pipeline_root_path: str = None, profile: str = None) ->
     _verify_is_pipeline_root_directory(pipeline_root_path=pipeline_root_path)
     try:
         if profile:
+            # Jinja expects template names in posixpath format relative to environment root,
+            # so use posixpath to construct the relative path here.
+            profile_relpath = posixpath.join(_PIPELINE_PROFILE_DIR, f"{profile}.yaml")
             profile_file_path = os.path.join(
                 pipeline_root_path, _PIPELINE_PROFILE_DIR, f"{profile}.yaml"
             )
@@ -59,7 +65,7 @@ def get_pipeline_config(pipeline_root_path: str = None, profile: str = None) ->
                     error_code=INVALID_PARAMETER_VALUE,
                 )
             return render_and_merge_yaml(
-                pipeline_root_path, _PIPELINE_CONFIG_FILE_NAME, profile_file_path
+                pipeline_root_path, _PIPELINE_CONFIG_FILE_NAME, profile_relpath
             )
         else:
             return read_yaml(pipeline_root_path, _PIPELINE_CONFIG_FILE_NAME)
diff --git a/mlflow/pyfunc/__init__.py b/mlflow/pyfunc/__init__.py
index 933bd5848..c9aa043f9 100644
--- a/mlflow/pyfunc/__init__.py
+++ b/mlflow/pyfunc/__init__.py
@@ -222,12 +222,12 @@ import threading
 import collections
 import subprocess
 
-from typing import Any, Union, List, Dict, Iterator, Tuple
+from typing import Any, Union, Iterator, Tuple
 import mlflow
 import mlflow.pyfunc.model
 from mlflow.models import Model, ModelSignature, ModelInputExample
 from mlflow.models.model import MLMODEL_FILE_NAME
-from mlflow.models.utils import _save_example
+from mlflow.models.utils import PyFuncInput, PyFuncOutput, _enforce_schema, _save_example
 from mlflow.pyfunc.model import (  # pylint: disable=unused-import
     PythonModel,
     PythonModelContext,
@@ -235,8 +235,6 @@ from mlflow.pyfunc.model import (  # pylint: disable=unused-import
 )
 from mlflow.pyfunc.model import get_default_pip_requirements
 from mlflow.tracking.artifact_utils import _download_artifact_from_uri
-from mlflow.types import DataType, Schema, TensorSpec
-from mlflow.types.utils import clean_tensor_type
 from mlflow.utils import PYTHON_VERSION, get_major_minor_py_version, _is_in_ipython_notebook
 from mlflow.utils.annotations import deprecated
 from mlflow.utils.file_utils import _copy_file_or_tree, write_to
@@ -270,12 +268,6 @@ from mlflow.protos.databricks_pb2 import (
     RESOURCE_DOES_NOT_EXIST,
 )
 
-try:
-    import scipy.sparse
-
-    HAS_SCIPY = True
-except ImportError:
-    HAS_SCIPY = False
 from mlflow.utils.requirements_utils import (
     _check_requirement_satisfied,
     _parse_requirements,
@@ -291,15 +283,6 @@ ENV = "env"
 PY_VERSION = "python_version"
 
 _logger = logging.getLogger(__name__)
-PyFuncInput = Union[
-    pandas.DataFrame,
-    np.ndarray,
-    "scipy.sparse.csc_matrix",
-    "scipy.sparse.csr_matrix",
-    List[Any],
-    Dict[str, Any],
-]
-PyFuncOutput = Union[pandas.DataFrame, pandas.Series, np.ndarray, list]
 
 
 def add_to_model(model, loader_module, data=None, code=None, env=None, **kwargs):
@@ -346,269 +329,6 @@ def _load_model_env(path):
     return _get_flavor_configuration(model_path=path, flavor_name=FLAVOR_NAME).get(ENV, None)
 
 
-def _enforce_mlflow_datatype(name, values: pandas.Series, t: DataType):
-    """
-    Enforce the input column type matches the declared in model input schema.
-
-    The following type conversions are allowed:
-
-    1. object -> string
-    2. int -> long (upcast)
-    3. float -> double (upcast)
-    4. int -> double (safe conversion)
-    5. np.datetime64[x] -> datetime (any precision)
-    6. object -> datetime
-
-    Any other type mismatch will raise error.
-    """
-    if values.dtype == object and t not in (DataType.binary, DataType.string):
-        values = values.infer_objects()
-
-    if t == DataType.string and values.dtype == object:
-        # NB: the object can contain any type and we currently cannot cast to pandas Strings
-        # due to how None is cast
-        return values
-
-    # NB: Comparison of pandas and numpy data type fails when numpy data type is on the left hand
-    # side of the comparison operator. It works, however, if pandas type is on the left hand side.
-    # That is because pandas is aware of numpy.
-    if t.to_pandas() == values.dtype or t.to_numpy() == values.dtype:
-        # The types are already compatible => conversion is not necessary.
-        return values
-
-    if t == DataType.binary and values.dtype.kind == t.binary.to_numpy().kind:
-        # NB: bytes in numpy have variable itemsize depending on the length of the longest
-        # element in the array (column). Since MLflow binary type is length agnostic, we ignore
-        # itemsize when matching binary columns.
-        return values
-
-    if t == DataType.datetime and values.dtype.kind == t.to_numpy().kind:
-        # NB: datetime values have variable precision denoted by brackets, e.g. datetime64[ns]
-        # denotes nanosecond precision. Since MLflow datetime type is precision agnostic, we
-        # ignore precision when matching datetime columns.
-        return values
-
-    if t == DataType.datetime and values.dtype == object:
-        # NB: Pyspark date columns get converted to object when converted to a pandas
-        # DataFrame. To respect the original typing, we convert the column to datetime.
-        try:
-            return values.astype(np.datetime64, errors="raise")
-        except ValueError:
-            raise MlflowException(
-                "Failed to convert column {0} from type {1} to {2}.".format(name, values.dtype, t)
-            )
-
-    numpy_type = t.to_numpy()
-    if values.dtype.kind == numpy_type.kind:
-        is_upcast = values.dtype.itemsize <= numpy_type.itemsize
-    elif values.dtype.kind == "u" and numpy_type.kind == "i":
-        is_upcast = values.dtype.itemsize < numpy_type.itemsize
-    elif values.dtype.kind in ("i", "u") and numpy_type == np.float64:
-        # allow (u)int => double conversion
-        is_upcast = values.dtype.itemsize <= 6
-    else:
-        is_upcast = False
-
-    if is_upcast:
-        return values.astype(numpy_type, errors="raise")
-    else:
-        # NB: conversion between incompatible types (e.g. floats -> ints or
-        # double -> float) are not allowed. While supported by pandas and numpy,
-        # these conversions alter the values significantly.
-        def all_ints(xs):
-            return all(pandas.isnull(x) or int(x) == x for x in xs)
-
-        hint = ""
-        if (
-            values.dtype == np.float64
-            and numpy_type.kind in ("i", "u")
-            and values.hasnans
-            and all_ints(values)
-        ):
-            hint = (
-                " Hint: the type mismatch is likely caused by missing values. "
-                "Integer columns in python can not represent missing values and are therefore "
-                "encoded as floats. The best way to avoid this problem is to infer the model "
-                "schema based on a realistic data sample (training dataset) that includes missing "
-                "values. Alternatively, you can declare integer columns as doubles (float64) "
-                "whenever these columns may have missing values. See `Handling Integers With "
-                "Missing Values <https://www.mlflow.org/docs/latest/models.html#"
-                "handling-integers-with-missing-values>`_ for more details."
-            )
-
-        raise MlflowException(
-            "Incompatible input types for column {0}. "
-            "Can not safely convert {1} to {2}.{3}".format(name, values.dtype, numpy_type, hint)
-        )
-
-
-def _enforce_tensor_spec(
-    values: Union[np.ndarray, "scipy.sparse.csc_matrix", "scipy.sparse.csr_matrix"],
-    tensor_spec: TensorSpec,
-):
-    """
-    Enforce the input tensor shape and type matches the provided tensor spec.
-    """
-    expected_shape = tensor_spec.shape
-    actual_shape = values.shape
-
-    actual_type = values.dtype if isinstance(values, np.ndarray) else values.data.dtype
-
-    if len(expected_shape) != len(actual_shape):
-        raise MlflowException(
-            "Shape of input {0} does not match expected shape {1}.".format(
-                actual_shape, expected_shape
-            )
-        )
-    for expected, actual in zip(expected_shape, actual_shape):
-        if expected == -1:
-            continue
-        if expected != actual:
-            raise MlflowException(
-                "Shape of input {0} does not match expected shape {1}.".format(
-                    actual_shape, expected_shape
-                )
-            )
-    if clean_tensor_type(actual_type) != tensor_spec.type:
-        raise MlflowException(
-            "dtype of input {0} does not match expected dtype {1}".format(
-                values.dtype, tensor_spec.type
-            )
-        )
-    return values
-
-
-def _enforce_col_schema(pfInput: PyFuncInput, input_schema: Schema):
-    """Enforce the input columns conform to the model's column-based signature."""
-    if input_schema.has_input_names():
-        input_names = input_schema.input_names()
-    else:
-        input_names = pfInput.columns[: len(input_schema.inputs)]
-    input_types = input_schema.input_types()
-    new_pfInput = pandas.DataFrame()
-    for i, x in enumerate(input_names):
-        new_pfInput[x] = _enforce_mlflow_datatype(x, pfInput[x], input_types[i])
-    return new_pfInput
-
-
-def _enforce_tensor_schema(pfInput: PyFuncInput, input_schema: Schema):
-    """Enforce the input tensor(s) conforms to the model's tensor-based signature."""
-
-    def _is_sparse_matrix(x):
-        if not HAS_SCIPY:
-            # we can safely assume that it's not a sparse matrix if scipy is not installed
-            return False
-        return isinstance(x, (scipy.sparse.csr_matrix, scipy.sparse.csc_matrix))
-
-    if input_schema.has_input_names():
-        if isinstance(pfInput, dict):
-            new_pfInput = dict()
-            for col_name, tensor_spec in zip(input_schema.input_names(), input_schema.inputs):
-                if not isinstance(pfInput[col_name], np.ndarray):
-                    raise MlflowException(
-                        "This model contains a tensor-based model signature with input names,"
-                        " which suggests a dictionary input mapping input name to a numpy"
-                        " array, but a dict with value type {0} was found.".format(
-                            type(pfInput[col_name])
-                        )
-                    )
-                new_pfInput[col_name] = _enforce_tensor_spec(pfInput[col_name], tensor_spec)
-        elif isinstance(pfInput, pandas.DataFrame):
-            new_pfInput = dict()
-            for col_name, tensor_spec in zip(input_schema.input_names(), input_schema.inputs):
-                new_pfInput[col_name] = _enforce_tensor_spec(
-                    np.array(pfInput[col_name], dtype=tensor_spec.type), tensor_spec
-                )
-        else:
-            raise MlflowException(
-                "This model contains a tensor-based model signature with input names, which"
-                " suggests a dictionary input mapping input name to tensor, but an input of"
-                " type {0} was found.".format(type(pfInput))
-            )
-    else:
-        if isinstance(pfInput, pandas.DataFrame):
-            new_pfInput = _enforce_tensor_spec(pfInput.to_numpy(), input_schema.inputs[0])
-        elif isinstance(pfInput, np.ndarray) or _is_sparse_matrix(pfInput):
-            new_pfInput = _enforce_tensor_spec(pfInput, input_schema.inputs[0])
-        else:
-            raise MlflowException(
-                "This model contains a tensor-based model signature with no input names,"
-                " which suggests a numpy array input, but an input of type {0} was"
-                " found.".format(type(pfInput))
-            )
-    return new_pfInput
-
-
-def _enforce_schema(pfInput: PyFuncInput, input_schema: Schema):
-    """
-    Enforces the provided input matches the model's input schema,
-
-    For signatures with input names, we check there are no missing inputs and reorder the inputs to
-    match the ordering declared in schema if necessary. Any extra columns are ignored.
-
-    For column-based signatures, we make sure the types of the input match the type specified in
-    the schema or if it can be safely converted to match the input schema.
-
-    For tensor-based signatures, we make sure the shape and type of the input matches the shape
-    and type specified in model's input schema.
-    """
-    if not input_schema.is_tensor_spec():
-        if isinstance(pfInput, (list, np.ndarray, dict)):
-            try:
-                pfInput = pandas.DataFrame(pfInput)
-            except Exception as e:
-                raise MlflowException(
-                    "This model contains a column-based signature, which suggests a DataFrame"
-                    " input. There was an error casting the input data to a DataFrame:"
-                    " {0}".format(str(e))
-                )
-        if not isinstance(pfInput, pandas.DataFrame):
-            raise MlflowException(
-                "Expected input to be DataFrame or list. Found: %s" % type(pfInput).__name__
-            )
-
-    if input_schema.has_input_names():
-        # make sure there are no missing columns
-        input_names = input_schema.input_names()
-        expected_cols = set(input_names)
-        actual_cols = set()
-        if len(expected_cols) == 1 and isinstance(pfInput, np.ndarray):
-            # for schemas with a single column, match input with column
-            pfInput = {input_names[0]: pfInput}
-            actual_cols = expected_cols
-        elif isinstance(pfInput, pandas.DataFrame):
-            actual_cols = set(pfInput.columns)
-        elif isinstance(pfInput, dict):
-            actual_cols = set(pfInput.keys())
-        missing_cols = expected_cols - actual_cols
-        extra_cols = actual_cols - expected_cols
-        # Preserve order from the original columns, since missing/extra columns are likely to
-        # be in same order.
-        missing_cols = [c for c in input_names if c in missing_cols]
-        extra_cols = [c for c in actual_cols if c in extra_cols]
-        if missing_cols:
-            message = "Model is missing inputs {0}.".format(missing_cols)
-            if extra_cols:
-                message += " Note that there were extra inputs: {0}".format(extra_cols)
-            raise MlflowException(message)
-    elif not input_schema.is_tensor_spec():
-        # The model signature does not specify column names => we can only verify column count.
-        num_actual_columns = len(pfInput.columns)
-        if num_actual_columns < len(input_schema.inputs):
-            raise MlflowException(
-                "Model inference is missing inputs. The model signature declares "
-                "{0} inputs  but the provided value only has "
-                "{1} inputs. Note: the inputs were not named in the signature so we can "
-                "only verify their count.".format(len(input_schema.inputs), num_actual_columns)
-            )
-
-    return (
-        _enforce_tensor_schema(pfInput, input_schema)
-        if input_schema.is_tensor_spec()
-        else _enforce_col_schema(pfInput, input_schema)
-    )
-
-
 class PyFuncModel:
     """
     MLflow 'python function' model.
diff --git a/mlflow/sagemaker/__init__.py b/mlflow/sagemaker/__init__.py
index f4fad0b34..a536d768f 100644
--- a/mlflow/sagemaker/__init__.py
+++ b/mlflow/sagemaker/__init__.py
@@ -19,7 +19,7 @@ from mlflow.models.model import MLMODEL_FILE_NAME
 from mlflow.protos.databricks_pb2 import RESOURCE_DOES_NOT_EXIST, INVALID_PARAMETER_VALUE
 from mlflow.tracking.artifact_utils import _download_artifact_from_uri
 from mlflow.utils import get_unique_resource_id
-from mlflow.utils.annotations import experimental
+from mlflow.utils.annotations import experimental, deprecated
 from mlflow.utils.file_utils import TempDir
 from mlflow.models.container import SUPPORTED_FLAVORS as SUPPORTED_DEPLOYMENT_FLAVORS
 from mlflow.models.container import DEPLOYMENT_CONFIG_KEY_FLAVOR_NAME, SERVING_ENVIRONMENT
@@ -160,6 +160,12 @@ def push_image_to_ecr(image=DEFAULT_IMAGE_NAME):
     os.system(cmd)
 
 
+@deprecated(
+    alternative=(
+        "mlflow.deployments.get_deploy_client('sagemaker').create_deployment()"
+        " and mlflow.deployments.get_deploy_client('sagemaker').update_deployment()"
+    )
+)
 def deploy(
     app_name,
     model_uri,
@@ -177,6 +183,7 @@ def deploy(
     synchronous=True,
     timeout_seconds=1200,
     data_capture_config=None,
+    variant_name=None,
 ):
     """
     Deploy an MLflow model on AWS SageMaker.
@@ -261,19 +268,19 @@ def deploy(
                        #SageMaker.Client.create_model>`_. For more information, see
                        https://docs.aws.amazon.com/sagemaker/latest/dg/API_VpcConfig.html.
 
-    .. code-block:: python
-        :caption: Example
+                       .. code-block:: python
+                           :caption: Example
 
-        import mlflow.sagemaker as mfs
-        vpc_config = {
-                        'SecurityGroupIds': [
-                            'sg-123456abc',
-                        ],
-                        'Subnets': [
-                            'subnet-123456abc',
-                        ]
-                     }
-        mfs.deploy(..., vpc_config=vpc_config)
+                           import mlflow.sagemaker as mfs
+                           vpc_config = {
+                               'SecurityGroupIds': [
+                                   'sg-123456abc',
+                               ],
+                               'Subnets': [
+                                   'subnet-123456abc',
+                               ]
+                            }
+                            mfs.deploy(..., vpc_config=vpc_config)
 
     :param flavor: The name of the flavor of the model to use for deployment. Must be either
                    ``None`` or one of mlflow.sagemaker.SUPPORTED_DEPLOYMENT_FLAVORS. If ``None``,
@@ -293,24 +300,25 @@ def deploy(
                             deployment using native SageMaker APIs or the AWS console. If
                             ``synchronous`` is ``False``, this parameter is ignored.
     :param data_capture_config: A dictionary specifying the data capture configuration to use when
-                       creating the new SageMaker model associated with this application. For more
-                       information, see
-                       https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DataCaptureConfig.html.
-
-    .. code-block:: python
-        :caption: Example
-
-        import mlflow.sagemaker as mfs
-        data_capture_config = {
-                        'EnableCapture': True,
-                        'InitalSamplingPercentage': 100,
-                        'DestinationS3Uri": 's3://my-bucket/path',
-                        'CaptureOptions': [
-                            {'CaptureMode': 'Output'}
-                        ],
-                     }
-        mfs.deploy(..., data_capture_config=data_capture_config)
-
+                                creating the new SageMaker model associated with this application.
+                                For more information, see
+                                https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DataCaptureConfig.html.
+
+                                .. code-block:: python
+                                    :caption: Example
+
+                                    import mlflow.sagemaker as mfs
+                                    data_capture_config = {
+                                                    'EnableCapture': True,
+                                                    'InitalSamplingPercentage': 100,
+                                                    'DestinationS3Uri": 's3://my-bucket/path',
+                                                    'CaptureOptions': [
+                                                        {'CaptureMode': 'Output'}
+                                                    ],
+                                                 }
+                                    mfs.deploy(..., data_capture_config=data_capture_config)
+
+    :param variant_name: The name to assign to the new production variant.
     """
     import boto3
 
@@ -403,6 +411,7 @@ def deploy(
             role=execution_role_arn,
             sage_client=sage_client,
             s3_client=s3_client,
+            variant_name=variant_name,
         )
     else:
         deployment_operation = _create_sagemaker_endpoint(
@@ -418,6 +427,7 @@ def deploy(
             data_capture_config=data_capture_config,
             role=execution_role_arn,
             sage_client=sage_client,
+            variant_name=variant_name,
         )
 
     if synchronous:
@@ -439,6 +449,7 @@ def deploy(
     return app_name, flavor
 
 
+@deprecated(alternative="mlflow.deployments.get_deploy_client('sagemaker').delete_deployment()")
 def delete(
     app_name,
     region_name="us-west-2",
@@ -633,19 +644,19 @@ def deploy_transform_job(
                        #SageMaker.Client.create_model>`_. For more information, see
                        https://docs.aws.amazon.com/sagemaker/latest/dg/API_VpcConfig.html.
 
-    .. code-block:: python
-        :caption: Example
+                       .. code-block:: python
+                           :caption: Example
 
-        import mlflow.sagemaker as mfs
-        vpc_config = {
-                        'SecurityGroupIds': [
-                            'sg-123456abc',
-                        ],
-                        'Subnets': [
-                            'subnet-123456abc',
-                        ]
-                     }
-        mfs.deploy_transform_job(..., vpc_config=vpc_config)
+                           import mlflow.sagemaker as mfs
+                           vpc_config = {
+                               'SecurityGroupIds': [
+                                   'sg-123456abc',
+                               ],
+                               'Subnets': [
+                                   'subnet-123456abc',
+                               ]
+                            }
+                            mfs.deploy_transform_job(..., vpc_config=vpc_config)
 
     :param flavor: The name of the flavor of the model to use for deployment. Must be either
                    ``None`` or one of mlflow.sagemaker.SUPPORTED_DEPLOYMENT_FLAVORS. If ``None``,
@@ -935,19 +946,19 @@ def push_model_to_sagemaker(
                        #SageMaker.Client.create_model>`_. For more information, see
                        https://docs.aws.amazon.com/sagemaker/latest/dg/API_VpcConfig.html.
 
-    .. code-block:: python
-        :caption: Example
+                       .. code-block:: python
+                           :caption: Example
 
-        import mlflow.sagemaker as mfs
-        vpc_config = {
-                        'SecurityGroupIds': [
-                            'sg-123456abc',
-                        ],
-                        'Subnets': [
-                            'subnet-123456abc',
-                        ]
-                     }
-        mfs.push_model_to_sagemaker(..., vpc_config=vpc_config)
+                           import mlflow.sagemaker as mfs
+                           vpc_config = {
+                               'SecurityGroupIds': [
+                                   'sg-123456abc',
+                               ],
+                               'Subnets': [
+                                   'subnet-123456abc',
+                               ]
+                            }
+                            mfs.push_model_to_sagemaker(..., vpc_config=vpc_config)
 
     :param flavor: The name of the flavor of the model to use for deployment. Must be either
                    ``None`` or one of mlflow.sagemaker.SUPPORTED_DEPLOYMENT_FLAVORS. If ``None``,
@@ -1077,9 +1088,6 @@ def run_local(model_uri, port=5000, image=DEFAULT_IMAGE_NAME, flavor=None):
 def target_help():
     """
     Provide help information for the SageMaker deployment client.
-
-    :return:
-    :rtype: str
     """
     help_str = """\
     For detailed documentation on the SageMaker deployment client, please visit
@@ -1439,6 +1447,7 @@ def _create_sagemaker_endpoint(
     instance_count,
     role,
     sage_client,
+    variant_name=None,
 ):
     """
     :param endpoint_name: The name of the SageMaker endpoint to create.
@@ -1456,6 +1465,7 @@ def _create_sagemaker_endpoint(
                        creating the new SageMaker model associated with this application.
     :param role: SageMaker execution ARN role.
     :param sage_client: A boto3 client for SageMaker.
+    :param variant_name: The name to assign to the new production variant.
     """
     _logger.info("Creating new endpoint with name: %s ...", endpoint_name)
 
@@ -1471,8 +1481,11 @@ def _create_sagemaker_endpoint(
     )
     _logger.info("Created model with arn: %s", model_response["ModelArn"])
 
+    if not variant_name:
+        variant_name = model_name
+
     production_variant = {
-        "VariantName": model_name,
+        "VariantName": variant_name,
         "ModelName": model_name,
         "InitialInstanceCount": instance_count,
         "InstanceType": instance_type,
@@ -1545,6 +1558,7 @@ def _update_sagemaker_endpoint(
     role,
     sage_client,
     s3_client,
+    variant_name=None,
 ):
     """
     :param endpoint_name: The name of the SageMaker endpoint to update.
@@ -1563,6 +1577,7 @@ def _update_sagemaker_endpoint(
     :param role: SageMaker execution ARN role.
     :param sage_client: A boto3 client for SageMaker.
     :param s3_client: A boto3 client for S3.
+    :variant_name: The name to assign to the new production variant if it doesn't already exist.
     """
     if mode not in [DEPLOYMENT_MODE_ADD, DEPLOYMENT_MODE_REPLACE]:
         msg = "Invalid mode `{md}` for deployment to a pre-existing application".format(md=mode)
@@ -1591,6 +1606,9 @@ def _update_sagemaker_endpoint(
     )
     _logger.info("Created new model with arn: %s", new_model_response["ModelArn"])
 
+    if not variant_name:
+        variant_name = model_name
+
     if mode == DEPLOYMENT_MODE_ADD:
         new_model_weight = 0
         production_variants = deployed_production_variants
@@ -1599,7 +1617,7 @@ def _update_sagemaker_endpoint(
         production_variants = []
 
     new_production_variant = {
-        "VariantName": model_name,
+        "VariantName": variant_name,
         "ModelName": model_name,
         "InitialInstanceCount": instance_count,
         "InstanceType": instance_type,
@@ -1891,6 +1909,7 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
             data_capture_config=None,
             synchronous=True,
             timeout_seconds=1200,
+            variant_name=None,
         )
 
         if create_mode:
@@ -2051,12 +2070,16 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
                          For more information, see
                          https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DataCaptureConfig.html.
                          Defaults to ``None``.
+
+                       - ``variant_name``: A string specifying the desired name when creating a
+                                           production variant.  Defaults to ``None``.
+
         :param endpoint: (optional) Endpoint to create the deployment under. Currently unsupported
 
         .. code-block:: python
             :caption: Python example
 
-            from mlflow.sagemaker import SageMakerDeploymentClient
+            from mlflow.deployments import get_deploy_client
 
             vpc_config = {
                 'SecurityGroupIds': [
@@ -2078,9 +2101,10 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
                 instance_count=1,
                 synchronous=True,
                 timeout_seconds=300,
-                vpc_config=vpc_config
+                vpc_config=vpc_config,
+                variant_name="prod-variant-1",
             )
-            client = SageMakerDeploymentClient("sagemaker")
+            client = get_deploy_client("sagemaker")
             client.create_deployment(
                 "my-deployment",
                 model_uri="/mlruns/0/abc/model",
@@ -2104,6 +2128,7 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
                     -C instance_count=1 \\
                     -C synchronous=True \\
                     -C timeout_seconds=300 \\
+                    -C variant_name=prod-variant-1 \\
                     -C vpc_config='{"SecurityGroupIds": ["sg-123456abc"], \\
                     "Subnets": ["subnet-123456abc"]}' \\
                     -C data_capture_config='{"EnableCapture": True, \\
@@ -2131,6 +2156,7 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
             data_capture_config=final_config["data_capture_config"],
             synchronous=final_config["synchronous"],
             timeout_seconds=final_config["timeout_seconds"],
+            variant_name=final_config["variant_name"],
         )
 
         return dict(name=app_name, flavor=flavor)
@@ -2248,6 +2274,9 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
                          ``synchronous`` is ``False``, this parameter is ignored.
                          Defaults to ``300``.
 
+                       - ``variant_name``: A string specifying the desired name when creating a
+                                           production variant.  Defaults to ``None``.
+
                        - ``vpc_config``: A dictionary specifying the VPC configuration to use when
                          creating the new SageMaker model associated with this application.
                          The acceptable values for this parameter are identical to those of the
@@ -2263,13 +2292,17 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
                          For more information, see
                          https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DataCaptureConfig.html.
                          Defaults to ``None``.
+
+                       - ``variant_name``: A string specifying the desired name when creating a
+                                           production variant.  Defaults to ``None``.
+
         :param endpoint: (optional) Endpoint containing the deployment to update. Currently
                          unsupported
 
         .. code-block:: python
             :caption: Python example
 
-            from mlflow.sagemaker import SageMakerDeploymentClient
+            from mlflow.deployments import get_deploy_client
 
             vpc_config = {
                 'SecurityGroupIds': [
@@ -2299,10 +2332,11 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
                 instance_count=1,
                 synchronous=True,
                 timeout_seconds=300,
+                variant_name="prod-variant-1",
                 vpc_config=vpc_config
                 data_capture_config=data_capture_config
             )
-            client = SageMakerDeploymentClient("sagemaker")
+            client = get_deploy_client("sagemaker")
             client.update_deployment(
                 "my-deployment",
                 model_uri="/mlruns/0/abc/model",
@@ -2326,6 +2360,7 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
                     -C instance_count=1 \\
                     -C synchronous=True \\
                     -C timeout_seconds=300 \\
+                    -C variant_name=prod-variant-1 \\
                     -C vpc_config='{"SecurityGroupIds": ["sg-123456abc"], \\
                     "Subnets": ["subnet-123456abc"]}' \\
                     -C data_capture_config='{"EnableCapture": True, \\
@@ -2366,6 +2401,7 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
             data_capture_config=final_config["data_capture_config"],
             synchronous=final_config["synchronous"],
             timeout_seconds=final_config["timeout_seconds"],
+            variant_name=final_config["variant_name"],
         )
 
         return dict(name=app_name, flavor=flavor)
@@ -2410,7 +2446,7 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
         .. code-block:: python
             :caption: Python example
 
-            from mlflow.sagemaker import SageMakerDeploymentClient
+            from mlflow.deployments import get_deploy_client
 
             config = dict(
                 assume_role_arn="arn:aws:123:role/assumed_role",
@@ -2419,7 +2455,7 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
                 synchronous=True,
                 timeout_seconds=300
             )
-            client = SageMakerDeploymentClient("sagemaker")
+            client = get_deploy_client("sagemaker")
             client.delete_deployment("my-deployment", config=config)
 
         .. code-block:: bash
@@ -2471,9 +2507,9 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
         .. code-block:: python
             :caption: Python example
 
-            from mlflow.sagemaker import SageMakerDeploymentClient
+            from mlflow.deployments import get_deploy_client
 
-            client = SageMakerDeploymentClient("sagemaker:/us-east-1/arn:aws:123:role/assumed_role")
+            client = get_deploy_client("sagemaker:/us-east-1/arn:aws:123:role/assumed_role")
             client.list_deployments()
 
         .. code-block:: bash
@@ -2514,9 +2550,9 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
         .. code-block:: python
             :caption: Python example
 
-            from mlflow.sagemaker import SageMakerDeploymentClient
+            from mlflow.deployments import get_deploy_client
 
-            client = SageMakerDeploymentClient("sagemaker:/us-east-1/arn:aws:123:role/assumed_role")
+            client = get_deploy_client("sagemaker:/us-east-1/arn:aws:123:role/assumed_role")
             client.get_deployment("my-deployment")
 
         .. code-block:: bash
@@ -2566,10 +2602,10 @@ class SageMakerDeploymentClient(BaseDeploymentClient):
             :caption: Python example
 
             import pandas as pd
-            from mlflow.sagemaker import SageMakerDeploymentClient
+            from mlflow.deployments import get_deploy_client
 
             df = pd.DataFrame(data=[[1, 2, 3]], columns=["feat1", "feat2", "feat3"])
-            client = SageMakerDeploymentClient("sagemaker:/us-east-1/arn:aws:123:role/assumed_role")
+            client = get_deploy_client("sagemaker:/us-east-1/arn:aws:123:role/assumed_role")
             client.predict("my-deployment", df)
 
         .. code-block:: bash
diff --git a/mlflow/sagemaker/cli.py b/mlflow/sagemaker/cli.py
index 2102c60de..103b02640 100644
--- a/mlflow/sagemaker/cli.py
+++ b/mlflow/sagemaker/cli.py
@@ -7,7 +7,7 @@ import mlflow
 import mlflow.sagemaker
 from mlflow.sagemaker import DEFAULT_IMAGE_NAME as IMAGE
 from mlflow.utils import cli_args
-from mlflow.utils.annotations import experimental
+from mlflow.utils.annotations import experimental, deprecated
 import mlflow.models.docker_utils
 from mlflow.utils import env_manager as em
 
@@ -111,6 +111,9 @@ def commands():
         " asynchronously using the `--async` flag, this value is ignored."
     ),
 )
+@deprecated(
+    alternative="mlflow deployments create -t sagemaker and mlflow deployments update -t sagemaker"
+)
 def deploy(
     app_name,
     model_uri,
@@ -204,6 +207,7 @@ def deploy(
         " asynchronously using the `--async` flag, this value is ignored."
     ),
 )
+@deprecated(alternative="mlflow deployments delete -t sagemaker")
 def delete(app_name, region_name, archive, asynchronous, timeout):
     """
     Delete the specified application. Unless ``--archive`` is specified, all SageMaker resources
diff --git a/mlflow/utils/annotations.py b/mlflow/utils/annotations.py
index d614185bf..006c82c05 100644
--- a/mlflow/utils/annotations.py
+++ b/mlflow/utils/annotations.py
@@ -32,7 +32,6 @@ def _experimental(api: Any, api_type: str):
 def deprecated(alternative=None, since=None, impact=None):
     """
     Decorator for marking APIs deprecated in the docstring.
-
     :param func: A function to mark
     :returns Decorated function.
     """
diff --git a/mlflow/utils/file_utils.py b/mlflow/utils/file_utils.py
index 22754a739..a9f81179d 100644
--- a/mlflow/utils/file_utils.py
+++ b/mlflow/utils/file_utils.py
@@ -216,9 +216,6 @@ def render_and_merge_yaml(root, template_name, context_name):
         if not pathlib.Path(path).is_file():
             raise MissingConfigException("Yaml file '%s' does not exist." % path)
 
-    with codecs.open(context_path, mode="r", encoding=ENCODING) as context_file:
-        context_dict = yaml.load(context_file, Loader=UniqueKeyLoader) or {}
-
     j2_env = jinja2.Environment(
         loader=jinja2.FileSystemLoader(root, encoding=ENCODING), undefined=jinja2.StrictUndefined
     )
@@ -230,6 +227,12 @@ def render_and_merge_yaml(root, template_name, context_name):
             return json.load(f)
 
     j2_env.filters["from_json"] = from_json
+    # Compute final source of context file (e.g. my-profile.yml), applying Jinja filters
+    # like from_json as needed to load context information from files, then load into a dict
+    context_source = j2_env.get_template(context_name).render({})
+    context_dict = yaml.load(context_source, Loader=UniqueKeyLoader) or {}
+
+    # Substitute parameters from context dict into template
     source = j2_env.get_template(template_name).render(context_dict)
     rendered_template_dict = yaml.load(source, Loader=UniqueKeyLoader)
     return merge_dicts(rendered_template_dict, context_dict)


Test patch:
diff --git a/tests/models/test_model.py b/tests/models/test_model.py
index ebd81ae11..38b26504a 100644
--- a/tests/models/test_model.py
+++ b/tests/models/test_model.py
@@ -7,7 +7,7 @@ import pandas as pd
 import numpy as np
 
 from mlflow.tracking.artifact_utils import _download_artifact_from_uri
-from mlflow.models import Model
+from mlflow.models import Model, infer_signature, validate_schema
 from mlflow.models.signature import ModelSignature
 from mlflow.models.utils import _save_example
 from mlflow.types.schema import Schema, ColSpec, TensorSpec
@@ -17,9 +17,27 @@ from mlflow.utils.proto_json_utils import _dataframe_from_json
 
 from unittest import mock
 from scipy.sparse import csc_matrix
+import sklearn.datasets
+import sklearn.neighbors
 from packaging.version import Version
 
 
+@pytest.fixture(scope="module")
+def iris_data():
+    iris = sklearn.datasets.load_iris()
+    x = iris.data[:, :2]
+    y = iris.target
+    return x, y
+
+
+@pytest.fixture(scope="module")
+def sklearn_knn_model(iris_data):
+    x, y = iris_data
+    knn_model = sklearn.neighbors.KNeighborsClassifier()
+    knn_model.fit(x, y)
+    return knn_model
+
+
 def test_model_save_load():
     m = Model(
         artifact_path="some/path",
@@ -128,17 +146,29 @@ def test_model_info():
             model_info = Model.log(
                 "some/path", TestFlavor, signature=sig, input_example=input_example
             )
-        local_path = _download_artifact_from_uri(
-            "runs:/{}/some/path".format(run.info.run_id), output_path=tmp.path("")
-        )
+        model_uri = "runs:/{}/some/path".format(run.info.run_id)
+
+        model_info_fetched = mlflow.models.get_model_info(model_uri)
+        with pytest.warns(
+            FutureWarning,
+            match="Field signature_dict is deprecated since v1.28.1. Use signature instead",
+        ):
+            assert model_info_fetched.signature_dict == sig.to_dict()
+
+        local_path = _download_artifact_from_uri(model_uri, output_path=tmp.path(""))
 
         assert model_info.run_id == run.info.run_id
+        assert model_info_fetched.run_id == run.info.run_id
         assert model_info.artifact_path == "some/path"
-        assert model_info.model_uri == "runs:/{}/some/path".format(run.info.run_id)
+        assert model_info_fetched.artifact_path == "some/path"
+        assert model_info.model_uri == model_uri
+        assert model_info_fetched.model_uri == model_uri
 
         loaded_model = Model.load(os.path.join(local_path, "MLmodel"))
         assert model_info.utc_time_created == loaded_model.utc_time_created
+        assert model_info_fetched.utc_time_created == loaded_model.utc_time_created
         assert model_info.model_uuid == loaded_model.model_uuid
+        assert model_info_fetched.model_uuid == loaded_model.model_uuid
 
         assert model_info.flavors == {
             "flavor1": {"a": 1, "b": 2},
@@ -149,9 +179,12 @@ def test_model_info():
         x = _dataframe_from_json(path)
         assert x.to_dict(orient="records")[0] == input_example
 
+        model_signature = model_info_fetched.signature
         assert model_info.signature_dict == sig.to_dict()
+        assert model_signature.to_dict() == sig.to_dict()
 
-        assert Version(model_info.mlflow_version) == Version(loaded_model.mlflow_version)
+        assert model_info.mlflow_version == loaded_model.mlflow_version
+        assert model_info_fetched.mlflow_version == loaded_model.mlflow_version
 
 
 def test_load_model_without_mlflow_version():
@@ -325,3 +358,20 @@ def test_model_uuid():
     m_dict.pop("model_uuid")
     m4 = Model.from_dict(m_dict)
     assert m4.model_uuid is None
+
+
+def test_validate_schema(sklearn_knn_model, iris_data, tmpdir):
+    sk_model_path = os.path.join(str(tmpdir), "sk_model")
+    X, y = iris_data
+    signature = infer_signature(X, y)
+    mlflow.sklearn.save_model(
+        sklearn_knn_model,
+        sk_model_path,
+        signature=signature,
+    )
+
+    validate_schema(X, signature.inputs)
+    prediction = sklearn_knn_model.predict(X)
+    reloaded_model = mlflow.sklearn.load_model(sk_model_path)
+    np.testing.assert_array_equal(prediction, reloaded_model.predict(X))
+    validate_schema(prediction, signature.outputs)
diff --git a/tests/pipelines/test_utils.py b/tests/pipelines/test_utils.py
index 73fecdb11..9ba56fe73 100644
--- a/tests/pipelines/test_utils.py
+++ b/tests/pipelines/test_utils.py
@@ -1,3 +1,4 @@
+import json
 import os
 import pathlib
 import shutil
@@ -81,6 +82,60 @@ def test_get_pipeline_config_returns_correctly_for_valid_pipeline_directory(
         )
 
 
+def test_get_pipeline_config_for_pipeline_directory_referencing_external_json(
+    enter_pipeline_example_directory, tmp_path
+):
+    pipeline_root_path = enter_pipeline_example_directory
+    test_pipeline_root_path = tmp_path / "test_pipeline"
+    shutil.copytree(pipeline_root_path, test_pipeline_root_path)
+
+    test_pipeline_config = {
+        "config1": 10,
+        "config2": {
+            "subconfig": ["A"],
+        },
+        "config3": "3",
+        "config_from_profile": "{{PROFILE_CONFIG}}",
+    }
+    write_yaml(test_pipeline_root_path, "pipeline.yaml", test_pipeline_config, overwrite=True)
+
+    # Write a profiles/test.yaml file that references keys in an external JSON file
+    test_profile_name = "test"
+    profile_contents = {
+        "PROFILE_CONFIG": """{{ ("example-json.json" | from_json)["my-key"]["nested-key"] }}"""
+    }
+    profiles_dir = os.path.join(test_pipeline_root_path, "profiles")
+    if not os.path.exists(profiles_dir):
+        os.mkdir(profiles_dir)
+    write_yaml(
+        test_pipeline_root_path,
+        f"profiles/{test_profile_name}.yaml",
+        profile_contents,
+        overwrite=True,
+    )
+
+    # Write the external JSON file
+    value_from_json = "my-value-from-external-json"
+    example_json = {
+        "my-key": {
+            "nested-key": value_from_json,
+        },
+    }
+    with open(os.path.join(test_pipeline_root_path, "example-json.json"), "w") as handle:
+        handle.write(json.dumps(example_json))
+    # Load pipeline config and assert correctness
+    expected_config = {}
+    expected_config.update(test_pipeline_config)
+    expected_config.update(
+        {
+            "config_from_profile": value_from_json,
+            "PROFILE_CONFIG": value_from_json,
+        }
+    )
+    with chdir(test_pipeline_root_path):
+        assert get_pipeline_config(profile=test_profile_name) == expected_config
+
+
 def test_get_pipeline_config_throws_for_invalid_pipeline_directory(tmp_path):
     with pytest.raises(MlflowException, match="Failed to find pipeline.yaml"), chdir(tmp_path):
         get_pipeline_config()
diff --git a/tests/types/test_schema.py b/tests/types/test_schema.py
index e6e7028a8..b1d05ed64 100644
--- a/tests/types/test_schema.py
+++ b/tests/types/test_schema.py
@@ -6,7 +6,7 @@ import pytest
 from scipy.sparse import csr_matrix, csc_matrix
 
 from mlflow.exceptions import MlflowException
-from mlflow.pyfunc import _enforce_tensor_spec
+from mlflow.models.utils import _enforce_tensor_spec
 from mlflow.types import DataType
 from mlflow.types.schema import ColSpec, Schema, TensorSpec
 from mlflow.types.utils import _infer_schema, _get_tensor_shape
diff --git a/tests/utils/test_file_utils.py b/tests/utils/test_file_utils.py
index 43ade9717..a47c414f3 100644
--- a/tests/utils/test_file_utils.py
+++ b/tests/utils/test_file_utils.py
@@ -81,13 +81,20 @@ def test_render_and_merge_yaml(tmpdir):
             test_1: [1, 2, 3]
             test_2: {{ TEST_VAR_1 }}
             test_3: {{ TEST_VAR_2 }}
+            test_4: {{ TEST_VAR_4 }}
             """
-            + r"test_4: {{{{ ('{0}' | from_json)['key'] }}}}".format(json_file)
+            + r"test_5: {{{{ ('{0}' | from_json)['key'] }}}}".format(json_file)
         )
-
-    data = {"MY_MLFLOW_SERVER": "./mlruns", "TEST_VAR_1": ["a", 1.2], "TEST_VAR_2": {"a": 2}}
     context_yaml_file = random_file("yaml")
-    file_utils.write_yaml(str(tmpdir), context_yaml_file, data)
+    with open(tmpdir / context_yaml_file, "w") as f:
+        f.write(
+            """
+            MY_MLFLOW_SERVER: "./mlruns"
+            TEST_VAR_1: ["a", 1.2]
+            TEST_VAR_2: {"a": 2}
+            """
+            + r"TEST_VAR_4: {{{{ ('{0}' | from_json)['key'] }}}}".format(json_file)
+        )
 
     with tmpdir.as_cwd():
         result = file_utils.render_and_merge_yaml(tmpdir, template_yaml_file, context_yaml_file)
@@ -95,11 +102,13 @@ def test_render_and_merge_yaml(tmpdir):
         "MY_MLFLOW_SERVER": "./mlruns",
         "TEST_VAR_1": ["a", 1.2],
         "TEST_VAR_2": {"a": 2},
+        "TEST_VAR_4": 123,
         "steps": {"preprocess": {"train_ratio": 0.5, "experiment": {"tracking_uri": "./mlruns"}}},
         "test_1": [1, 2, 3],
         "test_2": ["a", 1.2],
         "test_3": {"a": 2},
         "test_4": 123,
+        "test_5": 123,
     }
     assert result == expected