标签: mlops

我是 mlflow 的初学者，尝试使用 Anaconda 3 在本地进行设置。我在 anaconda 中创建了一个新环境，并在其中安装了 mlflow 和 sklearn。现在我使用 jupyter Notebook 来运行 mlflow 的示例代码。

'''

import os
import warnings
import sys

import pandas as pd
import numpy as np
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
from sklearn.model_selection import train_test_split
from sklearn.linear_model import ElasticNet
from urllib.parse import urlparse
import mlflow
import mlflow.sklearn

import logging

logging.basicConfig(level=logging.WARN)
logger = logging.getLogger(__name__)

warnings.filterwarnings("ignore")
np.random.seed(40)


mlflow.set_tracking_uri("file:///Users/Swapnil/Documents/LocalPython/MLFLowDemo/mlrun")

mlflow.get_tracking_uri()

mlflow.get_experiment

#experiment_id = mlflow.create_experiment("Mlflow_demo")
experiment_id = mlflow.create_experiment("Demo3")
experiment = mlflow.get_experiment(experiment_id)
print("Name: {}".format(experiment.name))
print("Experiment_id: {}".format(experiment.experiment_id))
print("Artifact Location: {}".format(experiment.artifact_location)) …

DVC 使用 git commits 来保存实验并在实验之间导航。

是否可以避免在 CI/CD 中进行自动提交（dvc repro在 CI/CD 端之后保存数据工件）。

我正在创建一个 mlflow 实验，它将逻辑回归模型以及指标和工件记录在一起。

import mlflow
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import precision_recall_fscore_support

with mlflow.start_run(run_name=run_name, experiment_id=experiment_id):

        logreg = LogisticRegression()
        logreg.fit(x_train, y_train)
        print('training over', flush=True)
        y_pred = logreg.predict(x_test)
        mlflow.sklearn.log_model(logreg, "model")
   
        mlflow.log_metric("f1", precision_recall_fscore_support(y_test, y_pred, average='weighted')[2])
        mlflow.log_artifact(x_train.to_csv('train.csv')

对于某些数据 ( x_train, y_train, x_test, y_test)

有没有办法访问此 run_name 的特定实验 ID 的工件并读取train.csv和model？

我可以使用 mlflow 客户端加载模型的特定版本：

import mlflow

model_version = 1

model = mlflow.pyfunc.load_model(
    model_uri=f"models:/c3760a15e6ac48f88ad7e5af940047d4/{model_version}"
)

但是有没有办法加载最新的模型版本呢？

我正在使用 Pipelines 和组件在 Vertex AI 中开发复杂的管道。我想导入一些为此用例开发的自定义模块和函数。不幸的是，我无法弄清楚如何在不创建临时 Docker 映像或不在 PyPi 等公共存储库上发布我的代码的情况下在代码中导入这些自定义函数。

在每个组件中粘贴这些自定义函数的代码有两个痛点：

1. 代码变得庞大且难以阅读
2. 该函数的代码完全失去了可维护性，因为在每个小的更改中，我都必须为每个组件复制它。

我正在使用 mlflow 来注册我的模型。当工件从本地加载到 S3 存储桶时，我尝试使用“场景 4”。

1. 将 S3 存储桶的凭证添加到 .aws/credentials

2. 设置端点和 mlflow URI：

   os.environ["MLFLOW_S3_ENDPOINT_URL"]='https://storage.yandexcloud.net' os.environ["MLFLOW_TRACKING_URI"]='http://:8000'

3. 通过 mlflow 将模型记录到 S3：

   导入mlflow导入mlflow.sklearn mlflow.set_experiment（“我的”）... mlflow.sklearn.log_model（模型，artifact_path =“models_mlflow”）

但得到错误：

MlflowException: API request to http://<IP>:8000/api/2.0/mlflow-artifacts/artifacts/6/95972bcc493c4a8cbd8432fea4cc8bac/artifacts/models_mlflow/model.pkl failed with exception HTTPConnectionPool(host='62.84.121.234', port=8000): Max retries exceeded with url: /api/2.0/mlflow-artifacts/artifacts/6/95972bcc493c4a8cbd8432fea4cc8bac/artifacts/models_mlflow/model.pkl (Caused by ResponseError('too many 503 error responses'))

我想将 Mlflow 设置为具有以下组件：

• 后端存储（本地）：在本地使用 SQLite 数据库来存储 Mlflow 实体（run_id、params、metrics...）
• 工件存储（远程）：使用 Azure Data Lake Storage Gen2 上的 Blob 存储来存储与模型相关的输出文件（版本化数据集、序列化模型、图像等）
• 跟踪服务器：通过使用类似于此命令的命令

z

mlflow server --backend-store-uri sqlite:///C:\sqlite\db\mlruns.db --default-artifact-root wasbs://container-name@storage_account_name.blob.core.windows.net/mlartifacts -h 0.0.0.0 -p 8000

其中 mlruns.db 是我在 SQLite 中创建的数据库（在 db 文件夹内），mlartifacts 是我在 blob 容器内创建的用于接收所有输出文件的文件夹。

我运行此命令，然后运行 ​​mlflow（或运行 kedro，因为我使用的是 Kedro），但几乎没有任何反应。该数据库填充了 12 个表，但全部为空，而数据湖内没有任何反应。

我想要的应该类似于文档中的场景 4 。

对于神器商店，我找不到详细的说明。我尝试在这里查看 Mlflow 的文档，但这不是很有帮助（我仍然是初学者）。他们说：

MLflow 需要 AZURE_STORAGE_CONNECTION_STRING、AZURE_STORAGE_ACCESS_KEY 环境变量中的 Azure 存储访问凭据，或者将凭据配置为 DefaultAzureCredential()。班级可以接他们。

然而，即使添加环境变量，数据湖中似乎也没有存储任何内容。我创建了两个环境变量（在 Windows 10 上）：

• AZURE_STORAGE_ACCESS_KEY = wasbs://container-name@storage_account_name.blob.core.windows.net/mlartifacts

• AZURE_STORAGE_CONNECTION_STRING = DefaultEndpointsProtocol=https；AccountName=storagesample；AccountKey=。我通过在 Azure 门户上遵循以下路径获得它：存储帐户/访问密钥/连接字符串（采用密钥 2 之一）。

他们还指出：

此外，必须单独运行 pip install azure-storage-blob（在客户端和服务器上）才能访问 …

我正在尝试将模型保存到 MLFlow，但由于我有一个自定义预测管道来检索数据，因此我需要将额外的元数据保存到模型中。

我尝试使用我的自定义签名类，它可以正确完成工作并将模型与额外的元数据保存在 MLModel 文件（YAML 格式）中。但是当想要从 MLFlow 注册表加载模型时，签名并不容易访问。

mlflow.sklearn.log_model(model, "model", signature = signature)

我还尝试在 log_model 函数中保存额外的字典，但它将它保存在 conda.yaml 文件中：

mlflow.sklearn.log_model(model, "model", {"metadata1":"value1", "metadata2":"value2"})

我应该自己做味道吗？还是我自己的 Model 继承？我在这里看到PyFuncModel 收到一些元数据类和解决此问题的实现，但我不知道应该在实验脚本上将我自己的实现传递给 PyFuncModel 的位置。这是一个最小的例子：

import mlflow
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression

metadata_dic = {"metadata1": "value1", 
                "metadata2": "value2"}

X = np.array([[-2, -1, 0, 1, 2, 1],[-2, -1, 0, 1, 2, 1]]).T
y = np.array([0, 0, 1, 1, 1, 0])

X = pd.DataFrame(X, columns=["X1", "X2"])
y = pd.DataFrame(y, columns=["y"]) …

这可能是一个非常具体的问题，但无论如何我都会尝试。

我想在模型注册表中为给定的模型版本显式设置阶段列：

该图片来自文档，仅当您运行他们提供的示例 SageMaker Projects MLOps 模板时才会设置。当我手动创建Model Package（即Model Version）时，该列保持为空。我该如何设置？我该调用什么API？

另外，浏览模型版本历史记录的文档中有如下一句话

我们如何手动发送确切的事件（“部署到阶段 XYZ”）？

我已经彻底检查了 SageMaker MLOps 项目生成的所有文件（CodeBuild Builds、CodePipeline、CloudFormation、各种.py文件、SageMaker Pipeline），但找不到对该事件的任何直接且显式的调用。

我认为它可能以某种方式连接到标签sagemaker:deployment-stage，但我已经将其设置为Endpoint、EndpointConfiguration和Model，但没有成功。我还尝试盲目调用UpdateModelPackage API 并Stage在CustomerMetadataProperties. 再次 - 没有运气。

我在“活动”选项卡中得到的唯一信息是给定的模型版本已部署到推理端点：

在返回 Pipeline 的函数内部，定义了 Parameter，例如（取自此处）

def get_pipeline(...):
   
    foo = ParameterString(
        name="Foo", default_value="foo"
    )

   # pipeline's steps definition here
   step = ProcessingStep(name=...,
                         job_arguments=["--foo", foo]
   )

   return pipeline = Pipeline(
        name=pipeline_name,
        parameters=[...],
        steps=[...],
        sagemaker_session=sagemaker_session,
    )

我知道我可以通过简单地调用来访问参数的默认值foo.default_value，但是当默认值在运行时被覆盖时，我如何访问它的值，例如通过使用

pipeline.start(parameters=dict(Foo='bar'))

？

我的假设是，在这种情况下，我不想读取默认值，因为它已被覆盖，但参数 API非常有限，并且没有提供name和所期望的任何内容default_value。