Snowflake Snowpark：全面入门

什么是 Snowflake Snowpark？

传统的机器学习流程通常是把数据库中的数据搬到模型所在的位置。随着近年来 AI 的兴起以及数据集规模的激增，这种做法越来越不切实际。

为了完成清洗、分析和模型训练，需要将 TB 级数据从数据库传输到客户端应用。这趟看似无害的“往返旅程”会浪费大量宝贵的资源。基于此，越来越多公司选择使用库内（in-database）技术，尽量减少数据移动并让数据操作更顺畅。

市面上最好的库内技术之一是由 Snowflake Cloud 提供的 Snowpark。Snowpark 是一组库和运行时，允许您在 Snowflake Cloud 中安全地运行编程语言，在与 Snowflake 数据库相同的环境中开发数据管道和机器学习模型。

在本教程中，我们将介绍 Snowpark 的基础知识以及如何在您的项目中使用它。我们假设您已经熟悉 SQL 和 Snowflake——如果您需要先补充这部分内容，可以学习我们的 SQL 基础技能路径，或阅读这篇Snowflake 新手教程。

让我们开始吧！

为什么选择 Snowflake Snowpark？

Snowflake Snowpark 是一组库和运行时，允许您安全地使用 Python、Java、Scala 等编程语言，直接在 Snowflake 云平台内处理数据。

这消除了为处理而将数据移出 Snowflake 的需求，从而提升效率与安全性。其主要优势包括：

在 Snowflake 内处理数据：用您熟悉的语言编写代码来操作和分析 Snowflake 中的 SQL 数据库，并在 Snowflake 的安全环境中执行。无需将数据迁移到其他环境。
性能更优：通过在 Snowflake 内直接处理数据，Snowpark 可利用平台的弹性与无服务器架构，实现高效处理。
降低成本与技术开销：由于 Snowflake 基本提供了所需资源，您无需分别管理计算与存储平台。
用熟悉的工具、在喜欢的环境里工作：Snowpark 的 API 允许您从任意环境（如 Jupyter 或 VSCode）连接到 SQL 数据库，构建数据管道和 ML 应用。更棒的是，您可以在 Snowpark 框架的配合下，继续使用自己喜爱的库，如 Pandas、Scikit-learn、XGBoost 等。

简而言之，Snowpark 为开发者提供了一种强大而简洁的方式，直接在 Snowflake Cloud 中构建数据管道、机器学习解决方案与数据驱动应用。

Snowpark 入门

本教程的最终目标是使用 Snowpark，在 Snowflake 数据库的一张表上训练并调参一个模型。为此，我们将先完成以下步骤：

创建包含相关库的虚拟环境。
将示例数据导入 Snowflake。
创建 Snowpark 会话以连接至 Snowflake。
将已导入的数据加载到会话中。

创建虚拟环境

本教程中，我们将使用一个新的 conda 环境：

$ conda create -n snowpark python==3.10 -y
$ conda activate snowpark技术支持

另外提醒一下，如果您是在新安装的 conda 环境中运行，激活 snowpark 环境前需要先执行 $conda init。

激活后，需要安装以下库：

$ pip install snowflake-snowpark-python #The Snowpark API
$ pip install pandas pyarrow numpy matplotlib seaborn技术支持

如果您将使用 Jupyter，还需安装 ipykernel，并运行以下命令将当前环境添加为一个 Jupyter 内核：

$ pip install ipykernel
$ ipython kernel install --user --name=snowpark技术支持

现在导入一些后续会用到的通用库：

import warnings
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
warnings.filterwarnings("ignore")技术支持

将示例数据导入 Snowflake

本教程将使用 Seaborn 的 Diamonds 数据集——您可以直接从我的 GitHub下载 CSV 文件。

登录Snowflake 应用后，按照下方 GIF 的步骤，将该数据集导入一张新的数据库表（若您还没有账号，请先创建新账号）：

在实际工作中，您很少会以 CSV 文件的形式将表导入 Snowflake。多数情况下，您会在公司数据库管理员授予的不同访问级别下，直接使用已有数据库。

创建 Snowpark 会话

接下来需要在 Snowpark 的 API 与 Snowflake 云之间建立连接，以查询我们的数据库。该连接需要以下 Snowflake 凭据：账户名、用户名和密码。

下图展示了如何从 Snowflake 控制台获取您的用户名和账户名：

创建一个名为 config.py 的独立文件，包含一个名为 credentials 的字典，格式如下：

credentials = (
   {
       "account": "3-4",  # Combine 3 and 4 with a hyphen
       "username": "bexgboost",  # Your username in lowercase
       "password": "your_password",  # Your Snowflake password
   },
)技术支持

出于安全考虑，我们将凭据存放在单独的文件中。将该文件加入 .gitignore 可以确保您的 Snowflake 凭据不会被意外泄露。

现在，导入 credentials 字典和 Snowpark 的 Session 类：

from config import credentials
from snowflake.snowpark import Session技术支持

为建立连接，我们将使用 Session.builder.configs.create() 方法：

connection_parameters = {
   "account": credentials["account"],
   "user": credentials["username"],
   "password": credentials["password"],
}

new_session = Session.builder.configs(connection_parameters).create()
new_session.get_current_user()技术支持

'"bexgboost"'技术支持

如果打印出了您的用户名，说明首次会话已成功创建！

将已导入数据加载到会话

new_session 对象可以访问您在 Snowflake 账户中拥有（或被授予权限）的所有内容。

在导入数据之前，第一步是告知会话我们将使用上一节创建的 test_db 数据库：

new_session.sql("USE DATABASE test_db;").collect()技术支持

[Row(status='Statement executed successfully.')]技术支持

我们通过 .sql 方法完成这一步，它允许我们运行任何与 Snowflake SQL 兼容的查询。

现在，我们可以使用会话对象的 table 方法加载 test_db 的 diamonds 表：

diamonds_df = new_session.table("diamonds")
diamonds_df.show(5)技术支持

.show() 方法相当于 Pandas 的 DataFrame.describe()，可以让我们确认已成功连接至该表。

没错——我们只是连接到了 diamonds 表。从其大小也能看出 diamonds_df 对象并未持有任何数据：

import sys
sys.getsizeof(diamonds_df)技术支持

48技术支持

它只有 48 字节，而实际上应该超过 3 MB。我们再花点时间理解这是为什么。

理解 Snowpark DataFrame

到目前为止，我们已完成以下步骤：

创建包含相关库的虚拟环境。
将示例数据导入 Snowflake。
创建 Snowpark 会话以连接至 Snowflake。

现在让我们来讨论 Snowpark DataFrame！

Snowpark DataFrame 采用惰性计算

Pandas DataFrame 使用的是您的本机内存（RAM），也就是说它们驻留在您的机器上。相较之下，Snowpark DataFrame 虽然可以在编码环境中查看其表示，但其实是驻留在 Snowflake 的云平台上。数据保存在云端，无需下载到本地即可对其进行操作。

另一个重要特性是 Snowpark DataFrame 采用惰性计算。这意味着在您明确要求执行之前，它们不会对数据执行任何操作。

取而代之的是，它们会构建所需操作的逻辑表示，然后将其转换为优化过的 SQL 语句，由 Snowflake 来执行。

相比之下，Pandas 会立即执行操作——也就是“及早执行”。

从性能角度看，惰性计算通常显著快于及早执行。由于惰性 DataFrame 利用的是 Snowflake 强大的弹性计算资源，它们可以将操作下推到数据库层面，从而运行得更快。

为了更好地理解惰性计算与及早执行的差异，可以将我们在 Snowflake 的数据想象成一座巨大的图书馆。我们有以下两种情景：

及早执行（Pandas DataFrame）：我们把需要的所有书都搬到桌上（本地机器），然后逐本分析。
惰性计算（Snowpark DataFrame）：我们告诉图书管理员（Snowpark）需要哪些书、要做什么分析，然后由图书管理员高效地检索并分析相关信息。

何时使用 Snowpark DataFrame

既然 Snowpark 可以转换为 Pandas DataFrame，那么我们该在何时选择其一？

pandas_diamonds = diamonds_df.to_pandas()
pandas_diamonds.head()技术支持

答案很大程度取决于数据集的大小。

如果您的数据集很小（比如 Diamonds 数据集），您可以将其下载到本地并交给 Pandas 处理（这正是 DataFrame.to_pandas() 方法所做的）。

但在今天的数据规模下，事情很容易失控，下载数据库可能要等上几个小时。并且别忘了，这是一张“往返票”——任何想要保留的新数据也必须传回去。这正是学习 Snowpark DataFrame API 的价值所在。

另一种方式是使用 Session.sql() 方法，它允许我们执行任何与 Snowflake 兼容的 SQL 表达式（见下例）。不过我强烈建议掌握 Snowpark DataFrame API，因为它内置了 100 多个 SQL 函数与表达式的函数封装。

result = new_session.sql(
   """
   SELECT PRICE, CUT FROM DIAMONDS LIMIT 10
"""
)

type(result)
技术支持

snowflake.snowpark.dataframe.DataFrame技术支持

result.show()技术支持

Snowpark DataFrame 的变换函数

您在 Snowpark 中的大部分时间都会花在变换函数上。它们位于以下子模块下：

# The convention is to import it as F
import snowflake.snowpark.functions as F技术支持

F 下的所有函数（可通过 dir(F) 列出）都用于指定 DataFrame 应如何变换。来看一个示例：

F.upper(F.col("CUT"))技术支持

这里，col 函数创建对给定列的引用并返回一个 Column 对象。然后，.upper() 将 CUT 列的值转换为大写：

diamonds_df.show(5)技术支持

但如果我们查看 diamonds 数据集，会发现 CUT 列中的值仍有小写字母。为什么？因为 F.upper(F.col("CUT")) 表达式等价于下面的 SQL 查询：

SELECT UPPER(CUT);技术支持

这里缺了什么？

当然是指定应用变换的表名的 FROM 子句！

因此，我们必须将 F.upper(F.col("CUT")) 表达式与 diamonds_df 对象结合起来。主要有三种方式：

针对变换或选择列的表达式，放入 .select()
针对选择行的表达式，放入 .filter()
针对创建新列的表达式，放入 .with_column()

由于我们的表达式是列变换，因此传给 .select()：

our_expression = F.upper(F.col("CUT"))
diamonds_df.select(our_expression).show()技术支持

我们已成功改变了这些值！

现在来写一个筛选行的表达式：

diamonds_df.filter(F.col("PRICE") > 10000).count()技术支持

5222技术支持

我们发现数据集中有 5000 多颗钻石的价格超过 1 万美元。再试试创建新列：

(
diamonds_df.with_column("PRICE_SQUARED", F.col("PRICE") ** 2)
.select("PRICE", "PRICE_SQUARED")
.show(5)
)技术支持

上面的例子也表明可以进行方法链式调用——我们创建了一个名为 “PRICE_SQUARED” 的新列。可以看到，Snowpark DataFrame 的函数与 Pandas 十分类似。

总之，如果您在寻找某个 Pandas 函数在 Snowpark 中的替代品，它要么在 F 下，要么是 Snowpark DataFrame 的方法。您可以通过 dir(F) 或 dir(diamonds_df) 查看，或阅读Snowpark Python 参考文档。

在 Snowpark DataFrame 上进行 EDA

探索性数据分析（EDA）的重要一环是可视化。幸运的是，我们不需要为此依赖 Snowpark，可以继续使用现代数据栈。

思路是这样的：EDA 主要是发掘总体趋势和洞见，我们常常可以用一个样本来代替整个数据集。下载具有代表性的样本后，将其转换为 Pandas DataFrame，就能使用熟悉的 Matplotlib 或 Seaborn。

先进行转换：

sample = diamonds_df.sample(0.25).to_pandas()
sample.head()技术支持

上面我们将 25% 的数据集下载为一个 Pandas DataFrame。接下来把列名转换为小写：

sample.columns = [col.lower() for col in sample.columns]
sample.columns技术支持

Index(['carat', 'cut', 'color', 'clarity', 'depth', 'table', 'price', 'x', 'y', 'z'], dtype='object')技术支持

现在，您就可以像对任何数据集那样进行 EDA 了。下面我们探索一些思路，先生成一个散点图来可视化价格与克拉的关系：

#Remember we imported seaborn as sns
sns.scatterplot(data=sample, x="price", y="carat", s=1)
plt.title("Diamond prices vs. carat")
plt.show()技术支持

接下来绘制相关矩阵，查看数值特征两两之间的相关性：

corr_matrix = sample.corr(numeric_only=True)
sns.heatmap(corr_matrix, center=0, square=True, annot=True
plt.title("The correlation matrix of Diamond numeric features")
plt.show()技术支持

我们还可以为钻石切工类别绘制柱状图：

cut_counts = sample.groupby("cut")["cut"].count()
cut_counts.plot(kind="bar")
plt.title("Category counts of diamond cuts")
plt.xlabel("Diamond cut")
plt.show()技术支持

为了确认在整个数据集上这些图形大致相同，我们将运行上面柱状图的 SQL 版本。

首先，编写按 CUT 对 diamonds 数据集分组的 SQL 表达式：

result = new_session.sql(
   """
   SELECT cut, COUNT(*) AS count
     FROM diamonds
    GROUP BY cut;
"""
)
技术支持

现在将结果转换并下载为 Pandas DataFrame：

result_pd = result.to_pandas()
result_pd技术支持

这一次，我们使用 plt.bar() 来绘图：

plt.bar(x=result_pd["CUT"], height=result_pd["COUNT"])
plt.show()技术支持

虽然柱子的顺序不同，但并排查看可以发现传达的信息是一致的。

在 Snowpark 中训练机器学习模型

我们已经准备好学习如何在 Snowpark 中训练机器学习模型了！

我们从清洗数据集开始。

在 Snowpark 中清洗数据

首先，确保所有列的数据类型正确：

list(diamonds_df.schema)技术支持

[StructField('CARAT', DecimalType(38, 2), nullable=True),
StructField('CUT', StringType(16777216), nullable=True),
StructField('COLOR', StringType(16777216), nullable=True),
StructField('CLARITY', StringType(16777216), nullable=True),
StructField('DEPTH', DecimalType(38, 1), nullable=True),
StructField('"table"', DecimalType(38, 1), nullable=True),
StructField('PRICE', LongType(), nullable=True),
StructField('X', DecimalType(38, 2), nullable=True),
StructField('Y', DecimalType(38, 2), nullable=True),
StructField('Z', DecimalType(38, 2), nullable=True)]技术支持

table 列的命名有点古怪，我们把它转换为大写的 TABLE_（在末尾加下划线以避免与 SQL 关键字冲突）：

diamonds_df = diamonds_df.with_column_renamed('"table"', "TABLE_")
diamonds_df.columns技术支持

['CARAT', 'CUT', 'COLOR', 'CLARITY', 'DEPTH', 'TABLE_', 'PRICE', 'X', 'Y', 'Z']技术支持

我们需要将数值特征从 DecimalType 改为 DoubleType，因为 Snowpark 目前尚不支持 decimals。

from snowflake.snowpark.types import DoubleType

numeric_features = ["CARAT", "X", "Y", "Z", "DEPTH", "TABLE_"]
for col in numeric_features:
   diamonds_df = diamonds_df.with_column(col, diamonds_df[col].cast(DoubleType()))

list(diamonds_df.select(*numeric_features))技术支持

[Column("CARAT"),
Column("X"),
Column("Y"),
Column("Z"),
Column("DEPTH"),
Column("TABLE_")]技术支持

在上述代码片段中，我们再次使用了 .with_column() 方法。另一个新方法是 .cast()，它属于 Snowpark DataFrame 的方法。

Snowpark 还要求在编码前，所有文本特征必须为大写且词间没有空格。目前 CUT 特征的取值不满足这一要求，我们将通过函数变换来修正：

import snowflake.snowpark.functions as F

def remove_space_and_upper(df):
   df = df.with_column("CUT", F.upper(F.replace(F.col("CUT"), " ", "_")))

   return df

diamonds_df = remove_space_and_upper(diamonds_df)
技术支持

现在数据已无清洗方面的问题。我们可以将其作为新表保存回 Snowflake：

diamonds_df.write.mode("overwrite").save_as_table("diamonds_cleaned")技术支持

务必使用 overwrite 模式，因为您可能需要回头修改某些清洗操作并再次保存。

在 Snowpark 中为建模做准备

本节我们将处理剩余的预处理问题，避免阻碍模型训练。先加载上一节保存的清洗后表：

clean_df = new_session.table("diamonds_cleaned")
clean_df.show()技术支持

很好！

接着，从 snowflake.ml 命名空间导入 preprocessing 和 pipeline 子模块。

import snowflake.ml.modeling.preprocessing as snowml
from snowflake.ml.modeling.pipeline import Pipeline技术支持

需要注意的是，Snowpark 的 DataFrame API 位于 snowflake.snowpark 下，而 Snowpark 的 ML 模块位于 snowflake.ml 下。一开始可能有点混淆，请耐心适应。

在下面的代码中，我们将：

列出需要编码的类别特征名。
为输出特征命名，因为 Snowpark 会将编码后的特征作为新列添加，而不是替换旧列。
定义有序类别的正确顺序；在 Diamonds 数据集中，所有类别都具备内在顺序，从左到右价格越高。

# List all the features for processing
cat_cols = ["CUT", "COLOR", "CLARITY"]
cat_cols_encoded = ["CUT_OE", "COLOR_OE", "CLARITY_OE"]

# We already have numeric_features

# List the correct ordering of categorical features
categories = {
   "CUT": np.array(["IDEAL", "PREMIUM", "VERY_GOOD", "GOOD", "FAIR"]),
   "CLARITY": np.array(
       ["IF", "VVS1", "VVS2", "VS1", "VS2", "SI1", "SI2", "I1", "I2", "I3"]
   ),
   "COLOR": np.array(["D", "E", "F", "G", "H", "I", "J"]),
}技术支持

现在，我们来构建一个类似 Scikit-learn 的管道：

# Build the pipeline
preprocessing_pipeline = Pipeline(
   steps=[
       (
           "OE",
           snowml.OrdinalEncoder(
               input_cols=cat_cols, output_cols=cat_cols_encoded, categories=categories
           ),
       ),
       (
           "SS",
           snowml.StandardScaler(
               input_cols=numeric_features, output_cols=numeric_features
           ),
       ),
   ]
)
技术支持

Pipeline 类接受一个步骤列表，每个步骤都是一个预处理类。这里我们使用 OrdinalEncoder 将类别编码为 0、1、3 等，用 StandardScaler 规范化数值特征。

最后，我们使用 joblib 将管道保存在本地，并在整个数据集上测试：

import joblib
PIPELINE_FILE = "pipeline.joblib"

# Pickle locally first
joblib.dump(preprocessing_pipeline, PIPELINE_FILE)

transformed_diamonds_df = preprocessing_pipeline.fit(clean_df).transform(clean_df)

# transformed_diamonds_df.show() - commented because of long output技术支持

该代码片段运行无误，说明我们的管道工作正常，这也意味着我们终于可以训练模型了。

在 Snowpark 中训练 XGBoost 模型

为了训练模型，我们再次列出特征名和目标名：

# Define the columns again
cat_cols_encoded = ["CUT_OE", "COLOR_OE", "CLARITY_OE"]
numeric_features = ["CARAT", "X", "Y", "Z", "DEPTH", "TABLE_"]

label_cols = ["PRICE"]  # Must be a list
output_cols = ["PRICE_PREDICTED"]  # Required in snowpark技术支持

现在使用 DataFrame.random_split() 划分数据，并指定权重来表示训练集与测试集的比例：

# Split the data
train_df, test_df = diamonds_df.random_split(weights=[0.8, 0.2], seed=42)技术支持

随后，加载管道并对训练集与测试集进行拟合与变换。

# Load the pre-processing pipeline locally
pipeline = joblib.load("pipeline.joblib")

# Apply it to both dataframes
train_df_transformed = pipeline.fit(train_df).transform(train_df)
test_df_transformed = pipeline.transform(test_df)技术支持

需要注意的是，在测试集上只调用 .transform()，以避免数据泄漏。

接着，从 ml.modeling.xgboost 子模块中初始化一个 XGBoost 回归模型：

# Snowpark has models from scikit-learn and lightgbm too
from snowflake.ml.modeling.xgboost import XGBRegressor

# Initialize
regressor = XGBRegressor(
   input_cols=cat_cols_encoded + numeric_features,
   label_cols=label_cols,
   output_cols=output_cols,
)技术支持

XGBRegressor 类需要提供所有输入名、所有目标名以及输出列名。提供后，调用 .fit() 开始训练：

# Train
regressor.fit(train_df_transformed)技术支持

如果您使用的是 Snowflake 免费账号，受限于计算资源且无 GPU，训练可能会比较慢。

训练完成后，我们可以生成预测：

# 预测

train_preds = regressor.predict(train_df_transformed)
test_preds = regressor.predict(test_df_transformed)技术支持

来看一下预测结果：

train_preds.select("PRICE", "PRICE_PREDICTED").show(5)技术支持

我们现在需要评估初始模型的表现。Snowpark 在其 ml.modeling.metrics 子模块下包含了来自 Scikit-learn 的数十种指标，我们将其导入为 M：

import snowflake.ml.modeling.metrics as M技术支持

然后两次运行 mean_squared_error 来计算训练集和测试集的均方根误差（RMSE）（下面 squared=False 表示返回 RMSE 而非 MSE）：

rmse_train = M.mean_squared_error(
   df=train_preds,
   y_true_col_names=label_cols,
   y_pred_col_names=output_cols,
   squared=False,
)

rmse_test = M.mean_squared_error(
   df=test_preds,
   y_true_col_names=label_cols,
   y_pred_col_names=output_cols,
   squared=False,
)

print(f"Train RMSE score for XGBRegressor: {rmse_train:.4f}")
print(f"Test RMSE score for XGBRegressor: {rmse_test:.4f}")技术支持

Train RMSE score for XGBRegressor: 371.9664
Test RMSE score for XGBRegressor: 542.1566技术支持

我们的模型平均相差 542 美元，并且可能存在一定过拟合，因为训练与测试 RMSE 的差距较大。

让我们通过调参来尝试解决这些问题。

在 Snowpark 中进行超参数调优

目前，Snowpark 提供两种超参数搜索类：

GridSearchCV：通过交叉验证穷举所有超参数组合。
RandomizedSearchCV：通过交叉验证在给定超参数分布中进行随机搜索。

XGBoost 大约有十来个可提升性能的超参数（您可以在这篇Python 中使用 XGBoost的教程中了解更多）。这促使我们选择随机搜索。就个人而言，我更愿意使用采用贝叶斯搜索的 Optuna，但在 Snowpark 中目前还无法使用。

现在来定义搜索：

from snowflake.ml.modeling.model_selection import RandomizedSearchCV

rscv = RandomizedSearchCV(
   estimator=XGBRegressor(),
   param_distributions={
       "n_estimators": [2000],
       "max_depth": list(range(3, 13)),
       "learning_rate": np.linspace(0.1, 0.5, num=10),
   },
   n_jobs=-1,
   scoring="neg_mean_squared_error",
   input_cols=numeric_features + cat_cols_encoded,
   label_cols=label_cols,
   output_cols=output_cols,
   n_iter=10,
)

rscv.fit(train_df_transformed)
技术支持

我们将基学习器数量固定为 2000，只调优 max_depth 和 learning_rate。rscv 的默认参数 n_iter=10 表示将执行 10 次搜索，每次随机选择一组参数组合。若要获得更准确的结果，最好将该参数设得更大一些。

现在评估最优模型的表现。代码与之前相同，但我们使用 rscv 对象来代替 regressor：

# Predict
train_preds = rscv.predict(train_df_transformed)
test_preds = rscv.predict(test_df_transformed)
rmse_train = M.mean_squared_error(
   df=train_preds,
   y_true_col_names=label_cols,
   y_pred_col_names=output_cols,
   squared=False,
)

optimal_rmse_test = M.mean_squared_error(
   df=test_preds,
   y_true_col_names=label_cols,
   y_pred_col_names=output_cols,
   squared=False,
)

print(f"Train RMSE score for optimal model: {rmse_train:.4f}")
print(f"Test RMSE score for optimal model: {rmse_test:.4f}")技术支持

Train RMSE score for optimal model: 224.2405
Test RMSE score for optimal model: 572.0517技术支持

训练 RMSE 下降了，但测试 RMSE 更大了。这说明我们仍在过拟合，需要扩大搜索范围，并加入其他用于剪枝的参数，以约束 XGBoost 底层的决策树。

这部分就留给您来完成了。

在 Snowpark 中保存最佳模型

如果现在关闭会话，我们将丢失已调优的模型。要保存它，我们将使用 Snowpark 的原生模型注册表（model registry），它是一个虚拟存储，用于保存任意模型及其元数据。

注册表通过 Registry 类提供，需要传入当前会话的数据库与架构名。同时还需要为项目指定一个名称，以避免与其他项目的模型发生冲突：

# Set up for Registry
from snowflake.ml.registry import Registry

# Get the current db and schema name
db_name = new_session.get_current_database()
schema_name = new_session.get_current_schema()

# Define global model name for the project
model_name = "diamond_prices_regression"

# Initialize a registry to log models
registry = Registry(session=new_session, database_name=db_name, schema_name=schema_name)技术支持

我们将使用 Registry.log_model() 方法来保存模型：

# Get sample data to pass into registry for schema
sample = train_df.select(cat_cols_encoded + numeric_features).limit(50)

# Log the first model
v0 = registry.log_model(
   model_name=model_name,
   version_name="v0",
   model=regressor,
   sample_input_data=sample,
)技术支持

我们可以使用 .set_metric() 方法为模型记录指标（可以记录任意多个）。我们还会添加一条注释来描述该模型：

# Add the models RMSE score
v0.set_metric(metric_name="RMSE", value=rmse_test)

# Add a description
v0.comment = "The first model to predict diamond prices"技术支持

对于随机搜索得到的最优模型，同样进行记录。记录指标时别忘了将值改为 optimal_rmse_test。

# Log the optimal model
v1 = registry.log_model(
   model_name=model_name,
   version_name="v1",
   model=regressor,
   sample_input_data=sample,
)

# Add the models RMSE score
v1.set_metric(metric_name="RMSE", value=optimal_rmse_test)

# Add a description
v1.comment = "Optimal model found with RandomizedSearchCV"技术支持

为确认模型已保存，您可以调用 .show_models()：

# Confirm the models are added
registry.show_models()技术支持

关于如何管理注册表的更多信息，请参阅 Snowpark 开发者文档此页面。

在 Snowpark 中进行推理

推理通常通过从注册表中选择最佳模型来完成。我们可以通过筛选 .show_models() 的结果来实现，或像下面这样直接使用版本标签检索模型：

# Doing inference with the optimal model
optimal_version = registry.get_model(model_name).version("v1")
results = optimal_version.run(test_df, function_name="predict")

results.columns技术支持

get_model() 返回的是通用的 model 对象，与 XGBRegressor 不同。因此您需要通过指定函数名来调用 .run() 执行推理。

总结

在本教程中，我们完成了在 Snowflake Snowpark 中训练机器学习模型的端到端流程。做得很好！

我们从一个“骨架版”的未清洗数据集出发，最终得到一个已调参的 XGBoost 回归模型。所有操作均在 Snowflake Cloud 上完成，也就意味着我们几乎无需占用本地系统资源。

我们了解到，在大数据集场景下，这类库内操作的效率非常高——而 Snowflake Snowpark 是推动这一过程的市场上最优秀的工具之一。

如果您想进一步了解 Snowpark 或 Snowflake，请查看以下资源：

感谢您的阅读！

作者

Bexruz (Bex) Tuychiev

我是一名数据科学内容创作者，拥有超过 2 年的经验，并在 Medium 上拥有较大的读者群。我喜欢用略带讽刺的笔调撰写有关 AI 和机器学习的深度文章——毕竟总得想点办法让这些内容不那么枯燥。我已发表 130 多篇文章，并制作了一门 DataCamp 课程，另有一门正在筹备中。我的内容已被超过 500 万人次阅读，其中有 2 万人在 Medium 和 LinkedIn 上成为关注者。

主题

大数据人工智能

主题

技术

Snowflake Snowpark：全面入门

使用 AI 探索

什么是 Snowflake Snowpark？

为什么选择 Snowflake Snowpark？

Snowpark 入门

创建虚拟环境

将示例数据导入 Snowflake

创建 Snowpark 会话

将已导入数据加载到会话

理解 Snowpark DataFrame

Snowpark DataFrame 采用惰性计算

何时使用 Snowpark DataFrame

Snowpark DataFrame 的变换函数

在 Snowpark DataFrame 上进行 EDA

在 Snowpark 中训练机器学习模型

在 Snowpark 中清洗数据

在 Snowpark 中为建模做准备

在 Snowpark 中训练 XGBoost 模型

在 Snowpark 中进行超参数调优

在 Snowpark 中保存最佳模型

在 Snowpark 中进行推理

总结

主题

深入了解 Snowflake 与大数据！

Snowflake SQL 入门

数据工程师在 Python 中

Snowflake 中的数据建模入门

主题

技术

深入了解 Snowflake 与大数据！

Snowflake SQL 入门

数据工程师在 Python 中

Snowflake 中的数据建模入门

什么是 Snowflake Snowpark？

为什么选择 Snowflake Snowpark？

Snowpark 入门

创建虚拟环境

将示例数据导入 Snowflake

创建 Snowpark 会话

将已导入数据加载到会话

理解 Snowpark DataFrame

Snowpark DataFrame 采用惰性计算

何时使用 Snowpark DataFrame

Snowpark DataFrame 的变换函数

在 Snowpark DataFrame 上进行 EDA

在 Snowpark 中训练机器学习模型

在 Snowpark 中清洗数据

在 Snowpark 中为建模做准备

在 Snowpark 中训练 XGBoost 模型

在 Snowpark 中进行超参数调优

在 Snowpark 中保存最佳模型

在 Snowpark 中进行推理

总结

主题

深入了解 Snowflake 与大数据！

Snowflake SQL 入门

数据工程师 在 Python 中

Snowflake 中的数据建模入门

深入了解 Snowflake 与大数据！

Snowflake SQL 入门

数据工程师 在 Python 中

Snowflake 中的数据建模入门

数据工程师在 Python 中

数据工程师在 Python 中