[新機能]モデルの実行情報を収集し比較できる「Snowflake ML Experiments」がパブリックプレビューとなったので試してみた

-- 自分のデータベース・スキーマを作成
CREATE DATABASE IF NOT EXISTS SNOWFLAKE_ML;
CREATE SCHEMA IF NOT EXISTS SNOWFLAKE_ML.EXPERIMENTS;
CREATE SCHEMA IF NOT EXISTS SNOWFLAKE_ML.titanic;

-- タイタニックのデータを格納する内部ステージを作成
CREATE STAGE titanic_stage
	DIRECTORY = ( ENABLE = true )
	ENCRYPTION = ( TYPE = 'SNOWFLAKE_SSE' );

import pandas as pd
from snowflake.snowpark.files import SnowflakeFile
from snowflake.snowpark import Session

# 1. train.csv の読み込み
# SnowflakeFile.open() を使ってファイルストリームを開き、pandasに渡します
with SnowflakeFile.open("@titanic_stage/train.csv") as f:
    train_df = pd.read_csv(f)

# 2. test.csv の読み込み
with SnowflakeFile.open("@titanic_stage/test.csv") as f:
    test_df = pd.read_csv(f)

# 確認
print("Train shape:", train_df.shape)
print("Test shape:", test_df.shape)
print(train_df.head(3))

## 必要なライブラリのインポート
import pandas as pd
from lightgbm import LGBMClassifier
from snowflake.snowpark import Session
from snowflake.ml.experiment import ExperimentTracking 
from snowflake.ml.experiment.callback.lightgbm import SnowflakeLightgbmCallback
from snowflake.ml.model.model_signature import infer_signature
from sklearn.model_selection import train_test_split

# 1. セッションの取得 (Notebook環境)
session = get_active_session()

# 2. データ準備 
features = ["Pclass", "Sex", "SibSp", "Parch"]
X = pd.get_dummies(train_df[features])
y = train_df["Survived"]
X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=1)

# 3. ExperimentTracking のインスタンス化
# ここで指定したいデータベースを選ぶ
session.use_database("SNOWFLAKE_ML")
# ここで指定したいスキーマを選ぶ
session.use_schema("EXPERIMENTS")
exp = ExperimentTracking(session=session)
# experimentの指定
exp.set_experiment("TITANIC_TEST")

# 4. モデル署名 (Signature) の作成
sig = infer_signature(X_train, y_train)

# 5. コールバックの設定
# 第1引数に `exp` (ExperimentTrackingインスタンス) を渡します
callback = SnowflakeLightgbmCallback(
    exp, 
    model_name="titanic_lgbm_auto_v2", 
    model_signature=sig
)

# 6. モデル定義
model = LGBMClassifier(
    n_estimators=100,
    learning_rate=0.05,
    random_state=42,
    verbose=-1
)

# 7. 学習と自動記録の実行 exp.start_run() を使ってコンテキストを開始します
print("Starting training with ExperimentTracking...")

with exp.start_run("callback_run_v2"):
    model.fit(
        X_train, y_train,
        eval_set=[(X_val, y_val)],
        eval_metric="binary_logloss",
        callbacks=[callback] # ここで自動記録
    )

print("Training finished. Check Experiments UI.")

import pandas as pd
from lightgbm import LGBMClassifier
from snowflake.snowpark import Session
from snowflake.ml.experiment import ExperimentTracking #★ここが重要
from snowflake.ml.experiment.callback.lightgbm import SnowflakeLightgbmCallback
from snowflake.ml.model.model_signature import infer_signature
from sklearn.model_selection import train_test_split

# --- パターン2: 特徴量(Age, Fare)を追加して再実験 ---

# 1. データの作り直し (AgeとFareを追加)
features_v2 = ["Pclass", "Sex", "SibSp", "Parch", "Age", "Fare"]
X_v2 = pd.get_dummies(train_df[features_v2]) # カテゴリ変数の処理
y_v2 = train_df["Survived"]

# データの分割 (乱数シードは同じにして公平に比較)
X_train_v2, X_val_v2, y_train_v2, y_val_v2 = train_test_split(X_v2, y_v2, test_size=0.2, random_state=1)

# ExperimentTracking のインスタンス化
session.use_database("SNOWFLAKE_ML")
# ここで指定したいスキーマを選ぶ
session.use_schema("EXPERIMENTS")
exp = ExperimentTracking(session=session)
exp.set_experiment("TITANIC_TEST")

# 2. 新しいデータに合わせてSignatureを更新
sig_v2 = infer_signature(X_train_v2, y_train_v2)

# 3. コールバックも新しく作り直す (Signatureが変わるため)
callback_v2 = SnowflakeLightgbmCallback(
    exp, 
    model_name="titanic_lgbm_more_features", 
    model_signature=sig_v2
)

# 4. モデル定義 (パラメータは標準に戻して比較)
model_features = LGBMClassifier(
    n_estimators=100,
    learning_rate=0.05,
    random_state=42,
    verbose=-1
)

# 5. 実験の実行
run_name_features = "LGBM_More_Features"
print(f"Starting run: {run_name_features}...")

with exp.start_run(run_name=run_name_features):
    model_features.fit(
        X_train_v2, y_train_v2,
        eval_set=[(X_val_v2, y_val_v2)],
        eval_metric="binary_logloss",
        callbacks=[callback_v2]
    )

print("Feature-rich run finished.")

import pandas as pd
from snowflake.ml.experiment import ExperimentTracking
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# データ準備 
features = ["Pclass", "Sex", "SibSp", "Parch"]
X = pd.get_dummies(train_df[features])
y = train_df["Survived"]
X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=1)

# 1. ExperimentTrackingの準備
session = get_active_session()
# ExperimentTracking のインスタンス化
session.use_database("SNOWFLAKE_ML")
# ここで指定したいスキーマを選ぶ
session.use_schema("EXPERIMENTS")
exp = ExperimentTracking(session=session)
exp.set_experiment("TITANIC_TEST")

run_name_2 = "RandomForest_Manual"

# 2. 実験の開始
with exp.start_run(run_name=run_name_2):

    print(f"Starting run: {run_name_2}")

    # --- モデルの設定 ---
    n_estimators = 100
    model = RandomForestClassifier(n_estimators=n_estimators, random_state=42)

    # --- 学習 ---
    model.fit(X_train, y_train)

    # --- 予測と評価 ---
    preds = model.predict(X_val)
    acc = accuracy_score(y_val, preds)

    # パラメータの記録 (exp.log_param)
    exp.log_param("model_type", "RandomForest")
    exp.log_param("n_estimators", n_estimators)

    # メトリクスの記録 (exp.log_metric)
    exp.log_metric("accuracy", acc)

    # タグの記録 
    exp.log_params({"description": "Challenger model using exp methods"})