Tutorial 5: Entwickeln eines Featuresatzes mit einer benutzerdefinierten Quelle

Mit einem Azure Machine Learning verwalteten Featurespeicher können Sie Features ermitteln, erstellen und operationalisieren. Features fungieren im Lebenszyklus des maschinellen Lernens als Bindeglieder. Dieser beginnt mit der Phase für die Prototyperstellung, in der Sie mit verschiedenen Features experimentieren. Dieser Lebenszyklus wird mit der Produktionsphase fortgesetzt, in der Sie Ihre Modelle bereitstellen und Inferenzschritte die Feature-Daten abrufen. Weitere Informationen zu Feature Stores finden Sie in der Ressource Feature Store-Konzepte.

In Teil 1 dieser Tutorialreihe wurde gezeigt, wie Sie eine Spezifikation für einen Featuresatz mit benutzerdefinierten Transformationen erstellen, die Materialisierung aktivieren und einen Abgleich durchführen. In Teil 2 wurde gezeigt, wie Sie mit Features in den Experimentier- und Trainingsflows experimentieren können. Teil 3 erläuterte die wiederkehrende Materialisierung für den Featuresatz transactions und zeigte, wie eine Batch-Rückschlusspipeline für das registrierte Modell ausgeführt wird. In Teil 4 wurde beschrieben, wie die Batch-Inferenz ausgeführt wird.

In diesem Tutorial führen Sie folgende Schritte aus:

Voraussetzungen

• Vergewissern Sie sich, dass Sie die vorausgehenden Tutorials in dieser Reihe abgeschlossen haben. In diesem Tutorial werden der Feature Store und andere Ressourcen wiederverwendet, die in den vorherigen Tutorials erstellt wurden.

Einrichten

In diesem Lernprogramm wird das Python Feature store Core SDK (azureml-featurestore) verwendet. Das Python SDK wird für Erstellungs-, Lese-, Aktualisierungs- und Löschvorgänge (CRUD) für Featurespeicher, Featuresätze und Featurespeicherentitäten verwendet.

Sie müssen diese Ressourcen für dieses Tutorial nicht explizit installieren, da sie in den hier gezeigten Setupanweisungen durch die Datei conda.yml abgedeckt werden.

Konfigurieren des Azure Machine Learning Spark-Notizbuchs

Sie können ein neues Notebook erstellen und die Anweisungen in diesem Tutorial Schritt für Schritt ausführen. Sie können auch das vorhandene Notebook featurestore_sample/notebooks/sdk_only/5.Develop-feature-set-custom-source.ipynb öffnen und ausführen. Halten Sie dieses Tutorial geöffnet und nutzen Sie es für Dokumentationslinks und weitere Erläuterungen.

1. Wählen Sie im oberen Menü in der Dropdownliste ComputeServerless Spark Compute unter Azure Machine Learning Serverless Spark aus.

2. Konfigurieren der Sitzung:

   1. Wählen Sie in der oberen Statusleiste Sitzung konfigurieren aus.
   2. Wählen Sie die Registerkarte Python pakete aus, und wählen Sie Upload Conda-Datei
   3. Wählen Sie Conda-Datei hochladen aus.
   4. Laden Sie die Datei conda.yml hoch, die Sie im ersten Tutorial hochgeladen haben.
   5. Optional: Erhöhen Sie das Sitzungstimeout (die Leerlaufzeit), um häufige Wiederholungen der Überprüfung von Voraussetzungen zu vermeiden.

Richten Sie das Stammverzeichnis für die Beispiele ein.

Diese Codezelle richtet das Stammverzeichnis für die Stichproben ein. Es dauert etwa 10 Minuten, um alle Abhängigkeiten zu installieren und die Spark-Sitzung zu starten.

import os

# Please update the dir to ./Users/{your_user_alias} (or any custom directory you uploaded the samples to).
# You can find the name from the directory structure in the left navigation panel.
root_dir = "./Users/<your_user_alias>/featurestore_sample"

if os.path.isdir(root_dir):
    print("The folder exists.")
else:
    print("The folder does not exist. Please create or fix the path")

Initialisieren Sie den CRUD-Client des Feature-Store-Arbeitsbereichs

Initialisieren Sie MLClient für den Feature Store-Arbeitsbereich, um die Vorgänge zum Erstellen, Lesen, Aktualisieren und Löschen (Create, Read, Update, Delete – CRUD) des Feature Store-Arbeitsbereichs abzudecken.

from azure.ai.ml import MLClient
from azure.ai.ml.identity import AzureMLOnBehalfOfCredential

# Feature store
featurestore_name = (
    "<FEATURESTORE_NAME>"  # use the same name that was used in the tutorial #1
)
featurestore_subscription_id = os.environ["AZUREML_ARM_SUBSCRIPTION"]
featurestore_resource_group_name = os.environ["AZUREML_ARM_RESOURCEGROUP"]

# Feature store ml client
fs_client = MLClient(
    AzureMLOnBehalfOfCredential(),
    featurestore_subscription_id,
    featurestore_resource_group_name,
    featurestore_name,
)

Initialisieren des primären SDK-Clients für den Feature Store

Wie bereits erwähnt, verwendet dieses Lernprogramm das Python Feature store Core SDK (azureml-featurestore). Dieser initialisierte SDK-Client deckt die Vorgänge zum Erstellen, Lesen, Aktualisieren und Löschen (CRUD) in Feature Stores, Featuresätzen und Feature-Store-Entitäten ab.

from azureml.featurestore import FeatureStoreClient
from azure.ai.ml.identity import AzureMLOnBehalfOfCredential

featurestore = FeatureStoreClient(
    credential=AzureMLOnBehalfOfCredential(),
    subscription_id=featurestore_subscription_id,
    resource_group_name=featurestore_resource_group_name,
    name=featurestore_name,
)

Benutzerdefinierte Quellendefinition

Sie können Ihre eigene Logik für das Laden von Quellen aus jedem Datenspeicher definieren, der über eine benutzerdefinierte Quellendefinition verfügt. Implementieren Sie eine Klasse für benutzerdefinierte Funktionen (UDF) des Quellprozessors (CustomSourceTransformer in diesem Tutorial), um dieses Feature zu verwenden. Diese Klasse sollte eine __init__(self, **kwargs)- und eine process(self, start_time, end_time, **kwargs)-Funktion definieren. Das kwargs-Wörterbuch wird als Teil der Definition der Featuresatzspezifikation bereitgestellt. Diese Definition wird dann an die UDF übergeben. Die Parameter start_time und end_time werden berechnet und an die UDF-Funktion übergeben.

Dies ist Beispielcode für die UDF-Klasse des Quellprozessors:

from datetime import datetime

class CustomSourceTransformer:
    def __init__(self, **kwargs):
        self.path = kwargs.get("source_path")
        self.timestamp_column_name = kwargs.get("timestamp_column_name")
        if not self.path:
            raise Exception("`source_path` is not provided")
        if not self.timestamp_column_name:
            raise Exception("`timestamp_column_name` is not provided")

    def process(
        self, start_time: datetime, end_time: datetime, **kwargs
    ) -> "pyspark.sql.DataFrame":
        from pyspark.sql import SparkSession
        from pyspark.sql.functions import col, lit, to_timestamp

        spark = SparkSession.builder.getOrCreate()
        df = spark.read.json(self.path)

        if start_time:
            df = df.filter(col(self.timestamp_column_name) >= to_timestamp(lit(start_time)))

        if end_time:
            df = df.filter(col(self.timestamp_column_name) < to_timestamp(lit(end_time)))

        return df

Erstellen einer Featuresatzspezifikation mit einer benutzerdefinierten Quelle und lokales Experimentieren

Erstellen Sie nun eine Featuresatzspezifikation mit einer benutzerdefinierten Quellendefinition, und verwenden Sie sie in Ihrer Entwicklungsumgebung, um mit dem Featuresatz zu experimentieren. Das Tutorial-Notebook, das an serverloses Spark Compute angefügt ist, dient als Entwicklungsumgebung.

from azureml.featurestore import create_feature_set_spec
from azureml.featurestore.feature_source import CustomFeatureSource
from azureml.featurestore.contracts import (
    SourceProcessCode,
    TransformationCode,
    Column,
    ColumnType,
    DateTimeOffset,
    TimestampColumn,
)

transactions_source_process_code_path = (
    root_dir
    + "/featurestore/featuresets/transactions_custom_source/source_process_code"
)
transactions_feature_transform_code_path = (
    root_dir
    + "/featurestore/featuresets/transactions_custom_source/feature_process_code"
)

udf_featureset_spec = create_feature_set_spec(
    source=CustomFeatureSource(
        kwargs={
            "source_path": "wasbs://data@azuremlexampledata.blob.core.windows.net/feature-store-prp/datasources/transactions-source-json/*.json",
            "timestamp_column_name": "timestamp",
        },
        timestamp_column=TimestampColumn(name="timestamp"),
        source_delay=DateTimeOffset(days=0, hours=0, minutes=20),
        source_process_code=SourceProcessCode(
            path=transactions_source_process_code_path,
            process_class="source_process.CustomSourceTransformer",
        ),
    ),
    feature_transformation=TransformationCode(
        path=transactions_feature_transform_code_path,
        transformer_class="transaction_transform.TransactionFeatureTransformer",
    ),
    index_columns=[Column(name="accountID", type=ColumnType.string)],
    source_lookback=DateTimeOffset(days=7, hours=0, minutes=0),
    temporal_join_lookback=DateTimeOffset(days=1, hours=0, minutes=0),
    infer_schema=True,
)

udf_featureset_spec

Definieren Sie als nächstes ein Featurefenster, und zeigen Sie die Featurewerte in diesem Featurefenster an.

from datetime import datetime

st = datetime(2023, 1, 1)
et = datetime(2023, 6, 1)

display(
    udf_featureset_spec.to_spark_dataframe(
        feature_window_start_date_time=st, feature_window_end_date_time=et
    )
)

Exportieren als Featuresatzspezifikation

Um die Featuresatzspezifikation beim Feature Store zu registrieren, speichern Sie diese Featuresatzspezifikation in einem bestimmten Format. Überprüfen Sie die Spezifikation des generierten transactions_custom_source-Featuresatzes. Öffnen Sie diese Datei aus dem Dateibaum, um die Spezifikation anzuzeigen: featurestore/featuresets/transactions_custom_source/spec/FeaturesetSpec.yaml.

Die Spezifikation weist folgende Elemente auf:

• features: eine Liste der Features und ihrer Datentypen.
• index_columns: die Joinschlüssel, die für den Zugriff auf Werte aus dem Funktionssatz erforderlich sind.

Weitere Informationen zur Spezifikation finden Sie in den Ressourcen Grundlegendes zu Entitäten auf oberster Ebene im verwalteten Feature Store und YAML-Schema für Featuresätze der CLI (v2).

Das Beibehalten der Featuresatzspezifikation bietet einen weiteren Vorteil: Für die Featuresatzspezifikation kann die Quellcodeverwaltung verwendet werden.

feature_spec_folder = (
    root_dir + "/featurestore/featuresets/transactions_custom_source/spec"
)

udf_featureset_spec.dump(feature_spec_folder)

Registrieren des Transaktionsfeaturesatzes beim Feature Store

Verwenden Sie diesen Code, um ein Featuresatzobjekt zu registrieren, das aus einer benutzerdefinierten Quelle mit dem Feature Store geladen wurde. Sie können diese Ressource dann wiederverwenden und problemlos freigeben. Die Registrierung einer Featuresatzressource bietet verwaltete Funktionen, einschließlich Versionsverwaltung und Materialisierung.

from azure.ai.ml.entities import FeatureSet, FeatureSetSpecification

transaction_fset_config = FeatureSet(
    name="transactions_custom_source",
    version="1",
    description="transactions feature set loaded from custom source",
    entities=["azureml:account:1"],
    stage="Development",
    specification=FeatureSetSpecification(path=feature_spec_folder),
    tags={"data_type": "nonPII"},
)

poller = fs_client.feature_sets.begin_create_or_update(transaction_fset_config)
print(poller.result())

Rufen Sie den registrierten Merkmalsatz ab. Geben Sie dann die dazugehörigen Informationen aus.

# Look up the feature set by providing name and version
transactions_fset_config = featurestore.feature_sets.get(
    name="transactions_custom_source", version="1"
)
# Print feature set information
print(transactions_fset_config)

Testen der Featuregenerierung aus dem registrierten Featuresatz

Verwenden Sie die to_spark_dataframe()-Funktion des Featuresatzes, um die Featuregenerierung aus dem registrierten Featuresatz zu testen und die Features anzuzeigen. print-txn-fset-sample-values

df = transactions_fset_config.to_spark_dataframe()
display(df)

Sie sollten in der Lage sein, den registrierten Featuresatz als Spark-Datenrahmen erfolgreich abzurufen und dann anzuzeigen. Sie können diese Features jetzt für einen Point-in-Time-Join mit Beobachtungsdaten und den nachfolgenden Schritten in Ihrer Machine Learning-Pipeline verwenden.

Bereinigen

Wenn Sie eine Ressourcengruppe für das Tutorial erstellt haben, können Sie diese Ressourcengruppe löschen, um alle diesem Tutorial zugeordneten Ressourcen zu löschen. Andernfalls können Sie die Ressourcen einzeln löschen:

• Um den Featurespeicher zu löschen, öffnen Sie die Ressourcengruppe im Azure-Portal, wählen Sie den Featurespeicher aus, und löschen Sie sie.
• Die dem Feature Store-Arbeitsbereich zugewiesene verwaltete Identität (User-Assigned Managed Identity, UAI) wird beim Löschen des Feature Stores nicht gelöscht. Befolgen Sie diese Anweisungen, um die UAI zu löschen.
• Um einen Offlinespeicher vom Typ "Speicherkonto" zu löschen, öffnen Sie die Ressourcengruppe im Azure Portal, wählen Sie den von Ihnen erstellten Speicher aus, und löschen Sie ihn.
• Um eine Azure Cache for Redis Instanz zu löschen, öffnen Sie die Ressourcengruppe im Azure-Portal, wählen Sie die erstellte Instanz aus, und löschen Sie sie.

Nächste Schritte

• Netzwerkisolation mit Feature Store
• Azure Machine Learning Feature speichert Beispiel-Repository