Crie, treine e use um modelo do Snowflake ML em um aplicativo

Este tópico fornece um exemplo de como treinar um modelo do Snowflake ML em um Snowflake Native App usando o pacote scikit-learn Python. O exemplo deste tópico pode ser usado para treinar modelos nos dados das contas do consumidor ou do provedor.

Criar um esquema com versão para manter os procedimentos armazenados

No script de configuração, crie um esquema com versão que contenha o procedimento armazenado, conforme mostrado no exemplo a seguir:

  1. Criar um esquema com versão para o procedimento armazenado

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

Criar um procedimento armazenado para criar e treinar um modelo

  1. Crie um procedimento armazenado para a função Python que você está usando para treinar um modelo, conforme mostrado no exemplo a seguir:

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

Este exemplo cria um procedimento armazenado chamado py_log_model e declara os pacotes Python necessários para treinar um modelo usando scikit-learn:

  • snowflake-snowpark-python

  • scikit-learn

  • snowflake-ml-python

  • pandas

  • numpy

  • xgboost

Depois de criar um procedimento armazenado, adicione o seguinte código ao corpo do procedimento armazenado:

  1. Adicionar o código Python ao corpo do procedimento armazenado

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

A função log_model executa o seguinte:

  • Usa pandas e numpy para criar um DataFrame para servir como dados de treinamento para o modelo.

  • Cria uma instância do XGBoost para servir como algoritmo de treinamento para os dados.

  • Chama a função fit() do XGBoost para criar um modelo e treiná-lo no conjunto de dados.

  • Chama a função log_model() do Snowflake Model Registry para adicionar o modelo ao registro de modelos.

Nota

Os modelos criados por um aplicativo devem ser armazenados em um registro de modelos. Os aplicativos não podem acessar modelos armazenados em um estágio.

  1. Opcional: para permitir que os consumidores executem o procedimento armazenado para treinar o modelo, conceda o privilégio USAGE no procedimento armazenado:

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

Criar um procedimento armazenado para executar um modelo

  1. Crie um procedimento armazenado para a função Python que você usa para chamar o modelo.

  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. Adicionar o código Python que você usa para chamar o modelo

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

A função run_model faz o seguinte:

  • Executa a função load_iris() para carregar o conjunto de dados de aprendizado de máquina iris.

  • Usa o pandas para criar um DataFrame com base no conjunto de dados iris.

  • Executa a função get_model() para obter o registro do modelo.

  • Executa a função de previsão no modelo.

  • Retorna o resultado.

  1. Opcional: para permitir que os consumidores executem o procedimento armazenado para treinar o modelo, conceda o privilégio USAGE no procedimento armazenado:

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

Executar os procedimentos armazenados

Se o aplicativo conceder o privilégio USAGE nesses procedimentos armazenados a uma função de aplicativo, os consumidores poderão chamar os procedimentos armazenados para treinar e executar os modelos, conforme mostrado nos exemplos a seguir:

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

Esse comando chama o procedimento armazenado py_log_model para treinar o modelo.

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

Esse comando chama o procedimento armazenado py_call_predict para chamar a função de previsão no modelo.

Linguagem: Português