MLflavors

The MLflavors package adds MLflow support for some popular machine learning frameworks currently not considered for inclusion as MLflow built-in flavors. Similar to the built-in flavors, you can use this package to save your model as an MLflow artifact, load your model from MLflow for batch inference, and deploy your model to a serving endpoint using MLflow deployment tools.

The following open-source libraries are currently supported:

Framework	Tutorials	Category
Orbit	MLflow-Orbit	Time Series Forecasting
Sktime	MLflow-Sktime	Time Series Forecasting
StatsForecast	MLflow-StatsForecast	Time Series Forecasting
PyOD	MLflow-PyOD	Anomaly Detection
SDV	MLflow-SDV	Synthetic Data Generation

The interface design for the supported frameworks is similar to many of the existing built-in flavors. Particularly, the interface for utilizing the custom model loaded as a pyfunc flavor for generating predictions uses a single-row Pandas DataFrame configuration argument to expose the parameters of the flavor's inference API.

Documentation

Usage examples for all flavors and the API reference can be found in the package documenation.

Installation

Installing from PyPI:

$ pip install mlflavors

Quickstart

This example trains a PyOD KNN outlier detection model using a synthetic dataset. A new MLflow experiment is created to log the evaluation metrics and the trained model as an artifact and anomaly scores are computed loading the trained model in native flavor and pyfunc flavor. Finally, the model is served for real-time inference to a local endpoint.

Saving the model as an MLflow artifact

import json

import mlflow
import pandas as pd
from pyod.models.knn import KNN
from pyod.utils.data import generate_data
from sklearn.metrics import roc_auc_score

import mlflavors

ARTIFACT_PATH = "model"

with mlflow.start_run() as run:
    contamination = 0.1  # percentage of outliers
    n_train = 200  # number of training points
    n_test = 100  # number of testing points

    X_train, X_test, _, y_test = generate_data(
        n_train=n_train, n_test=n_test, contamination=contamination
    )

    # Train kNN detector
    clf = KNN()
    clf.fit(X_train)

    # Evaluate model
    y_test_scores = clf.decision_function(X_test)

    metrics = {
        "roc": roc_auc_score(y_test, y_test_scores),
    }

    print(f"Metrics: \n{json.dumps(metrics, indent=2)}")

    # Log metrics
    mlflow.log_metrics(metrics)

    # Log model using pickle serialization (default).
    mlflavors.pyod.log_model(
        pyod_model=clf,
        artifact_path=ARTIFACT_PATH,
        serialization_format="pickle",
    )
    model_uri = mlflow.get_artifact_uri(ARTIFACT_PATH)

# Print the run id wich is used below for serving the model to a local REST API endpoint
print(f"\nMLflow run id:\n{run.info.run_id}")

Loading the model from MLflow

Make a prediction loading the model from MLflow in native format:

loaded_model = mlflavors.pyod.load_model(model_uri=model_uri)
print(loaded_model.decision_function(X_test))

Make a prediction loading the model from MLflow in pyfunc format:

loaded_pyfunc = mlflavors.pyod.pyfunc.load_model(model_uri=model_uri)

# Create configuration DataFrame
predict_conf = pd.DataFrame(
    [
        {
            "X": X_test,
            "predict_method": "decision_function",
        }
    ]
)

print(loaded_pyfunc.predict(predict_conf)[0])

Serving the model to an endpoint

To serve the model to a local REST API endpoint run the command below where you substitute the run id printed above:

mlflow models serve -m runs:/<run_id>/model --env-manager local --host 127.0.0.1

Similarly, you could serve the model to an endpoint in the cloud (e.g. Azure ML, AWS SageMaker, etc.) using MLflow deployment tools. Open a new terminal and run the below model scoring script to request a prediction from the served model:

import pandas as pd
import requests
from pyod.utils.data import generate_data

contamination = 0.1  # percentage of outliers
n_train = 200  # number of training points
n_test = 100  # number of testing points

_, X_test, _, _ = generate_data(
    n_train=n_train, n_test=n_test, contamination=contamination
)

# Define local host and endpoint url
host = "127.0.0.1"
url = f"http://{host}:5000/invocations"

# Convert to list for JSON serialization
X_test_list = X_test.tolist()

# Create configuration DataFrame
predict_conf = pd.DataFrame(
    [
        {
            "X": X_test_list,
            "predict_method": "decision_function",
        }
    ]
)

# Create dictionary with pandas DataFrame in the split orientation
json_data = {"dataframe_split": predict_conf.to_dict(orient="split")}

# Score model
response = requests.post(url, json=json_data)
print(response.json())

Contributing

Contributions from the community are welcome, I will be happy to support the inclusion and development of new features and flavors. To open an issue or request a new feature, please open a GitHub issue.

Versioning

Versions and changes are documented in the changelog .

Development

To set up your local development environment, create a virtual environment, such as:

$ conda create -n mlflavors-dev python=3.9
$ source activate mlflavors-dev

Install project locally:

$ python -m pip install --upgrade pip
$ pip install -e ".[dev]"

Install pre-commit hooks:

$ pre-commit install

Run tests:

$ pytest

Build Sphinx docs:

$ cd docs
$ make html