社区模型风格

其他有用的 MLflow 风格由 MLflow 社区开发和维护，使您能够将 MLflow 模型与更广泛的机器学习库生态系统一起使用。有关更多信息，请查看以下每个社区开发的风格的描述。

MLflow VizMod

mlflow-vizmod 项目使数据科学家能够更高效地进行可视化。我们将可视化视为模型——就像机器学习模型一样——因此能够使用与MLflow相同的基础设施来跟踪、创建项目、注册和部署可视化。

安装：

pip install mlflow-vizmod

示例：

from sklearn.datasets import load_iris
import altair as alt
import mlflow_vismod

df_iris = load_iris(as_frame=True)

viz_iris = (
    alt.Chart(df_iris)
    .mark_circle(size=60)
    .encode(x="x", y="y", color="z:N")
    .properties(height=375, width=575)
    .interactive()
)

mlflow_vismod.log_model(
    model=viz_iris,
    artifact_path="viz",
    style="vegalite",
    input_example=df_iris.head(5),
)

BigML (bigmlflow)

bigmlflow 库实现了 bigml 模型风格。它支持使用 BigML 监督模型 并提供了 save_model()、log_model() 和 load_model() 方法。

安装 bigmlflow

BigMLFlow 可以从 PyPI 安装，如下所示：

pip install bigmlflow

BigMLFlow 使用

bigmlflow 模块定义了实现 save_model() 和 log_model() 方法的 flavor。它们可以用来将 BigML 模型及其相关信息保存为 MLflow 模型格式。

import json
import mlflow
import bigmlflow

MODEL_FILE = "logistic_regression.json"
with mlflow.start_run():
    with open(MODEL_FILE) as handler:
        model = json.load(handler)
        bigmlflow.log_model(
            model, artifact_path="model", registered_model_name="my_model"
        )

这些方法还将 python_function 风格添加到它们生成的 MLflow 模型中，允许通过 mlflow.pyfunc.load_model() 将模型解释为通用的 Python 函数进行推理。这个加载的 PyFunc 模型只能使用 DataFrame 输入进行评分。

# saving the model
save_model(model, path=model_path)
# retrieving model
pyfunc_model = pyfunc.load_model(model_path)
pyfunc_predictions = pyfunc_model.predict(dataframe)

你也可以使用 bigmlflow.load_model() 方法来加载带有 bigmlflow 模型风格的 MLflow 模型作为 BigML SupervisedModel。

更多信息，请参阅 BigMLFlow 文档 和 BigML 的博客。

Sktime

sktime 自定义模型风格通过 save_model() 和 log_model() 方法实现了 sktime 模型在 MLflow 格式中的记录。这些方法还将 python_function 风格添加到它们生成的 MLflow 模型中，允许通过 mlflow.pyfunc.load_model() 将模型解释为用于推理的通用 Python 函数。这个加载的 PyFunc 模型只能使用 DataFrame 输入进行评分。您还可以使用 load_model() 方法以原生 sktime 格式加载带有 sktime 模型风格的 MLflow 模型。

安装 Sktime

安装带有 mlflow 依赖的 sktime：

pip install sktime[mlflow]

使用示例

有关使用作为 pyfunc 类型加载的 sktime 模型的接口的详细信息，请参阅 sktime mlflow 文档 ，并参考 示例笔记本 以获取扩展的代码使用示例。

import pandas as pd

from sktime.datasets import load_airline
from sktime.forecasting.arima import AutoARIMA
from sktime.utils import mlflow_sktime

airline = load_airline()
model_path = "model"


auto_arima_model = AutoARIMA(sp=12, d=0, max_p=2, max_q=2, suppress_warnings=True).fit(
    airline, fh=[1, 2, 3]
)

mlflow_sktime.save_model(
    sktime_model=auto_arima_model,
    path=model_path,
)

loaded_model = mlflow_sktime.load_model(
    model_uri=model_path,
)
loaded_pyfunc = mlflow_sktime.pyfunc.load_model(
    model_uri=model_path,
)

print(loaded_model.predict())
print(loaded_pyfunc.predict(pd.DataFrame()))

MLflavors

MLflavors 包为一些目前未被考虑作为 MLflow 内置风格的流行机器学习框架添加了 MLflow 支持。类似于内置风格，您可以使用此包将您的模型保存为 MLflow 工件，从 MLflow 加载您的模型进行批量推理，并使用 MLflow 部署工具将您的模型部署到服务端点。

目前支持以下开源库：

框架	教程	类别
Orbit 是一个用于时间序列建模和预测的Python库。	MLflow-Orbit	时间序列预测
Sktime 是一个用于时间序列分析的Python库。	MLflow-Sktime	时间序列预测
StatsForecast	MLflow-StatsForecast	时间序列预测
PyOD	MLflow-PyOD	异常检测
SDV 是一个开源的数据虚拟化工具，旨在帮助用户生成合成数据。	MLflow-SDV	合成数据生成

支持框架的接口设计与许多现有的内置风格相似。特别是，用于生成预测的自定义模型加载为 pyfunc 风格的接口使用单行 Pandas DataFrame 配置参数来暴露该风格推理 API 的参数。

文档

所有风格的使用示例和API参考可以在包的 文档 中找到。

安装

从 PyPI 安装：

$ pip install mlflavors

快速入门

此示例使用合成数据集训练一个 PyOD KNN 异常检测模型。创建一个新的 MLflow 实验以记录评估指标并将训练好的模型作为工件保存，同时计算异常分数，加载训练好的模型以原生风格和 pyfunc 风格进行计算。最后，使用本地端点为模型提供实时推理服务。

将模型保存为 MLflow 工件

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}")

从 MLflow 加载模型

以原生格式从MLflow加载模型进行预测：

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

使用 pyfunc 格式从 MLflow 加载模型进行预测：

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])

使用端点提供模型服务

要在本地REST API端点上提供模型服务，请运行以下命令，其中您替换上面打印的运行ID：

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

同样地，你可以使用 MLflow 部署工具 在云端（例如 Azure ML、AWS SageMaker 等）通过一个端点来提供模型服务。打开一个新的终端并运行下面的模型评分脚本，以从提供的模型请求预测：

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())