Pipelines erstellen und bereitstellen¶

Code vorbereiten¶

Bevor Sie Ihre Pipeline operationalisieren, bereiten Sie Ihren Code für die Produktion vor. Wenn Sie mit Notebooks begonnen haben, beginnen Sie mit der Umstrukturierung Ihres Codes in modulare, wiederverwendbare Funktionen, bei denen jeder größere Schritt (Datenaufbereitung, Feature-Engineering, Modelltraining, Evaluation) zu einer separaten Funktion mit klaren Eingaben und Ausgaben wird. Wenn Sie bereits modulare Skripte verwenden, stellen Sie sicher, dass jede Funktion über genau definierte Schnittstellen und Verantwortlichkeiten verfügt. Parametrisieren Sie alle Konfigurationswerte wie Tabellennamen und Hyperparameter, um eine umgebungsübergreifende Bereitstellung zu ermöglichen. Wir empfehlen auch, ein Einstiegspunkt-Skript zu erstellen, das die End-to-End-Pipeline lokal für Debugging und zukünftige Entwicklungen ausführt.

Beispiel einer Verzeichnisstruktur:

ml_pipeline_project/
├── README.md
├── requirements.txt
├── config/
├── src/ml_pipeline/
│   ├── utils/                     # Common utilities
│   ├── data/                      # Data preparation
│   ├── features/                  # Feature engineering
│   ├── models/                    # Model training
│   └── inference/                 # Model inference
├── scripts/
│   ├── run_pipeline.py            # Main entry point
│   └── dag.py
├── tests/
└── notebooks/

Beispielskript „run_pipeline.py“:

import argparse
from ml_pipeline.utils.config_loader import load_config
from ml_pipeline.data.ingestion import load_raw_data
from ml_pipeline.data.validation import validate_data_quality
from ml_pipeline.features.transformers import create_features
from ml_pipeline.models.training import train_model
from ml_pipeline.models.evaluation import evaluate_model
from ml_pipeline.models.registry import register_model

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", required=True, help="Config file path")
    parser.add_argument("--env", default="dev", help="Environment (dev/prod)")
    args = parser.parse_args()

    # Load configuration
    config = load_config(args.config, args.env)

    # Execute pipeline stages
    raw_data = load_raw_data(config.data.source_table)
    validate_data_quality(raw_data, config.data.quality_checks)
    features = create_features(raw_data, config.features.transformations)
    model = train_model(features, config.model.hyperparameters)
    metrics = evaluate_model(model, features, config.model.eval_metrics)
    register_model(model, metrics, config.model.registry_name)

if __name__ == "__main__":
    main()

Migration von Notebooks zu ML-Jobs¶

Der Großteil des Codes, der in Snowflake Notebooks geschrieben ist, funktioniert in ML-Jobs, ohne dass Codeänderungen erforderlich sind. Die folgenden Aspekte sollten Sie beachten:

Laufzeit-APIs

Bestimmte verteilte ML APIs sind nur innerhalb der Container Runtime verfügbar. Der Versuch, sie außerhalb der Container Runtime-Umgebung zu importieren, schlägt fehl. Diese APIs sind innerhalb von ML-Jobs verfügbar, müssen aber innerhalb der Nutzlast des ML-Jobs importiert werden.

# Attempting to import distributed runtime APIs in local/external
# environments will fail!
from snowflake.ml.modeling.distributors.xgboost import XGBEstimator

from snowflake.ml.jobs import remote

@remote(...)
def my_remote_function(...):
  # Move imports *inside* your ML Job payloads
  from snowflake.ml.modeling.distributors.xgboost import XGBEstimator  # This works!
  ...

job = my_remote_function()  # Start ML Job
job.wait()  # Wait for job to complete

Cluster-Skalierung

Das scale_cluster() API funktioniert nur innerhalb von Notebooks und nicht innerhalb von ML-Jobs. Geben Sie stattdessen die gewünschte Clustergröße zum Zeitpunkt der Übermittlung des Jobs an. Weitere Informationen dazu finden Sie unter Snowflake mit ML-Jobs mit mehreren Knoten.

from snowflake.ml.jobs import remote

@remote(..., target_instances=4)
def my_remote_function(...):
  # 4-node cluster will be provisioned for distributed processing
  # inside this job. The cluster will be automatically cleaned up on
  # job termination.

Pipeline-Orchestrierung¶

Sobald Sie Ihre End-to-End-Pipeline vorbereitet haben, operationalisieren Sie diese mit einem Orchestrator wie Snowflake Task-Graphen, Scheduled Notebooks oder externen Orchestratoren wie Airflow. Die Verwendung eines Orchestrierungs-Frameworks bietet mehrere entscheidende Vorteile:

• Fehlertoleranz und Zuverlässigkeit durch automatische Wiederholungsversuche und Fehlerisolation

• Beobachtbarkeit mit Ausführungsverlauf, Echtzeitstatus und Alerts

• Planung und Koordination von komplexen Abhängigkeitsdiagrammen und verschiedenen Triggern

• Operative Optimierung mit Versionskontrollintegration und Konfigurationsmanagement

Snowflake ML ist mit den meisten Orchestrierungs-Frameworks kompatibel, einschließlich Airflow, Dagster und Prefect. Wenn Sie bereits ein Workflow-/DAG-Setup haben, empfehlen wir, einfach Ihre bestehenden Workflows in Snowflake ML-Features zu integrieren und rechen- oder datenintensive Schritte in ML-Jobs oder UDFs auszulagern. Wenn Sie keine bestehendes DAG-Setup haben, können Sie Snowflake Task-Graphen für eine native Snowflake-Lösung verwenden.

So richten Sie die Orchestrierung mit einer DAG auf Snowflake ein:

1. Bereiten Sie Ihren lokalen Pipeline-Code gemäß Code vorbereiten vor

2. Erstellen Sie eine neue dag.py-Datei (oder ein anderer Name), die Ihre DAG-Definition enthält

3. Implementieren Sie die DAG aus Ihrer Pipeline gemäß dieser Anleitung

4. Führen Sie das dag.py-Skript aus, um den Task-Graphen in Ihrem Snowflake-Konto bereitzustellen

Trennung von Entwicklung und Produktion¶

Wir empfehlen, Ihr DAG-Skript zur Unterstützung der Isolierung Ihrer Entwicklungs(DEV)- und Produktions(PROD)-Umgebungen zu parametrisieren. Sie können dazu die Verbindungsverwaltung von Snowflake, anwendungsspezifische Konfigurationen oder eine beliebige Kombination der beiden verwenden. Das erforderliche Maß an Isolation hängt von Ihren Governance-Anforderungen ab, aber generell empfehlen wir die Verwendung separater Datenbanken für DEV und PROD, wobei die PROD-Datenbank durch RBAC-Richtlinien geschützt wird, die den Zugriff auf Administratoren und spezielle Dienstkonten beschränken.

CI/CD¶

Sie können die Validierung und Bereitstellung Ihrer Pipelines mit CI/CD-Pipelines wie Azure Pipelines und GitHub-Aktionen automatisieren. Im Allgemeinen empfehlen wir das Testen in einer DEV- oder STAGING-Umgebung, bevor die Bereitstellung in PROD erfolgt. Es empfiehlt sich, Ihr Quellkontroll-Repository mit Merge-Gateways zu konfigurieren, die Codeänderungen vor der Zusammenführung mit Ihrem Produktionszweig in DEV validieren. Änderungen am Produktionszweig können in PROD kontinuierlich (d. h. bei jeder Änderung) oder regelmäßig (täglich/wöchentlich) bereitgestellt werden. Es empfiehlt sich, eine abschließende Validierung des Status des Produktionszweigs in einer DEV- oder STAGING-Umgebung durchzuführen, bevor Änderungen in PROD bereitgestellt werden. Verwenden Sie Plattform-Features wie GitHub-Aktionsbereitstellungen und -Umgebungen zum Definieren und Konfigurieren von Verbindungen zu jeder Bereitstellungsumgebung. Konfigurieren Sie Ihre CI/CD-Pipeline, um Ihre Änderungen in die Bereitstellungsumgebung zu übertragen, einschließlich:

• (Optional) Erstellen von Bibliotheken und Modulen als Python-Pakete und deren Verwendung in einem proprietären Paket-Feed

• (Optional) Hochladen von Dateien in einen Snowflake-Stagingbereich

  Dies ist am häufigsten erforderlich, wenn Sie snowflake.ml.jobs.submit_from_stage() in Ihrer Pipeline verwenden

  Alternativ können Sie auch die Snowflake GitHub-Integration verwenden, um Ihr GitHub-Repository direkt als Snowflake-Stagingbereich zu verfolgen

• Ausführung von dag.py um den Task-Graphen in der konfigurierten Umgebung bereitzustellen

• (Optional) Auslösen und Überwachen der Ausführung des neu bereitgestellten Task-Graphen, um die Gültigkeit zu überprüfen