Freigeben über

Tutorial 5: Entwickeln eines Featuresatzes mit einer benutzerdefinierten Quelle

  • Artikel

Mit dem verwalteten Feature Store von Azure Machine Learning können Sie Features entdecken, 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 Operationalisierungsphase fortgesetzt, in der Sie Modelle bereitstellen und Rückschlussschritte nach Featuredaten suchen. Weitere Informationen zu Feature Stores finden Sie im Artikel zu den Feature Store-Konzepten.

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 der Batchrückschluss ausgeführt wird.

In diesem Tutorial führen Sie folgende Schritte aus:

  • Definieren Sie die Logik zum Laden von Daten aus einer benutzerdefinierten Datenquelle.
  • Konfigurieren und registrieren Sie einen Featuresatz, der von dieser benutzerdefinierten Datenquelle verwendet werden soll.
  • Testen Sie den registrierten Featuresatz.

Voraussetzungen

Hinweis

In diesem Tutorial wird ein Azure Machine Learning-Notebook mit serverlosem Spark Compute verwendet.

  • 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 Tutorial wird das Python Feature Store Core SDK (azureml-featurestore) verwendet. Das SDK für Python wird für CRUD-Vorgänge (Create, Read, Update, Delete; Erstellen, Lesen, Aktualisieren, Löschen) in Feature Stores, Featuresätzen und Feature Store-Entitä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 Spark-Notebooks für Azure Machine Learning

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. Lassen Sie dieses Dokument geöffnet, und sehen Sie dort nach, um Links zur Dokumentation und weitere Erläuterungen zu erhalten.

  1. Wählen Sie im oberen Menü unter Serverlose Spark-Computeversion von Azure Machine Learning die Option Serverlose Spark-Computeversion in der Dropdownliste Compute aus.

  2. Konfigurieren der Sitzung:

    1. Wählen Sie in der oberen Statusleiste Sitzung konfigurieren aus.
    2. Wählen Sie die Registerkarte Python-Pakete.
    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 des CRUD-Clients 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 wird in diesem Tutorial das Python Feature Store Core SDK (azureml-featurestore) verwendet. Dieser initialisierte SDK-Client deckt CRUD-Vorgänge 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 der Dateistruktur, 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 der Funktionsgruppe erforderlich sind.

Weitere Informationen zur Spezifikation finden Sie unter Grundlegendes zu Entitäten auf oberster Ebene im verwalteten Feature Store und YAML-Schema für die Featuresatzspezifikation 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 Featuresatz ab, und geben Sie verwandte 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 Feature Store zu löschen, öffnen Sie die Ressourcengruppe im Azure-Portal, wählen den Feature Store aus und löschen ihn.
  • 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 den von Ihnen erstellten Speicher aus und löschen ihn.
  • Um eine Azure Cache for Redis-Instanz zu löschen, öffnen Sie die Ressourcengruppe im Azure-Portal, wählen die von Ihnen erstellte Instanz aus und löschen sie.

Nächste Schritte