Snowflake Snowpark: Giới thiệu toàn diện

Snowflake Snowpark là gì?

Cách làm học máy truyền thống là đưa dữ liệu từ cơ sở dữ liệu về nơi đặt mô hình. Với làn sóng AI gần đây và quy mô đồ sộ của các bộ dữ liệu ngày nay, cách tiếp cận này ngày càng thiếu thực tế.

Hàng terabyte dữ liệu phải được chuyển từ cơ sở dữ liệu sang các ứng dụng phía khách để làm sạch, phân tích và huấn luyện mô hình. Chuyến đi hai chiều tưởng chừng vô hại này lại tiêu tốn tài nguyên quý giá. Vì vậy, ngày càng nhiều công ty lựa chọn công nghệ trong cơ sở dữ liệu (in-database) để giảm thiểu việc di chuyển dữ liệu và vận hành dữ liệu trơn tru.

Một trong những công nghệ in-database tốt nhất trên thị trường là Snowpark do Snowflake Cloud cung cấp. Snowpark là bộ thư viện và môi trường chạy (runtime) cho phép bạn chạy an toàn các ngôn ngữ lập trình trong Snowflake Cloud để phát triển pipeline dữ liệu và mô hình học máy ngay trong cùng môi trường với cơ sở dữ liệu Snowflake của bạn.

Trong hướng dẫn này, chúng tôi sẽ bàn về những điều căn bản của Snowpark và cách bạn có thể sử dụng nó trong dự án của mình. Chúng tôi giả định bạn đã quen với SQL và Snowflake—nếu cần học trước, bạn có thể tham gia SQL Fundamental Skill Track của chúng tôi hoặc đọc Hướng dẫn Snowflake cho Người mới bắt đầu.

Bắt đầu thôi!

Vì sao chọn Snowflake Snowpark?

Snowflake Snowpark là bộ thư viện và runtime cho phép bạn sử dụng an toàn các ngôn ngữ như Python, Java và Scala để xử lý dữ liệu trực tiếp trong nền tảng đám mây của Snowflake.

Điều này loại bỏ nhu cầu đưa dữ liệu ra khỏi Snowflake để xử lý, cải thiện hiệu quả và bảo mật. Dưới đây là một số lợi ích chính:

Xử lý dữ liệu trong Snowflake: Viết mã bằng ngôn ngữ ưa thích để thao tác và phân tích cơ sở dữ liệu SQL của bạn trong Snowflake và thực thi ngay trong môi trường bảo mật của Snowflake. Điều này loại bỏ nhu cầu chuyển dữ liệu sang môi trường khác.
Cải thiện hiệu năng: Bằng cách xử lý dữ liệu trực tiếp trong Snowflake, Snowpark có thể tận dụng kiến trúc co giãn và không máy chủ của nền tảng để xử lý hiệu quả.
Giảm chi phí và gánh nặng công nghệ: Vì Snowflake chủ yếu cung cấp tài nguyên, bạn không cần quản lý nền tảng riêng cho tính toán và lưu trữ.
Làm việc với những gì bạn biết và ở nơi bạn muốn: API của Snowpark cho phép bạn kết nối tới cơ sở dữ liệu SQL từ bất kỳ môi trường nào, như Jupyter hoặc VSCode, và xây dựng pipeline dữ liệu cùng ứng dụng ML. Điểm hay nhất là bạn có thể dùng các thư viện yêu thích như Pandas, Scikit-learn, XGBoost, v.v. cùng với các framework của Snowpark.

Tóm lại, Snowpark là cách mạnh mẽ nhưng đơn giản để nhà phát triển xây dựng pipeline dữ liệu, giải pháp học máy và ứng dụng dựa trên dữ liệu trực tiếp trong Snowflake Cloud.

Bắt đầu với Snowpark

Mục tiêu cuối cùng trong hướng dẫn này là xây dựng một mô hình đã tinh chỉnh siêu tham số, được huấn luyện trên một bảng trong cơ sở dữ liệu Snowflake bằng Snowpark. Để làm vậy, chúng ta sẽ bắt đầu bằng cách:

Tạo môi trường ảo với các thư viện liên quan.
Nạp dữ liệu mẫu vào Snowflake.
Tạo phiên Snowpark để kết nối tới Snowflake.
Nạp dữ liệu đã ingest vào phiên.

Tạo môi trường ảo

Trong hướng dẫn này, chúng ta sẽ dùng một môi trường conda mới:

$ conda create -n snowpark python==3.10 -y
$ conda activate snowparkĐược hỗ trợ bởi

Lưu ý thêm: nếu bạn đang chạy trên một môi trường conda vừa cài mới, bạn cần chạy $conda init trước khi kích hoạt môi trường snowpark.

Sau khi kích hoạt, chúng ta cần cài các thư viện sau:

$ pip install snowflake-snowpark-python #The Snowpark API
$ pip install pandas pyarrow numpy matplotlib seabornĐược hỗ trợ bởi

Nếu bạn sẽ dùng Jupyter, hãy cài thêm ipykernel và chạy lệnh sau để môi trường đang dùng được thêm làm một kernel của Jupyter:

$ pip install ipykernel
$ ipython kernel install --user --name=snowparkĐược hỗ trợ bởi

Giờ hãy import một số thư viện chung chúng ta sẽ cần trong quá trình thực hiện:

import warnings
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
warnings.filterwarnings("ignore")Được hỗ trợ bởi

Nạp dữ liệu mẫu vào Snowflake

Trong hướng dẫn này, chúng ta sẽ dùng bộ dữ liệu Diamonds của Seaborn—bạn có thể tải trực tiếp tệp CSV từ GitHub của tôi.

Sau khi đăng nhập vào ứng dụng Snowflake, hãy làm theo GIF dưới đây để nạp bộ dữ liệu vào một bảng cơ sở dữ liệu mới (tạo tài khoản mới nếu bạn chưa có):

Trong thực tế, bạn hiếm khi import tệp CSV thành bảng vào cơ sở dữ liệu Snowflake. Phần lớn thời gian, bạn sẽ làm việc với các cơ sở dữ liệu sẵn có với những mức quyền truy cập khác nhau do quản trị viên cơ sở dữ liệu của công ty cấp.

Tạo một phiên Snowpark

Giờ chúng ta phải thiết lập kết nối giữa API của Snowpark và đám mây Snowflake để truy vấn cơ sở dữ liệu của mình. Kết nối này cần các thông tin xác thực của Snowflake: tên tài khoản, tên người dùng và mật khẩu.

Trong hình dưới, tôi sẽ chỉ cách bạn có thể lấy tên người dùng và tên tài khoản từ bảng điều khiển Snowflake:

Tạo một tệp riêng tên config.py chứa một dictionary duy nhất tên credentials với định dạng sau:

credentials = (
   {
       "account": "3-4",  # Combine 3 and 4 with a hyphen
       "username": "bexgboost",  # Your username in lowercase
       "password": "your_password",  # Your Snowflake password
   },
)Được hỗ trợ bởi

Vì lý do bảo mật, chúng ta lưu thông tin xác thực trong tệp riêng. Thêm tệp này vào .gitignore để đảm bảo thông tin xác thực Snowflake của bạn không bị lộ ngoài ý muốn.

Giờ hãy import dictionary credentials và lớp Session của Snowpark:

from config import credentials
from snowflake.snowpark import SessionĐược hỗ trợ bởi

Để thiết lập kết nối, chúng ta sẽ dùng phương thức 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()Được hỗ trợ bởi

'"bexgboost"'Được hỗ trợ bởi

Nếu tên người dùng của bạn được in ra, phiên đầu tiên đã được tạo thành công!

Nạp dữ liệu đã ingest vào phiên

Đối tượng new_session có quyền truy cập mọi thứ mà tài khoản của bạn sở hữu trong Snowflake (và những gì bạn được cấp quyền truy cập).

Bước đầu tiên trước khi import dữ liệu là cho phiên biết chúng ta đang dùng cơ sở dữ liệu test_db đã tạo ở phần trước:

new_session.sql("USE DATABASE test_db;").collect()Được hỗ trợ bởi

[Row(status='Statement executed successfully.')]Được hỗ trợ bởi

Chúng ta làm việc này bằng phương thức .sql, cho phép chạy bất kỳ truy vấn tương thích SQL của Snowflake.

Giờ chúng ta có thể nạp bảng diamonds của test_db bằng phương thức table của đối tượng session:

diamonds_df = new_session.table("diamonds")
diamonds_df.show(5)Được hỗ trợ bởi

Phương thức .show() tương đương với DataFrame.describe() của Pandas, và cho thấy chúng ta đã kết nối thành công tới bảng.

Đúng vậy — chúng ta chỉ kết nối tới bảng diamonds. Đối tượng diamonds_df không giữ dữ liệu nào, thể hiện qua kích thước của nó:

import sys
sys.getsizeof(diamonds_df)Được hỗ trợ bởi

48Được hỗ trợ bởi

Nó chỉ 48 byte trong khi lẽ ra phải hơn 3 MB. Hãy dành chút thời gian để hiểu vì sao lại như vậy.

Tìm hiểu Snowpark DataFrame

Đến đây, chúng ta đã thực hiện các bước sau:

Tạo môi trường ảo với các thư viện liên quan.
Nạp dữ liệu mẫu vào Snowflake.
Tạo phiên Snowpark để kết nối tới Snowflake.

Giờ hãy bàn về Snowpark DataFrame!

Snowpark DataFrame hoạt động kiểu lười thực thi (lazy)

Pandas DataFrame sử dụng RAM của bạn, nghĩa là chúng nằm trên máy của bạn. So với đó, Snowpark DataFrame tồn tại trong nền tảng đám mây của Snowflake, dù bạn có thể xem biểu diễn của chúng trong môi trường mã của mình. Dữ liệu nằm trên đám mây và không cần tải về máy cục bộ để thao tác.

Một khía cạnh quan trọng khác của Snowpark DataFrame là chúng hoạt động theo cơ chế lười thực thi. Nghĩa là chúng không thực hiện bất kỳ phép toán nào trên dữ liệu cho tới khi bạn yêu cầu rõ ràng.

Thay vào đó, chúng xây dựng biểu diễn logic của các thao tác mong muốn, rồi dịch sang các truy vấn SQL tối ưu để Snowflake thực thi.

Ngược lại, Pandas thực thi ngay lập tức—nói cách khác, nó thực hiện eager execution.

Về hiệu năng, đánh giá lười (lazy evaluation) luôn nhanh hơn đáng kể so với thực thi tức thì. Vì DataFrame lười sử dụng tài nguyên tính toán co giãn mạnh mẽ của Snowflake, chúng có thể đẩy phép toán xuống tầng cơ sở dữ liệu và chạy nhanh hơn nhiều.

Để hiểu rõ hơn sự khác biệt giữa lười thực thi và thực thi tức thì, hãy hình dung dữ liệu của chúng ta trong Snowflake là một thư viện khổng lồ. Có hai kịch bản sau:

Thực thi tức thì (Pandas DataFrame): Chúng ta gom tất cả sách cần thiết về bàn làm việc (máy cục bộ) rồi phân tích từng cuốn.
Đánh giá lười (Snowpark DataFrame): Chúng ta nói với thủ thư (Snowpark) những cuốn sách cần và phân tích muốn thực hiện, rồi thủ thư sẽ truy xuất và phân tích thông tin liên quan một cách hiệu quả.

Khi nào dùng Snowpark DataFrame

Vì Snowpark có thể chuyển đổi sang Pandas DataFrame, vậy khi nào nên dùng cái nào?

pandas_diamonds = diamonds_df.to_pandas()
pandas_diamonds.head()Được hỗ trợ bởi

Câu trả lời chủ yếu phụ thuộc vào kích thước bộ dữ liệu.

Nếu bộ dữ liệu nhỏ (như Diamonds), bạn có thể tải về máy và đưa vào Pandas mà không gặp vấn đề (điều mà phương thức DataFrame.to_pandas() thực hiện).

Nhưng với dữ liệu ngày nay, mọi thứ có thể vượt tầm kiểm soát rất nhanh và ta có thể phải chờ hàng giờ để tải cơ sở dữ liệu. Và nhớ rằng, đó là vé khứ hồi—bất kỳ dữ liệu mới nào muốn lưu lại cũng phải gửi ngược lên. Đó là lý do học API Snowpark DataFrame rất có giá trị.

Ngoài ra, phương thức Session.sql() cho phép chúng ta thực thi bất kỳ biểu thức SQL nào tương thích với Snowflake (xem ví dụ dưới). Tuy nhiên, tôi khuyến nghị bạn làm chủ API Snowpark DataFrame vì nó có rất nhiều hàm dựng sẵn cho hơn 100 hàm và biểu thức SQL.

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

type(result)
Được hỗ trợ bởi

snowflake.snowpark.dataframe.DataFrameĐược hỗ trợ bởi

result.show()Được hỗ trợ bởi

Các hàm biến đổi Snowpark DataFrame

Phần lớn thời gian của bạn với Snowpark sẽ dành cho các hàm biến đổi (transformation). Chúng có sẵn dưới mô-đun con sau:

# Thông lệ là import với bí danh F
import snowflake.snowpark.functions as FĐược hỗ trợ bởi

Tất cả hàm dưới F (bạn có thể liệt kê bằng dir(F)) xác định cách một DataFrame nên được biến đổi. Hãy xem ví dụ:

F.upper(F.col("CUT"))Được hỗ trợ bởi

Ở đây, hàm col tạo tham chiếu tới cột đã cho và trả về một đối tượng Column. Sau đó, .upper() chuyển các giá trị của cột CUT thành chữ hoa:

diamonds_df.show(5)Được hỗ trợ bởi

Nhưng nếu xem bộ dữ liệu diamonds, ta thấy các giá trị trong cột CUT vẫn có chữ thường. Vì sao? Biểu thức F.upper(F.col("CUT")) tương đương với truy vấn SQL sau:

SELECT UPPER(CUT);Được hỗ trợ bởi

Thiếu gì ở đây?

Dĩ nhiên là mệnh đề FROM chỉ định tên bảng mà phép biến đổi sẽ áp dụng!

Vì vậy, chúng ta phải kết hợp biểu thức F.upper(F.col("CUT")) với đối tượng diamonds_df. Có ba cách chính:

Với biểu thức biến đổi hoặc chọn cột, truyền vào .select()
Với biểu thức chọn hàng, truyền vào .filter()
Với biểu thức tạo cột, truyền vào .with_column()

Vì biểu thức của chúng ta biến đổi một cột, ta sẽ truyền vào .select():

our_expression = F.upper(F.col("CUT"))
diamonds_df.select(our_expression).show()Được hỗ trợ bởi

Chúng ta đã thay đổi giá trị thành công!

Giờ hãy viết một biểu thức lọc hàng:

diamonds_df.filter(F.col("PRICE") > 10000).count()Được hỗ trợ bởi

5222Được hỗ trợ bởi

Ta thấy có hơn 5.000 viên kim cương trong bộ dữ liệu có giá trên 10.000$. Hãy thử tạo thêm một cột:

(
diamonds_df.with_column("PRICE_SQUARED", F.col("PRICE") ** 2)
.select("PRICE", "PRICE_SQUARED")
.show(5)
)Được hỗ trợ bởi

Ở trên, chúng ta cũng minh họa việc có thể xâu chuỗi phương thức—chúng ta có cột mới tên “PRICE_SQUARED.” Có thể thấy các hàm của Snowpark DataFrame tương đồng với Pandas.

Tóm lại, nếu bạn đang tìm hàm Snowpark tương đương một hàm Pandas, nó hoặc nằm dưới F hoặc là một phương thức của Snowpark DataFrame. Bạn có thể kiểm tra bằng dir(F) hoặc dir(diamonds_df) hoặc đọc tài liệu tham chiếu Snowpark Python.

Thực hiện EDA trên Snowpark DataFrame

Một phần đáng kể của phân tích khám phá dữ liệu (EDA) là tạo trực quan hóa. May mắn là chúng ta không cần Snowpark cho việc này vì có thể tiếp tục dùng bộ công cụ dữ liệu hiện đại.

Ý tưởng là: Vì EDA chủ yếu nhằm khám phá xu hướng và insight tổng quát, ta thường có thể dùng mẫu thay vì toàn bộ dữ liệu. Sau khi tải xuống một mẫu đại diện, ta có thể chuyển nó sang Pandas DataFrame và dùng Matplotlib hoặc Seaborn quen thuộc.

Bắt đầu với việc chuyển đổi:

sample = diamonds_df.sample(0.25).to_pandas()
sample.head()Được hỗ trợ bởi

Ở trên, chúng ta đang tải 25% bộ dữ liệu dưới dạng Pandas DataFrame. Hãy chuyển tên cột về chữ thường:

sample.columns = [col.lower() for col in sample.columns]
sample.columnsĐược hỗ trợ bởi

Index(['carat', 'cut', 'color', 'clarity', 'depth', 'table', 'price', 'x', 'y', 'z'], dtype='object')Được hỗ trợ bởi

Giờ bạn có thể làm EDA như với bất kỳ bộ dữ liệu nào. Dưới đây, chúng ta khám phá vài ý tưởng và bắt đầu bằng scatter plot để trực quan mối quan hệ giữa giá và carat:

#Remember we imported seaborn as sns
sns.scatterplot(data=sample, x="price", y="carat", s=1)
plt.title("Diamond prices vs. carat")
plt.show()Được hỗ trợ bởi

Giờ hãy vẽ ma trận tương quan để xem tương quan giữa từng cặp đặc trưng số:

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()Được hỗ trợ bởi

Chúng ta cũng có thể tạo biểu đồ cột cho các hạng mục độ cắt của kim cương:

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()Được hỗ trợ bởi

Để xác nhận rằng các biểu đồ này sẽ xấp xỉ giống khi áp dụng lên toàn bộ dữ liệu, chúng ta sẽ chạy phiên bản SQL của biểu đồ cột ở trên.

Trước tiên, chúng ta sẽ viết một biểu thức SQL để nhóm bộ dữ liệu diamonds theo CUT:

result = new_session.sql(
   """
   SELECT cut, COUNT(*) AS count
     FROM diamonds
    GROUP BY cut;
"""
)
Được hỗ trợ bởi

Giờ hãy chuyển và tải kết quả về dưới dạng Pandas DataFrame:

result_pd = result.to_pandas()
result_pdĐược hỗ trợ bởi

Lần này, chúng ta sẽ dùng hàm plt.bar() để tạo biểu đồ:

plt.bar(x=result_pd["CUT"], height=result_pd["COUNT"])
plt.show()Được hỗ trợ bởi

Thứ tự các cột khác nhau nhưng khi đặt cạnh nhau có thể thấy chúng truyền tải cùng thông tin.

Huấn luyện mô hình Học máy trong Snowpark

Giờ chúng ta đã sẵn sàng học cách huấn luyện mô hình học máy trong Snowpark!

Chúng ta sẽ bắt đầu bằng việc làm sạch bộ dữ liệu.

Làm sạch dữ liệu trong Snowpark

Trước hết, hãy đảm bảo mọi cột đều có kiểu dữ liệu đúng:

list(diamonds_df.schema)Được hỗ trợ bởi

[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)]Được hỗ trợ bởi

Cột table có tên hơi kỳ, nên chúng ta sẽ chuyển nó thành chữ hoa TABLE_ (thêm dấu gạch dưới ở cuối để không trùng với từ khóa SQL dựng sẵn):

diamonds_df = diamonds_df.with_column_renamed('"table"', "TABLE_")
diamonds_df.columnsĐược hỗ trợ bởi

['CARAT', 'CUT', 'COLOR', 'CLARITY', 'DEPTH', 'TABLE_', 'PRICE', 'X', 'Y', 'Z']Được hỗ trợ bởi

Chúng ta sẽ cần đổi kiểu dữ liệu của các đặc trưng số từ DecimalType sang DoubleType vì số thập phân chưa được hỗ trợ trong Snowpark.

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))Được hỗ trợ bởi

[Column("CARAT"),
Column("X"),
Column("Y"),
Column("Z"),
Column("DEPTH"),
Column("TABLE_")]Được hỗ trợ bởi

Trong đoạn mã trên, chúng ta lại dùng phương thức .with_column(). Một phương thức mới khác là .cast(), vốn là một phần của các phương thức Snowpark DataFrame.

Snowpark cũng yêu cầu tất cả đặc trưng văn bản ở dạng chữ hoa và không có khoảng trắng giữa các từ trước khi mã hóa. Hiện tại, các giá trị của đặc trưng CUT vi phạm yêu cầu này, nên chúng ta sẽ sửa bằng các phép biến đổi hàm:

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)
Được hỗ trợ bởi

Dữ liệu của chúng ta giờ không còn vấn đề làm sạch. Ta có thể lưu nó lại thành một bảng mới trong Snowflake:

diamonds_df.write.mode("overwrite").save_as_table("diamonds_cleaned")Được hỗ trợ bởi

Nhớ thêm chế độ overwrite vì bạn có thể cần quay lại, thay đổi vài thao tác làm sạch và lưu lại.

Chuẩn bị dữ liệu cho mô hình trong Snowpark

Trong phần này, chúng ta sẽ xử lý các vấn đề tiền xử lý còn lại có thể cản trở việc huấn luyện mô hình. Hãy nạp bảng đã làm sạch từ phần trước:

clean_df = new_session.table("diamonds_cleaned")
clean_df.show()Được hỗ trợ bởi

Rất tốt!

Tiếp theo, chúng ta sẽ import các mô-đun con preprocessing và pipeline từ namespace snowflake.ml.

import snowflake.ml.modeling.preprocessing as snowml
from snowflake.ml.modeling.pipeline import PipelineĐược hỗ trợ bởi

Trong khi API DataFrame của Snowpark nằm dưới snowflake.snowpark, các mô-đun Snowpark ML lại có trong snowflake.ml. Điều này lúc đầu có thể gây bối rối, nên hãy dành thời gian làm quen.

Trong đoạn mã dưới đây, chúng ta:

Liệt kê tên các đặc trưng phân loại cần mã hóa.
Định nghĩa tên cho các đặc trưng đầu ra vì Snowpark thêm các đặc trưng đã mã hóa thành cột mới thay vì thay thế cột cũ.
Xác định thứ tự đúng của các hạng mục thứ bậc; trong bộ dữ liệu diamonds, tất cả hạng mục đều có trật tự nội tại, giá trị càng về bên phải càng đắt.

# 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"]),
}Được hỗ trợ bởi

Giờ chúng ta sẵn sàng xây dựng một pipeline tương tự pipeline của 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
           ),
       ),
   ]
)
Được hỗ trợ bởi

Lớp Pipeline nhận một danh sách bước, mỗi bước là một lớp tiền xử lý. Ở đây, chúng ta dùng OrdinalEncoder để mã hóa biến phân loại thành 0, 1, 3, v.v. và StandardScaler để chuẩn hóa các đặc trưng số.

Cuối cùng, chúng ta sẽ lưu pipeline cục bộ bằng joblib và thử trên toàn bộ bộ dữ liệu:

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Được hỗ trợ bởi

Đoạn mã chạy không lỗi, nghĩa là pipeline hoạt động, báo hiệu chúng ta có thể huấn luyện mô hình.

Huấn luyện mô hình XGBoost trong Snowpark

Để huấn luyện mô hình, chúng ta sẽ liệt kê lại tên các đặc trưng và mục tiêu:

# 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Được hỗ trợ bởi

Giờ chúng ta sẽ chia dữ liệu bằng DataFrame.random_split() và cho trọng số để biểu diễn tỷ lệ tập huấn luyện và kiểm tra:

# Split the data
train_df, test_df = diamonds_df.random_split(weights=[0.8, 0.2], seed=42)Được hỗ trợ bởi

Sau đó, chúng ta sẽ tải pipeline và fit-transform cho cả tập huấn luyện và kiểm tra.

# 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)Được hỗ trợ bởi

Ở trên, hãy đảm bảo bạn chỉ gọi .transform() trên tập kiểm tra để tránh rò rỉ dữ liệu.

Tiếp theo, chúng ta khởi tạo mô hình hồi quy XGBoost từ ml.modeling.xgboost sub-module:

# 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,
)Được hỗ trợ bởi

Lớp XGBRegressor yêu cầu tất cả tên đầu vào, tất cả tên mục tiêu và tên cột đầu ra. Khi đã nhập vào, chúng ta gọi .fit() để bắt đầu huấn luyện:

# Train
regressor.fit(train_df_transformed)Được hỗ trợ bởi

Việc huấn luyện có thể chạy chậm nếu bạn dùng tài khoản Snowflake miễn phí với tài nguyên tính toán hạn chế và không có GPU.

Khi huấn luyện xong, chúng ta có thể sinh dự đoán:

# Dự đoán

train_preds = regressor.predict(train_df_transformed)
test_preds = regressor.predict(test_df_transformed)Được hỗ trợ bởi

Hãy xem các dự đoán:

train_preds.select("PRICE", "PRICE_PREDICTED").show(5)Được hỗ trợ bởi

Giờ chúng ta phải đo lường mức độ tốt của mô hình ban đầu. Snowpark bao gồm hàng chục thước đo từ Scikit-learn dưới mô-đun con ml.modeling.metrics mà chúng ta import là M:

import snowflake.ml.modeling.metrics as MĐược hỗ trợ bởi

Tiếp theo, chúng ta chạy mean_squared_error hai lần để đo sai số bình phương trung bình căn (RMSE) trên cả tập huấn luyện và kiểm tra (bên dưới, squared=False trả về RMSE thay vì 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}")Được hỗ trợ bởi

Train RMSE score for XGBRegressor: 371.9664
Test RMSE score for XGBRegressor: 542.1566Được hỗ trợ bởi

Mô hình của chúng ta lệch khoảng 542$ trung bình và có thể cũng đang overfit đôi chút vì chênh lệch giữa RMSE huấn luyện và kiểm tra khá lớn.

Hãy tinh chỉnh siêu tham số của mô hình để giải quyết các vấn đề này.

Tinh chỉnh siêu tham số trong Snowpark

Hiện tại, Snowpark cung cấp hai lớp tinh chỉnh siêu tham số:

GridSearchCV: tìm kiếm toàn diện mọi tổ hợp siêu tham số kèm cross-validation.
RandomizedSearchCV: tìm kiếm ngẫu nhiên theo các phân phối siêu tham số cho trước kèm cross-validation.

XGBoost có khoảng cả chục siêu tham số có thể cải thiện hiệu năng (bạn có thể tìm hiểu thêm trong hướng dẫn sử dụng XGBoost trong Python). Điều này hướng chúng ta tới tìm kiếm ngẫu nhiên. Cá nhân tôi thích dùng Optuna với tìm kiếm Bayes, nhưng hiện chưa có lựa chọn đó trong Snowpark.

Hãy định nghĩa tìm kiếm giờ đây:

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)
Được hỗ trợ bởi

Chúng ta cố định số lượng estimators là 2000 và chỉ tinh chỉnh các tham số max_depth và learning_rate. rscv có tham số mặc định n_iter=10, nghĩa là quá trình tìm kiếm sẽ chạy 10 lần, mỗi lần chọn ngẫu nhiên một kết hợp tham số. Để có kết quả chính xác hơn, tốt nhất nên chọn giá trị lớn hơn cho tham số này.

Giờ hãy đo hiệu năng của mô hình tốt nhất tìm được. Mã giống như trước nhưng thay vì regressor, chúng ta sẽ dùng đối tượng rscv:

# 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}")Được hỗ trợ bởi

Train RMSE score for optimal model: 224.2405
Test RMSE score for optimal model: 572.0517Được hỗ trợ bởi

RMSE huấn luyện giảm xuống nhưng RMSE kiểm tra còn lớn hơn. Điều này cho thấy chúng ta vẫn đang overfit và cần mở rộng phạm vi tìm kiếm, bao gồm thêm các tham số để tỉa bớt cây quyết định mà XGBoost dùng bên dưới.

Phần đó tôi xin dành cho bạn.

Lưu mô hình tốt nhất trong Snowpark

Nếu đóng phiên ngay bây giờ, chúng ta sẽ mất mô hình đã tinh chỉnh. Để lưu, chúng ta sẽ dùng sổ đăng ký mô hình (model registry) gốc của Snowpark, là một kho lưu trữ ảo bạn có thể dùng để lưu bất kỳ mô hình và metadata của nó.

Registry có sẵn dưới lớp Registry, và nó yêu cầu tên cơ sở dữ liệu và schema hiện tại của phiên. Nó cũng yêu cầu chúng ta đặt tên dự án để tránh xung đột với mô hình từ các dự án khác:

# 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)Được hỗ trợ bởi

Chúng ta sẽ dùng phương thức Registry.log_model() để lưu các mô hình:

# 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,
)Được hỗ trợ bởi

Chúng ta có thể ghi lại một thước đo cho mô hình bằng phương thức .set_metric() (có thể ghi bao nhiêu thước đo cũng được). Chúng ta cũng sẽ thêm một ghi chú mô tả mô hình:

# 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"Được hỗ trợ bởi

Chúng ta làm tương tự cho mô hình tốt nhất tìm được với random search. Khi ghi thước đo, đừng quên đổi giá trị thành 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"Được hỗ trợ bởi

Để xác nhận rằng các mô hình đã được lưu, bạn có thể gọi .show_models():

# Confirm the models are added
registry.show_models()Được hỗ trợ bởi

Tìm hiểu thêm cách quản lý registry từ trang này của tài liệu nhà phát triển Snowpark.

Suy luận (inference) trong Snowpark

Suy luận thường được thực hiện bằng cách chọn mô hình tốt nhất từ registry của bạn. Chúng ta có thể làm điều đó bằng cách lọc kết quả của .show_models() hoặc truy xuất trực tiếp mô hình bằng thẻ phiên bản như bên dưới:

# 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Được hỗ trợ bởi

Phương thức get_model() trả về một đối tượng model chung, khác với XGBRegressor. Đó là lý do bạn phải gọi .run() và chỉ định tên hàm để thực hiện suy luận.

Kết luận

Trong hướng dẫn này, chúng ta đã đi qua một quy trình end-to-end để huấn luyện mô hình học máy trong Snowflake Snowpark. Làm tốt lắm!

Chúng ta bắt đầu với một bộ dữ liệu thô chưa làm sạch và kết thúc với một mô hình hồi quy XGBoost đã tinh chỉnh. Mọi thao tác đều thực hiện trên Snowflake Cloud, nghĩa là chúng ta không phải dùng tài nguyên hệ thống.

Chúng ta đã học được hiệu quả của kiểu vận hành trong cơ sở dữ liệu khi làm việc với bộ dữ liệu lớn—và rằng Snowflake Snowpark là một trong những công cụ tốt nhất trên thị trường để hỗ trợ quá trình này.

Nếu bạn muốn tìm hiểu thêm về Snowpark hoặc Snowflake, hãy xem các tài nguyên sau:

Cảm ơn bạn đã đọc!

Tác giả

Bexruz (Bex) Tuychiev

Tôi là người sáng tạo nội dung về khoa học dữ liệu với hơn 2 năm kinh nghiệm và là một trong những tài khoản có lượng người theo dõi lớn nhất trên Medium. Tôi thích viết các bài chuyên sâu về AI và ML với chút giọng điệu mỉa mai, vì bạn cũng phải làm gì đó để chúng bớt khô khan. Tôi đã xuất bản hơn 130 bài viết và một khóa học trên DataCamp, cùng với một khóa nữa đang được thực hiện. Nội dung của tôi đã được hơn 5 triệu lượt xem, trong đó 20 nghìn người đã trở thành người theo dõi trên cả Medium và LinkedIn.

Chủ đề

Big Data Trí tuệ Nhân tạo

Chủ đề

Công nghệ

Snowflake Snowpark: Giới thiệu toàn diện

Khám phá cùng AI

Snowflake Snowpark là gì?

Vì sao chọn Snowflake Snowpark?

Bắt đầu với Snowpark

Tạo môi trường ảo

Nạp dữ liệu mẫu vào Snowflake

Tạo một phiên Snowpark

Nạp dữ liệu đã ingest vào phiên

Tìm hiểu Snowpark DataFrame

Snowpark DataFrame hoạt động kiểu lười thực thi (lazy)

Khi nào dùng Snowpark DataFrame

Các hàm biến đổi Snowpark DataFrame

Thực hiện EDA trên Snowpark DataFrame

Huấn luyện mô hình Học máy trong Snowpark

Làm sạch dữ liệu trong Snowpark

Chuẩn bị dữ liệu cho mô hình trong Snowpark

Huấn luyện mô hình XGBoost trong Snowpark

Tinh chỉnh siêu tham số trong Snowpark

Lưu mô hình tốt nhất trong Snowpark

Suy luận (inference) trong Snowpark

Kết luận

Chủ đề

Tìm hiểu thêm về Snowflake và dữ liệu lớn!

Nhập môn Snowflake SQL

Kỹ sư dữ liệu trong Python

Nhập môn Mô hình dữ liệu trong Snowflake

Liên quan

Claude Opus 4.6: Tính năng, Điểm chuẩn, Bài kiểm tra thực hành và hơn thế nữa

Chủ đề

Công nghệ

Tìm hiểu thêm về Snowflake và dữ liệu lớn!

Nhập môn Snowflake SQL

Kỹ sư dữ liệu trong Python

Nhập môn Mô hình dữ liệu trong Snowflake