Trainieren von Modellen über Datenpartitionen hinweg¶

Verwenden Sie Many Model Training (MMT), um mehrere Machine-Learning-Modelle effizient über Datenpartitionen hinweg zu trainieren. Die Funktion übernimmt automatisch die verteilte Orchestrierung, die Speicherung von Modellen und die Persistenz von Artefakten.

MMT unterteilt Ihren Snowpark DataFrame durch eine bestimmte Spalte und trainiert auf jeder Partition separate Modelle parallel. Konzentrieren Sie sich auf die Logik Ihres Modelltrainings, während MMT die Komplexität der Infrastruktur bewältigt und automatisch skaliert.

Sie können MMT verwenden, um mehrere Modelle über verschiedene Datensegmente hinweg effizient zu trainieren. Dieses Tool eignet sich ideal für Szenarios wie das Trainieren von regionsspezifischen Prognosemodellen, den Aufbau personalisierter Empfehlungssysteme, bei denen jede Kundengruppe ein eigenes Modell benötigt, oder die Erstellung segmentspezifischer, prädiktiver Modelle. MMT übernimmt das verteilte Modelltraining automatisch, wodurch die Komplexität der Verwaltung der verteilten Computing-Infrastruktur beseitigt wird.

Sie können MMT verwenden, um Modelle mit Open-Source-Modellen für Machine Learning und Frameworks zu trainieren, wie XGBoost, scikit-learn, PyTorchund TensorFlow. MMT serialisiert Modellartefakte automatisch, sodass Sie zum Zeitpunkt der Ableitung darauf zugreifen können.

Sie können auch die Schnittstelle ModelSerde implementieren, um kundenspezifische Modelle zu trainieren oder nicht unterstützte ML -Frameworks zu verwenden. Dies ermöglicht Ihnen die Integration von MMT mit jedem Machine Learning Framework oder kundenspezifischen Modellarchitekturen, die Sie verwenden.

Wichtig

Bevor Sie MMT verwenden, vergewissern Sie sich, dass Sie Folgendes haben:

Container-Laufzeitumgebung: MMT erfordert eine Snowflake ML Container-Laufzeitumgebung.
Staging-Zugriffsberechtigungen: MMT speichert Modellartefakte automatisch in Snowflake-Stagingbereichen. Stellen Sie sicher, dass Sie über die entsprechenden Berechtigungen für den Zugriff auf den angegebenen Stagingbereich verfügen.
ML Framework Support: Eingebettete Integrationen sind verfügbar für XGBoost, scikit-learn, PyTorch und TensorFlow. Für kundenspezifische Modelle implementieren Sie die ModelSerde-Schnittstelle.

Der folgende Abschnitt führt Sie durch die Verwendung von MMT in einem Beispiel-Workflow.

Trainieren eines Modells mit MMT¶

In diesem Abschnitt wird der vollständige MMT-Workflow in fünf Hauptschritten gezeigt:

Daten importieren: Laden Sie Trainingsdaten mit Snowpark
Trainingsfunktion definieren: Legen Sie die Trainingsfunktion fest
Modelle über Partitionen hinweg trainieren: Verwenden Sie MMT, um Modelle auf jeder Partition parallel zu trainieren
Auf trainierte Modelle zugreifen:Rufen Sie die trainierten Modelle für jede Partition ab und verwenden Sie sie
Modellpersistent und -abruf: Speichern Sie Modelle in Stagingbereichen und stellen Sie sie später wieder her

Der Workflow übernimmt automatisch das verteilte Training, die Modellserialisierung und die Speicherung von Artefakten über Ihre Datenpartitionen hinweg.

Ihre Daten importieren¶

Verwenden Sie eine Snowpark-Sitzung, um mit dem Import Ihrer Daten zu beginnen. Mit der Funktion Many Model Training werden die Daten, die Sie importieren, anhand der von Ihnen angegebenen Spalte in verschiedene Partitionen aufgeteilt.

Bevor Sie MMT verwenden, erstellen Sie eine Snowpark-Sitzung. Weitere Informationen dazu finden Sie unter Erstellen einer Sitzung für Snowpark Python.

Der folgende Code verwendet eine Snowpark-Sitzung, um Ihre Trainingsdaten zu importieren.

# Example: sales_data with columns: region, feature1, feature2, feature3, target
sales_data = session.table("SALES_TRAINING_DATA")

Copy

Definieren der Trainingsfunktion¶

Nachdem Sie Ihre Daten abgerufen haben, definieren Sie die Trainingsfunktion, die MMT verwendet, um Modelle über Partitionen hinweg zu trainieren. Die Trainingsfunktion erhält einen Datenkonnektor und ein Kontextobjekt, das auf die Datenpartition verweist, auf der sie trainiert wird. Dieser Abschnitt enthält Beispiele, die eine Trainingsfunktion zum Trainieren eines XGBoost-Modells definieren, zusätzlich zu Beispielen, bei denen TensorFlow und PyTorch verwendet werden.

Ihre Trainingsfunktion muss diese genaue Signatur haben: (data_connector, context). Für jede Datenpartition ruft MMT train_xgboost_model mit den folgenden Argumenten auf:

data_connector: Ein Datenkonnektor, der den Zugriff auf die Daten ermöglicht, die von MMT unterteilt werden. train_xgboost_model konvertiert diesen DataFrame in pandas.
context: Ein Objekt, das die partition_id für die Funktion train_xgboost_model bereitstellt. Diese ID ist der Name der Spalte, nach der Sie die Unterteilung vornehmen.

Sie rufen diese Funktion nicht selbst auf. MMT übernimmt die Ausführung über alle Partitionen hinweg.

Verwenden Sie den folgenden Code, um Ihre Trainingsfunktion zu definieren. Nachdem Sie den Code geändert haben, um die Funktionen in Ihren Daten widerzuspiegeln, können Sie ihn an die Funktion MMT übergeben.

Verwenden Sie XGBoost zum Trainieren von Modellen über Datenpartitionen hinweg. XGBoost bietet eine hervorragende Leistung für strukturierte Daten und verarbeitet fehlende Werte automatisch.

def train_xgboost_model(data_connector, context):
    df = data_connector.to_pandas()
    print(f"Training model for partition: {context.partition_id}")

    # Prepare features and target
    X = df[['feature1', 'feature2', 'feature3']]
    y = df['target']

    # Train the model
    from xgboost import XGBRegressor
    model = XGBRegressor(
        n_estimators=100,
        max_depth=6,
        learning_rate=0.1,
        random_state=42
    )
    model.fit(X, y)
    return model

trainer = ManyModelTraining(train_xgboost_model, "model_stage")

Copy

Verwenden Sie PyTorch zum Trainieren von Deep-Learning-Modellen über Datenpartitionen hinweg. PyTorch bietet flexible, neuronale Netzwerkarchitekturen und Diagramme zur dynamischen Rechenleistung.

def train_pytorch_model(data_connector, context):
    import torch
    import torch.nn as nn

    df = data_connector.to_pandas()
    # ... prepare data for PyTorch ...

    model = nn.Sequential(nn.Linear(10, 1))
    # ... training logic ...
    return model  # Automatically saved as model.pth

from snowflake.ml.modeling.distributors.many_model import TorchSerde
trainer = ManyModelTraining(train_pytorch_model, "models_stage", serde=TorchSerde())

Copy

Verwenden Sie TensorFlow zum Trainieren von Deep-Learning-Modellen über Datenpartitionen hinweg. TensorFlow bietet umfassende Tools sowohl für die Forschung als auch für den Einsatz in der Produktion.

def train_tf_model(data_connector, context):
    import tensorflow as tf

    df = data_connector.to_pandas()
    # ... prepare data for TensorFlow ...

    model = tf.keras.Sequential([tf.keras.layers.Dense(1)])
    # ... training logic ...
    return model  # Automatically saved as model.h5

from snowflake.ml.modeling.distributors.many_model import TensorFlowSerde
trainer = ManyModelTraining(train_tf_model, "models_stage", serde=TensorFlowSerde())

Copy

Verwenden Sie kundenspezifische Modelle oder nicht unterstützte ML-Frameworks durch Implementierung der ModelSerde-Schnittstelle. Dieses Beispiel zeigt scikit-learn bei der Verarbeitung kundenspezifischer Metadaten.

from snowflake.ml.modeling.distributors.many_model import ModelSerde
import json

class ScikitLearnSerde(ModelSerde):
    '''Custom serializer for scikit-learn models with metadata'''

    @property
    def filename(self) -> str:
        return "sklearn_model.joblib"

    def write(self, model, file_path: str) -> None:
        import joblib
        # Save model with metadata
        model_data = {
            'model': model,
            'feature_names': getattr(model, 'feature_names_in_', None),
            'model_type': type(model).__name__
        }
        joblib.dump(model_data, file_path)

    def read(self, file_path: str):
        import joblib
        return joblib.load(file_path)

def train_sklearn_model(data_connector, context):
    from sklearn.ensemble import RandomForestRegressor
    df = data_connector.to_pandas()
    X, y = df[['feature1', 'feature2']], df['target']

    model = RandomForestRegressor()
    model.fit(X, y)
    return model  # Automatically saved with metadata

trainer = ManyModelTraining(train_sklearn_model, "models_stage", serde=ScikitLearnSerde())

Copy

Trainieren von Modellen über Partitionen hinweg¶

Nachdem Sie Ihre Trainingsfunktion definiert haben, können Sie MMT zum Trainieren von Modellen über Partitionen hinweg verwenden. Geben Sie die Spalte an, nach der die Einteilung vorgenommen werden soll, und den Stagingbereich, in dem die Modelle gespeichert werden.

Der folgende Code unterteilt die Daten nach der Spalte region und verwendet die Funktion train_xgboost_model, um parallel separate Modelle für jede Region zu trainieren.

Wenn beispielsweise die folgenden Werte für die Spalte region möglich wären:

North
South
East
West
Central

Die Funktion ManyModelTraining würde für jede der vorhergehenden Regionen eine separate Datenpartition erstellen und mit jeder Partition ein Modell trainieren.

from snowflake.ml.modeling.distributors.many_model import ManyModelTraining

trainer = ManyModelTraining(train_xgboost_model, "model_stage") # Specify the stage to store the models
training_run = trainer.run(
    partition_by="region",  # Train separate models for each region
    snowpark_dataframe=sales_data,
    run_id="regional_models_v1" # Specify a unique ID for the training run
)

# Monitor training progress
final_status = training_run.wait()
print(f"Training completed with status: {final_status}")

Copy

Modelle werden im Stagingbereich unter run_id/{partition_id} gespeichert, wobei partition_id der Wert der Partitionsspalte ist.

Zugriff auf trainierte Modelle¶

Nachdem MMT beendet ist, haben Sie Modelle für jede Datenpartition trainiert, die in dem von Ihnen angegebenen Stagingbereich gespeichert ist. Jedes Modell wird mit den für seine Partition spezifischen Daten trainiert. Zum Beispiel wird ein North-Modell nur mit Daten der Region North trainiert.

Das Trainingsausführungsobjekt bietet Methoden, um auf diese Modelle zuzugreifen und den Trainingsstatus für jede Partition zu überprüfen.

Der folgende Code ruft den Status des Trainingslaufs und die trainierten Modelle für jede Partition ab:

if final_status == RunStatus.SUCCESS:
    # Access models for each partition
    for partition_id in training_run.partition_details:
        trained_model = training_run.get_model(partition_id)
        print(f"Model for {partition_id}: {trained_model}")

        # You can now use the model for predictions or further analysis
        # Example: model.predict(new_data)
else:
    # Handle training failures
    for partition_id, details in training_run.partition_details.items():
        if details.status != "DONE":
            print(f"Training failed for {partition_id}")
            error_logs = details.logs

Copy

Modellpersistenz und -abruf¶

MMT setzt trainierte Modelle während des Trainingsprozesses automatisch in den von Ihnen angegebenen Snowflake-Stagingbereich fort. Jedes Modell wird mit einem strukturierten Pfad gespeichert, der die Ausführungs-ID und Partitionsbezeichner beinhaltet, damit Modelle später problemlos organisiert und abgerufen werden können.

Die automatische Persistenz bedeutet, dass Sie Modelle nicht manuell speichern müssen. MMT übernimmt die Serialisierung und Speicherung für Sie und beseitigt das Risiko, trainierte Modelle aufgrund von Sitzungs-Timeouts oder Verbindungsproblemen zu verlieren.

Sie können frühere Trainingsläufe wiederherstellen und auf deren Modelle zugreifen, auch nachdem Ihre ursprüngliche Sitzung beendet wurde. Mit diesem Persistenzmechanismus können Sie Folgendes tun:

Ihre Arbeit über verschiedene Sitzungen hinweg fortsetzen
Trainierte Modelle mit Teammitgliedern teilen
Workflows für die Versionierung von Modellen erstellen
Eine Integration in nachgelagerte Inferenz-Pipelines vornehmen

Modelle werden automatisch im angegebenen Stagingbereich gespeichert und können später abgerufen werden:

# Restore training run from stage
restored_run = ManyModelTraining.restore_from("regional_models_v1", "model_stage")

# Access models from restored run
north_model = restored_run.get_model("North")
south_model = restored_run.get_model("South")

Copy

Trainieren von kundenspezifischen Modellen¶

Für kundenspezifische Modelle oder nicht unterstützte ML-Frameworks implementieren die die Schnittstelle ModelSerde. Sie können Ihre eigene Serialisierungs- und Deserialisierungslogik für kundenspezifische Modelle definieren. Dies ermöglicht Ihnen die Integration von MMT mit jedem Machine Learning Framework oder kundenspezifischen Modellarchitekturen, die Sie verwenden.

from snowflake.ml.modeling.distributors.many_model import ModelSerde

class CustomModelSerde(ModelSerde):
    def serialize(self, model, path):
        # Custom serialization logic
        pass

    def deserialize(self, path):
        # Custom deserialization logic
        pass

def train_custom_model(data_connector, context):
    # Your custom training logic
    model = your_custom_model_training(data_connector.to_pandas())
    return model

trainer = ManyModelTraining(
    train_custom_model,
    "custom_model_stage",
    model_serde=CustomModelSerde()
)

Copy

Integration mit Model Registry¶

MMT kann in Model Registry von Snowflake integriert werden, um die Verwaltung von Modellen zu verbessern. Model Registry bietet eine zentrale Versionierung von Modellen, Metadatenverfolgung und Bereitstellungsverwaltung in Ihrer Organisation. Diese Integration ist besonders nützlich, wenn mehrere Modelle mit MMT trainiert werden, da sie Ihnen hilft, alle partitionsspezifischen Modelle von einem einzigen Ort aus zu organisieren, zu verfolgen und zu verwalten.

Durch die Verwendung von Model Registry mit MMT können Sie Folgendes tun:

Verfolgen von verschiedenen Iterationen Ihrer partitionsspezifischen Modelle
Speichern von Modellleistungsmetriken, Trainingsparametern und Herkunftsinformationen
Verwaltung, welche Modellversionen für jede Partition in der Produktion bereitgestellt werden
Freigeben von Modellen über Teams hinweg mit geeigneten Zugriffssteuerungen und Dokumentation
Implementieren von Genehmigungsworkflows und Compliance-Nachverfolgung für die Bereitstellung von Modellen

# Register trained models to Model Registry
for partition_id in training_run.partition_details:
    model = training_run.get_model(partition_id)

    # Register to Model Registry
    model_ref = registry.log_model(
        model,
        model_name=f"sales_model_{partition_id.lower()}",
        version_name="v1"
    )

Copy