Snowflake Snowpark: एक व्यापक परिचय

Snowflake Snowpark क्या है?

पारंपरिक मशीन लर्निंग में आम तौर पर डेटा को डेटाबेस से निकालकर वहाँ ले जाया जाता है जहाँ मॉडल होते हैं। AI में हालिया उछाल और आज के डेटासेट के विशाल आकार के कारण, यह तरीका तेजी से अव्यावहारिक होता जा रहा है।

टेराबाइट्स डेटा को सफाई, विश्लेषण और मॉडल प्रशिक्षण के लिए डेटाबेस से क्लाइंट-साइड एप्लिकेशन तक स्थानांतरित करना पड़ता है। यह देखने में सरल दो-तरफ़ा यात्रा मूल्यवान संसाधनों की बर्बादी है। इसी कारण, अधिक से अधिक कंपनियाँ इन-डेटाबेस तकनीकों को अपना रही हैं ताकि डेटा मूवमेंट कम से कम हो और डेटा ऑपरेशंस सुचारू रूप से चलें।

बाज़ार में उपलब्ध बेहतरीन इन-डेटाबेस तकनीकों में से एक है Snowflake Cloud द्वारा प्रस्तुत Snowpark। Snowpark लाइब्रेरियों और रनटाइम्स का एक सेट है जो आपको Snowflake Cloud के भीतर ही सुरक्षित रूप से प्रोग्रामिंग भाषाएँ चलाने देता है ताकि आप अपने Snowflake डेटाबेस वाले उसी वातावरण में डेटा पाइपलाइनों और मशीन लर्निंग मॉडल तैयार कर सकें।

इस ट्यूटोरियल में, हम Snowpark की बुनियादी बातें और इसे अपने प्रोजेक्ट्स में उपयोग करने के तरीके पर चर्चा करेंगे। हम मानकर चल रहे हैं कि आप SQL और Snowflake से पहले से परिचित हैं—यदि आपको पहले इन्हें कवर करना है, तो आप हमारा SQL Fundamental Skill Track ले सकते हैं या यह शुरुआती लोगों के लिए Snowflake ट्यूटोरियल पढ़ सकते हैं।

चलिए शुरू करते हैं!

क्यों Snowflake Snowpark?

Snowflake Snowpark लाइब्रेरियों और रनटाइम्स का एक सेट है जो आपको Python, Java और Scala जैसी प्रोग्रामिंग भाषाओं का सुरक्षित रूप से उपयोग करके Snowflake के क्लाउड प्लेटफ़ॉर्म में सीधे डेटा प्रोसेस करने देता है।

यह प्रोसेसिंग के लिए डेटा को Snowflake के बाहर ले जाने की ज़रूरत खत्म कर देता है, जिससे दक्षता और सुरक्षा दोनों बेहतर होती हैं। इसके कुछ प्रमुख लाभ नीचे दिए गए हैं:

Snowflake में ही डेटा प्रोसेस करें: अपनी पसंदीदा भाषा में कोड लिखें ताकि आप Snowflake में अपने SQL डेटाबेस को परिवर्तित और विश्लेषित कर सकें और उसे Snowflake के सुरक्षित वातावरण में ही निष्पादित करें। इससे डेटा को किसी अन्य वातावरण में ले जाने की ज़रूरत नहीं रहती।
बेहतर प्रदर्शन: Snowflake में सीधे डेटा प्रोसेस करके, Snowpark कुशल प्रोसेसिंग के लिए प्लेटफ़ॉर्म की इलास्टिक और सर्वरलेस आर्किटेक्चर का लाभ उठा सकता है।
कम लागत और तकनीकी ओवरहेड: चूँकि Snowflake अधिकांश संसाधन प्रदान करता है, आपको कंप्यूट और स्टोरेज के लिए अलग-अलग प्लेटफ़ॉर्म प्रबंधित नहीं करने पड़ते।
जो जानते हैं, उसी से और जहाँ चाहें काम करें: Snowpark के API आपको किसी भी वातावरण, जैसे Jupyter या VSCode, से अपने SQL डेटाबेस से कनेक्ट होने और डेटा पाइपलाइंस व ML ऐप्स बनाने देते हैं। सबसे अच्छी बात यह है कि आप अपनी पसंदीदा लाइब्रेरियों—जैसे Pandas, Scikit-learn, XGBoost आदि—को Snowpark फ्रेमवर्क के साथ उपयोग कर सकते हैं।

संक्षेप में, Snowpark डेवलपर्स के लिए Snowflake Cloud के भीतर ही डेटा पाइपलाइन, मशीन लर्निंग समाधान और डेटा-आधारित एप्लिकेशन बनाने का एक शक्तिशाली और सरल तरीका है।

Snowpark के साथ शुरुआत

इस ट्यूटोरियल में हमारा अंतिम लक्ष्य Snowpark का उपयोग करके Snowflake डेटाबेस की एक टेबल पर प्रशिक्षित और हाइपरपैरामीटर-ट्यून किया हुआ मॉडल बनाना है। ऐसा करने के लिए, हम शुरुआत करेंगे:

संबंधित लाइब्रेरियों के साथ एक वर्चुअल एनवायरनमेंट बनाकर।
नमूना डेटा को Snowflake में इनजेस्ट करके।
Snowflake से कनेक्ट होने के लिए एक Snowpark सेशन बनाकर।
इनजेस्ट किए गए डेटा को सेशन में लोड करके।

वर्चुअल एनवायरनमेंट बनाना

इस ट्यूटोरियल के लिए, हम नया conda एनवायरनमेंट उपयोग करेंगे:

$ conda create -n snowpark python==3.10 -y
$ conda activate snowparkसंचालित द्वारा

साइड नोट के रूप में, यदि आप नए इंस्टॉल किए गए conda वातावरण पर चला रहे हैं, तो $conda init चलाने की आवश्यकता होगी, और फिर snowpark वातावरण को सक्रिय करें।

एक्टिवेशन के बाद, हमें निम्न लाइब्रेरियाँ इन्स्टॉल करनी होंगी:

$ 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 डेटासेट का उपयोग करेंगे—आप CSV फ़ाइल सीधे मेरे GitHub से डाउनलोड कर सकते हैं।

एक बार जब आप 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 DataFrames को समझना

अभी तक, हमने निम्न चरण कवर किए हैं:

संबंधित लाइब्रेरियों के साथ एक वर्चुअल एनवायरनमेंट बनाया।
नमूना डेटा को Snowflake में इनजेस्ट किया।
Snowflake से कनेक्ट होने के लिए Snowpark सेशन बनाया।

अब Snowpark DataFrames पर बात करते हैं!

Snowpark DataFrames लेज़ी होते हैं

Pandas DataFrames आपके RAM का उपयोग करते हैं, अर्थात वे आपकी मशीन पर रहते हैं। इसके विपरीत, Snowpark DataFrames Snowflake के क्लाउड प्लेटफ़ॉर्म के भीतर रहते हैं, भले ही आप उनके प्रतिनिधित्व को अपने कोडिंग वातावरण में देख सकें। डेटा क्लाउड पर ही रहता है और उस पर ऑपरेशंस करने के लिए उसे आपकी लोकल मशीन पर डाउनलोड करने की आवश्यकता नहीं होती।

Snowpark DataFrames का एक और महत्वपूर्ण पहलू यह है कि वे लेज़ी तरीके से काम करते हैं। यानी वे तब तक डेटा पर कोई ऑपरेशन निष्पादित नहीं करते जब तक आप उन्हें स्पष्ट रूप से ऐसा करने को नहीं कहते।

इसके बजाय, वे वांछित ऑपरेशंस का एक लॉजिकल प्रतिनिधित्व बनाते हैं, जिसे बाद में Snowflake के लिए ऑप्टिमाइज़्ड SQL क्वेरीज़ में अनुवादित किया जाता है।

तुलना के लिए, Pandas ऑपरेशंस को तुरंत निष्पादित करता है—दूसरे शब्दों में, यह ईगर एक्सिक्यूशन करता है।

प्रदर्शन के लिहाज से, लेज़ी इवैल्युएशन हमेशा ईगर एक्सिक्यूशन की तुलना में काफ़ी तेज़ होता है। चूँकि लेज़ी DataFrames Snowflake के शक्तिशाली इलास्टिक कंप्यूटिंग संसाधनों का उपयोग करते हैं, वे ऑपरेशंस को डेटाबेस स्तर तक पुश कर सकते हैं और कहीं अधिक तेज़ी से चल सकते हैं।

लेज़ी इवैल्युएशन और ईगर एक्सिक्यूशन के अंतर को बेहतर समझने के लिए कल्पना करें कि Snowflake में हमारा डेटा एक विशाल पुस्तकालय है। हमारे पास दो परिदृश्य हैं:

ईगर एक्सिक्यूशन (Pandas DataFrames): हम सारी ज़रूरी किताबें अपने डेस्क (लोकल मशीन) पर इकट्ठा कर लेते हैं और फिर उन्हें एक-एक करके विश्लेषित करते हैं।
लेज़ी इवैल्युएशन (Snowpark DataFrames): हम लाइब्रेरियन (Snowpark) को बताते हैं कि हमें कौन-सी किताबें चाहिए और किस तरह का विश्लेषण करवाना है, और फिर लाइब्रेरियन संबंधित जानकारी को कुशलता से ढूँढकर विश्लेषित कर देता है।

Snowpark DataFrames कब उपयोग करें

क्योंकि Snowpark को Pandas DataFrames में बदला जा सकता है, तो हमें एक को दूसरे पर कब प्राथमिकता देनी चाहिए?

pandas_diamonds = diamonds_df.to_pandas()
pandas_diamonds.head()संचालित द्वारा

इसका उत्तर काफी हद तक डेटासेट के आकार पर निर्भर करता है।

यदि आपका डेटासेट छोटा है (जैसे Diamonds डेटासेट), तो आप उसे लोकल में डाउनलोड कर सकते हैं और बिना समस्या Pandas में दे सकते हैं (जो कि DataFrame.to_pandas() करता है)।

लेकिन आज के डेटासेट के साथ चीज़ें जल्दी बेकाबू हो सकती हैं और हमें अपना डेटाबेस डाउनलोड करने में घंटों इंतज़ार करना पड़ सकता है। और याद रखें, यह दो-तरफ़ा टिकट है—कोई भी नया डेटा जिसे हम सुरक्षित रखना चाहते हैं, उसे वापस भी भेजना पड़ता है। यही कारण है कि Snowpark DataFrame API सीखना इतना उपयोगी है।

वैकल्पिक रूप से, Session.sql() मेथड हमें कोई भी SQL अभिव्यक्ति चलाने देता है जो Snowflake के साथ संगत हो (नीचे उदाहरण देखें)। हालाँकि, मैं दृढ़ता से सलाह दूँगा कि आप 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 में आपका अधिकांश समय इसके ट्रांसफ़ॉर्मेशन फ़ंक्शंस का उपयोग करने में जाएगा। ये निम्न सब-मॉड्यूल के अंतर्गत उपलब्ध हैं:

# प्रचलन है कि इसे F नाम से इम्पोर्ट करें
import snowflake.snowpark.functions as Fसंचालित द्वारा

F के अंतर्गत सभी फ़ंक्शंस (जिन्हें आप dir(F) कॉल करके सूचीबद्ध कर सकते हैं) यह निर्दिष्ट करते हैं कि किसी DataFrame को कैसे ट्रांसफ़ॉर्म किया जाना चाहिए। एक उदाहरण देखते हैं:

F.upper(F.col("CUT"))संचालित द्वारा

यहाँ, col फ़ंक्शन दिए गए कॉलम का संदर्भ बनाता है और एक Column ऑब्जेक्ट लौटाता है। फिर, .upper() CUT कॉलम के मानों को अपरकेस में बदल देता है:

diamonds_df.show(5)संचालित द्वारा

लेकिन यदि हम डायमंड्स डेटासेट देखें, तो हम पाते हैं कि 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संचालित द्वारा

हम पाते हैं कि हमारे डेटासेट में $10,000 से अधिक कीमत वाले 5000 से अधिक डायमंड्स हैं। चलिए एक कॉलम बनाकर भी देखते हैं:

(
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 DataFrames के मेथड्स का हिस्सा। आप dir(F) या dir(diamonds_df) कॉल करके या Snowpark Python संदर्भ पढ़कर जाँच सकते हैं।

Snowpark DataFrames पर 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 कर सकते हैं। नीचे, हम कुछ संभावित विचारों की पड़ताल करते हैं और कीमत व कैरेट के बीच संबंध को दर्शाने के लिए एक स्कैटर प्लॉट बनाकर शुरू करते हैं:

#याद रखें कि हमने seaborn को 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 के अनुसार डायमंड्स डेटासेट को समूहित करने के लिए एक 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 में अभी डेसिमल्स सपोर्ट नहीं हैं।

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 DataFrames के मेथड्स का हिस्सा है।

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 एनकोडेड फ़ीचर्स को पुराने कॉलम्स को बदलने के बजाय नए कॉलम्स के रूप में जोड़ता है।
ऑर्डिनल श्रेणियों का सही क्रम परिभाषित करते हैं; डायमंड्स डेटासेट में, सभी श्रेणियों में अंतर्निहित ऑर्डरिंग होती है, जो बाएँ से दाएँ अधिक महँगी होती जाती हैं।

# प्रोसेसिंग के लिए सभी फ़ीचर्स की सूची
cat_cols = ["CUT", "COLOR", "CLARITY"]
cat_cols_encoded = ["CUT_OE", "COLOR_OE", "CLARITY_OE"]

# हमारे पास पहले से numeric_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 पाइपलाइन जैसी एक पाइपलाइन बनाने के लिए तैयार हैं:

# पाइपलाइन बनाएं
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 क्लास स्टेप्स की सूची स्वीकार करता है, जिनमें से हर एक एक प्री-प्रोसेसिंग क्लास है। यहाँ, हम श्रेणीबद्ध मानों को 0, 1, 3, आदि के रूप में एनकोड करने के लिए OrdinalEncoder का उपयोग कर रहे हैं और संख्यात्मक फ़ीचर्स को सामान्यीकरण करने के लिए StandardScaler का।

अंत में, हम पाइपलाइन को लोकल रूप से joblib के साथ सेव करेंगे और उसे पूरे डेटासेट पर टेस्ट करेंगे:

import joblib
PIPELINE_FILE = "pipeline.joblib"

# पहले लोकल रूप से पिकल करें
joblib.dump(preprocessing_pipeline, PIPELINE_FILE)

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

# transformed_diamonds_df.show() - लंबा आउटपुट होने के कारण कमेंट किया गया हैसंचालित द्वारा

स्निपेट बिना त्रुटि के चलता है, इसलिए हमारी पाइपलाइन काम कर रही है, जो संकेत देता है कि अब हम अंततः मॉडल ट्रेन कर सकते हैं।

Snowpark में XGBoost मॉडल ट्रेन करना

मॉडल ट्रेन करने के लिए, हम फिर से फ़ीचर नाम और टार्गेट नाम सूचीबद्ध करेंगे:

# कॉलम्स को फिर से परिभाषित करें
cat_cols_encoded = ["CUT_OE", "COLOR_OE", "CLARITY_OE"]
numeric_features = ["CARAT", "X", "Y", "Z", "DEPTH", "TABLE_"]

label_cols = ["PRICE"]  # सूची होना चाहिए
output_cols = ["PRICE_PREDICTED"]  # snowpark में आवश्यकसंचालित द्वारा

अब हम DataFrame.random_split() का उपयोग करके डेटा को स्प्लिट करेंगे और ट्रेनिंग व टेस्टिंग सेट के अंश को दर्शाने के लिए वेट्स देंगे:

# डेटा स्प्लिट करें
train_df, test_df = diamonds_df.random_split(weights=[0.8, 0.2], seed=42)संचालित द्वारा

फिर, हम पाइपलाइन लोड करेंगे और दोनों ट्रेन और टेस्ट सेट पर फिट-ट्रांसफ़ॉर्म करेंगे।

# प्री-प्रोसेसिंग पाइपलाइन को लोकल रूप से लोड करें
pipeline = joblib.load("pipeline.joblib")

# इसे दोनों डेटा फ़्रेम्स पर लागू करें
train_df_transformed = pipeline.fit(train_df).transform(train_df)
test_df_transformed = pipeline.transform(test_df)संचालित द्वारा

ऊपर, यह सुनिश्चित करें कि टेस्ट सेट पर केवल .transform() कॉल करें ताकि डेटा लीकेज से बचा जा सके।

अगला, हम ml.modeling.xgboost सब-मॉड्यूल से XGBoost रिग्रेसर मॉडल इनिशियलाइज़ करते हैं:

# Snowpark में scikit-learn और lightgbm के मॉडल भी हैं
from snowflake.ml.modeling.xgboost import XGBRegressor

# इनिशियलाइज़ करें
regressor = XGBRegressor(
   input_cols=cat_cols_encoded + numeric_features,
   label_cols=label_cols,
   output_cols=output_cols,
)संचालित द्वारा

XGBRegressor क्लास को सभी इनपुट नाम, सभी टार्गेट नाम और आउटपुट कॉलम नाम चाहिए। इन्हें देने के बाद, हम ट्रेनिंग प्रक्रिया शुरू करने के लिए .fit() कॉल करते हैं:

# ट्रेन करें
regressor.fit(train_df_transformed)संचालित द्वारा

यदि आपके पास सीमित कंप्यूटिंग संसाधनों और बिना GPU वाला फ्री Snowflake खाता है, तो ट्रेनिंग धीमी चल सकती है।

ट्रेनिंग पूरी होने के बाद, हम प्रेडिक्शंस जनरेट कर सकते हैं:

# Predict

train_preds = regressor.predict(train_df_transformed)
test_preds = regressor.predict(test_df_transformed)संचालित द्वारा

आइए प्रेडिक्शंस पर नज़र डालें:

train_preds.select("PRICE", "PRICE_PREDICTED").show(5)संचालित द्वारा

अब हमें मापना है कि हमारा आरंभिक मॉडल कितना अच्छा रहा। Snowpark में Scikit-learn से लिए गए दर्जनों मेट्रिक्स ml.modeling.metrics सब-मॉड्यूल में मौजूद हैं जिन्हें हम M के रूप में इम्पोर्ट करते हैं:

import snowflake.ml.modeling.metrics as Mसंचालित द्वारा

फिर, हम mean_squared_error को दो बार चलाते हैं ताकि ट्रेनिंग और टेस्ट दोनों सेट पर रूट मीन स्क्वेयर्ड एरर (RMSE) मापा जा सके (नीचे, squared=False MSE के बजाय RMSE लौटाता है):

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 का उपयोग करना पसंद करता, जो Bayesian सर्च करता है, लेकिन फिलहाल 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 बार किया जाएगा, हर बार पैरामीटर्स के एक रैंडम संयोजन को चुनते हुए। अधिक सटीक परिणामों के लिए, इस पैरामीटर के लिए बड़ा मान चुनना बेहतर है।

अब, पाए गए श्रेष्ठ मॉडल के प्रदर्शन को मापते हैं। कोड पहले जैसा ही है, लेकिन regressor के बजाय हम 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}")संचालित द्वारा

Train RMSE score for optimal model: 224.2405
Test RMSE score for optimal model: 572.0517संचालित द्वारा

ट्रेनिंग RMSE नीचे आया लेकिन टेस्ट RMSE और भी बड़ा है। यह बताता है कि हम अब भी ओवरफ़िट कर रहे हैं और हमें अपनी खोज का दायरा बढ़ाने और अन्य पैरामीटर्स को शामिल करने की ज़रूरत है ताकि XGBoost के अंतर्निहित निर्णय वृक्षों को प्रून किया जा सके।

यह भाग मैं आपको छोड़ता हूँ।

Snowpark में श्रेष्ठ मॉडल को सेव करना

यदि हम अभी सेशन बंद कर दें, तो हमारा ट्यून किया गया मॉडल खो जाएगा। इसे सेव करने के लिए, हम Snowpark के नेटिव मॉडल रजिस्ट्री का उपयोग करेंगे, जो एक वर्चुअल स्टोरेज है जिसमें आप कोई भी मॉडल और उसका मेटाडेटा सेव कर सकते हैं।

रजिस्ट्री Registry क्लास के रूप में उपलब्ध है, और इसे वर्तमान सेशन के डेटाबेस और स्कीमा नाम चाहिए। यह हमसे प्रोजेक्ट को एक नाम देने की भी माँग करता है ताकि अन्य प्रोजेक्ट्स के मॉडलों के साथ कोई टकराव न हो:

# Registry के लिए सेटअप
from snowflake.ml.registry import Registry

# वर्तमान DB और स्कीमा नाम प्राप्त करें
db_name = new_session.get_current_database()
schema_name = new_session.get_current_schema()

# प्रोजेक्ट के लिए वैश्विक मॉडल नाम परिभाषित करें
model_name = "diamond_prices_regression"

# मॉडल्स लॉग करने के लिए रजिस्ट्री इनिशियलाइज़ करें
registry = Registry(session=new_session, database_name=db_name, schema_name=schema_name)संचालित द्वारा

हम अपने मॉडलों को सेव करने के लिए Registry.log_model() मेथड का उपयोग करेंगे:

# रजिस्ट्री में स्कीमा के लिए सैंपल डेटा पास करने हेतु प्राप्त करें
sample = train_df.select(cat_cols_encoded + numeric_features).limit(50)

# पहला मॉडल लॉग करें
v0 = registry.log_model(
   model_name=model_name,
   version_name="v0",
   model=regressor,
   sample_input_data=sample,
)संचालित द्वारा

हम .set_metric() मेथड से मॉडल के लिए एक मेट्रिक लॉग कर सकते हैं (हम चाहें तो कई मेट्रिक्स लॉग कर सकते हैं)। हम मॉडल का वर्णन करने के लिए एक टिप्पणी भी जोड़ेंगे:

# मॉडल का RMSE स्कोर जोड़ें
v0.set_metric(metric_name="RMSE", value=rmse_test)

# विवरण जोड़ें
v0.comment = "The first model to predict diamond prices"संचालित द्वारा

रैंडम सर्च से पाए गए श्रेष्ठ मॉडल के लिए भी यही करते हैं। मेट्रिक लॉग करते समय, मान को optimal_rmse_test में बदलना न भूलें।

# श्रेष्ठ मॉडल लॉग करें
v1 = registry.log_model(
   model_name=model_name,
   version_name="v1",
   model=regressor,
   sample_input_data=sample,
)

# मॉडल का RMSE स्कोर जोड़ें
v1.set_metric(metric_name="RMSE", value=optimal_rmse_test)

# विवरण जोड़ें
v1.comment = "Optimal model found with RandomizedSearchCV"संचालित द्वारा

इसकी पुष्टि करने के लिए कि मॉडल सेव हो गए हैं, आप .show_models() कॉल कर सकते हैं:

# पुष्टि करें कि मॉडल जोड़े गए हैं
registry.show_models()संचालित द्वारा

रजिस्ट्री को कैसे मैनेज करें, इसके बारे में और जानें—Snowpark के डेवलपर डॉक्यूमेंट्स के इस पेज से।

Snowpark में इंफरेंस करना

इंफरेंस आमतौर पर आपकी रजिस्ट्री से श्रेष्ठ मॉडल चुनकर किया जाता है। हम .show_models() के परिणाम को फ़िल्टर करके या उसके वर्शन टैग के साथ सीधे मॉडल रिट्रीव करके ऐसा कर सकते हैं, जैसा कि हम नीचे करते हैं:

# श्रेष्ठ मॉडल के साथ इंफरेंस करना
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 पर सबसे बड़े फॉलोइंग्स में से एक रखता हूँ। मुझे एआई और एमएल पर विस्तार से लेख लिखना पसंद है, जिसमें हल्का-सा व्यंग्यात्मक अंदाज रहता है—क्योंकि कुछ तो करना पड़ता है ताकि वे कम उबाऊ लगें। मैंने 130 से अधिक लेख और एक DataCamp कोर्स तैयार किया है, और एक और बन रहा है। मेरा कंटेंट 50 लाख से अधिक लोगों ने देखा है, जिनमें से 20 हज़ार लोग Medium और LinkedIn पर फॉलोअर्स बन गए।

विषय

बिग डेटा कृत्रिम बुद्धिमत्ता

प्रौद्योगिकियाँ

विषय

Snowflake Snowpark: एक व्यापक परिचय

AI के साथ खोजें

Snowflake Snowpark क्या है?

क्यों Snowflake Snowpark?

Snowpark के साथ शुरुआत

वर्चुअल एनवायरनमेंट बनाना

नमूना डेटा को Snowflake में इनजेस्ट करना

Snowpark सेशन बनाना

इनजेस्ट किए गए डेटा को सेशन में लोड करना

Snowpark DataFrames को समझना

Snowpark DataFrames लेज़ी होते हैं

Snowpark DataFrames कब उपयोग करें

Snowpark DataFrame ट्रांसफ़ॉर्मेशन फ़ंक्शंस

Snowpark DataFrames पर EDA करना

Snowpark में मशीन लर्निंग मॉडल ट्रेन करना

Snowpark में डेटा साफ़ करना

Snowpark में मॉडलिंग के लिए डेटा तैयार करना

Snowpark में XGBoost मॉडल ट्रेन करना

Snowpark में हाइपरपैरामीटर ट्यूनिंग

Snowpark में श्रेष्ठ मॉडल को सेव करना

Snowpark में इंफरेंस करना

निष्कर्ष

विषय

Snowflake और बिग डेटा के बारे में और जानें!

Snowflake SQL परिचय

डेटा इंजीनियर में Python

Snowflake में डेटा मॉडलिंग परिचय

प्रौद्योगिकियाँ

विषय

Snowflake और बिग डेटा के बारे में और जानें!

Snowflake SQL परिचय

डेटा इंजीनियर में Python

Snowflake में डेटा मॉडलिंग परिचय