在应用程序中创建、训练和使用 Snowflake ML 模型

本主题提供了如何使用 scikit-learn Python 包在 Snowflake Native App 中训练 Snowflake ML 模型的示例。本主题中的示例可用于根据使用者或提供商账户中的数据训练模型。

创建版本化架构来保存存储过程

在安装脚本中创建包含存储过程的版本化架构，如以下示例所示：

  1. 为存储过程创建版本化架构

    CREATE OR ALTER VERSIONED SCHEMA core;
GRANT USAGE ON SCHEMA core TO APPLICATION ROLE app_public;
    Copy

创建存储过程来创建和训练模型

  1. 为您用于训练模型的 Python 函数创建存储过程，如以下示例所示：

CREATE OR REPLACE PROCEDURE core.py_log_model(db STRING, schema STRING, mname STRING, mvname STRING)
RETURNS STRING
LANGUAGE python
RUNTIME_VERSION = 3.11
HANDLER = 'log_model'
PACKAGES = ('snowflake-snowpark-python','scikit-learn', 'snowflake-ml-python >=1.6.2', 'pandas', 'numpy')
AS '
  -- <body of the stored procedure>
';
Copy

此示例创建了一个名为 py_log_model 的存储过程，并声明了使用 scikit-learn 训练模型所需的 Python 包：

  • snowflake-snowpark-python

  • scikit-learn

  • snowflake-ml-python

  • pandas

  • numpy

  • xgboost

创建存储过程后，将以下代码添加到存储过程的主体中：

  1. 将 Python 代码添加到存储过程的主体中

import _snowflake
from snowflake.ml.registry import Registry
import pandas as pd
import numpy as np
from sklearn import datasets
from snowflake.ml.modeling.xgboost import XGBClassifier

def log_model(sp_session, mname, mvname):
    reg = Registry(session=sp_session, schema_name=''stateful_schema'')

    iris = datasets.load_iris()
    df = pd.DataFrame(data=np.c_[iris["data"], iris["target"]], columns=iris["feature_names"] + ["target"])
    df.columns = [s.replace(" (CM)", "").replace(" ", "") for s in df.columns.str.upper()]
    input_cols = ["SEPALLENGTH", "SEPALWIDTH", "PETALLENGTH", "PETALWIDTH"]
    label_cols = "TARGET"
    output_cols = "PREDICTED_TARGET"

    clf_xgb = XGBClassifier(
        input_cols=input_cols, output_cols=output_cols, label_cols=label_cols, drop_input_cols=True
    )
    clf_xgb.fit(df)
    model_ref = reg.log_model(
        clf_xgb,
        model_name=f"{mname}",
        version_name=f"{mvname}",
        options={"enable_explainability": False},
    )
    return "success"
Copy

log_model 函数执行以下操作：

  • 使用 pandas``和 ``numpy 创建 DataFrame 来提供模型的训练数据。

  • 创建 XGBoost 的实例用作数据的训练算法。

  • 调用 XGBoost 的 fit() 函数来创建模型并在数据集上对其进行训练。

  • 调用 Snowflake Model Registry 的 log_model() 函数，将模型添加到模型注册表中。

备注

应用程序创建的模型必须存储在模型注册表中。应用程序无法访问存储在暂存区的模型。

  1. 可选：要允许使用者运行存储过程来训练模型，请授予存储过程的 USAGE 权限：

    GRANT USAGE ON PROCEDURE core.py_log_model(STRING, STRING) TO APPLICATION ROLE app_public;
    Copy

创建存储过程来运行模型

  1. 为用于调用模型的 Python 函数创建存储过程。

  CREATE OR REPLACE PROCEDURE core.py_call_predict(mname STRING, mvname STRING)
  RETURNS TABLE()
  LANGUAGE python
  RUNTIME_VERSION = 3.11
  HANDLER = 'run_model'
  PACKAGES = ('snowflake-snowpark-python','scikit-learn', 'snowflake-ml-python>=1.6.2', 'pandas', 'xgboost')
  AS
'
-- <body of the stored procedure>
';
Copy

  1. 添加用于调用模型的 Python 代码

import _snowflake
from snowflake.ml.registry import Registry
import pandas as pd
from sklearn import datasets

def run_model(sp_session, mname, mvname):
  iris = datasets.load_iris()
  df = pd.DataFrame(data=iris["data"], columns=iris["feature_names"])
  df.columns = [s.replace(" (CM)", "").replace(" ", "") for s in df.columns.str.upper()]

  reg = Registry(session=sp_session, schema_name="stateful_schema")
  mv = reg.get_model(mname).version(mvname)

  pred = mv.run(df.head(10), function_name="predict")
  return sp_session.create_dataframe(pred)
Copy

run_model 函数执行以下操作：

  • 运行 load_iris() 函数来 加载虹膜机器学习数据集 (https://scikit-learn.org/stable/modules/generated/sklearn.datasets.load_iris.html)。

  • 使用 pandas 以根据虹膜数据集创建 DataFrame。

  • 运行 get_model() 函数以获取模型注册表。

  • 在模型上运行预测函数。

  • 返回结果。

  1. 可选：要允许使用者运行存储过程来训练模型，请授予存储过程的 USAGE 权限：

    GRANT USAGE ON PROCEDURE core.py_call_predict(STRING, STRING) TO APPLICATION ROLE app_public;
    Copy

运行存储过程

如果应用程序向应用程序角色授予这些存储过程的 USAGE 权限，则使用者可以调用存储过程来训练和运行模型，如以下示例所示：

CALL my_app.core.py_log_model('md1', 'V1');
Copy

此命令调用 py_log_model 存储过程来训练模型。

CALL my_app.core.py_call_predict('md1', 'V1');
Copy

此命令调用 py_call_predict 存储过程来调用模型上的预测函数。

语言: 中文