Självstudie 5: Utveckla en funktionsuppsättning med en anpassad källa

Med ett Azure Machine Learning hanterat funktionsarkiv kan du identifiera, skapa och operationalisera funktioner. Funktioner fungerar som bindväv i maskininlärningslivscykeln, från prototypfasen, där du experimenterar med olika funktioner. Den livscykeln fortsätter till driftsättningsfasen, där du distribuerar dina modeller och härledningsstegen letar upp funktionsdata. Mer information om funktionslager finns i resursen om funktionslager-koncept.

Del 1 i den här självstudieserien visade hur du skapar en funktionsuppsättningsspecifikation med anpassade transformeringar, aktiverar materialisering och utför en återfyllnad. Del 2 visade hur du experimenterar med funktioner i experimenterings- och träningsflödena. I del 3 förklaras återkommande materialisering för funktionsuppsättningen transactions och hur du kör en pipeline för batchinferens på den registrerade modellen. I del 4 beskrivs hur du kör batchinferens.

I den här handledningen kommer du att

Förutsättningar

• Se till att slutföra de tidigare självstudierna i den här serien. Den här självstudien återanvänder feature store och andra resurser som skapats i de tidigare självstudierna.

Konfigurera

I den här handledningen används Python feature store core SDK:n (azureml-featurestore). Python SDK används för att skapa, läsa, uppdatera och ta bort (CRUD) åtgärder på funktionslager, funktionsuppsättningar och funktionsarkiventiteter.

Du behöver inte uttryckligen installera dessa resurser för den här handledningen, eftersom conda.yml filen omfattar dem i konfigurationsinstruktionerna som visas här.

Konfigurera Azure Machine Learning Sparknotebook

Du kan skapa en ny notebook-fil och köra anvisningarna i den här självstudien steg för steg. Du kan också öppna och köra den befintliga notebook-filen featurestore_sample/notebooks/sdk_only/5.Develop-feature-set-custom-source.ipynb. Håll den här handledningen öppen och använd den för dokumentationslänkar och mer information.

1. På den översta menyn i listrutan Compute väljer du Serverless Spark Compute under Azure Machine Learning Serverless Spark.

2. Konfigurera sessionen:

   1. Välj Konfigurera session i det översta statusfältet
   2. Välj fliken Python paket och välj Ladda upp Conda-fil
   3. Välj Ladda upp Conda-fil
   4. Ladda upp den conda.yml fil som du laddade upp i den första självstudien
   5. Du kan också öka tidsgränsen för sessionen (inaktiv tid) för att undvika frekventa nödvändiga omkörningar

Konfigurera rotkatalogen för exemplen

Den här kodcellen konfigurerar rotkatalogen för exemplen. Det tar cirka 10 minuter att installera alla beroenden och starta Spark-sessionen.

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")

Initiera CRUD-klienten för funktionslagringsarbetsytan

Initiera MLClient för arbetsytan för feature store, för att täcka åtgärderna för skapa, läsa, uppdatera och ta bort (CRUD) på feature store-arbetsytan.

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,
)

Initialisera SDK-klienten för funktionslagringens kärna

Som tidigare nämnts använder den här handledningen Python feature store kärn-SDK (azureml-featurestore). Den här initierade SDK-klienten omfattar åtgärder för att skapa, läsa, uppdatera och ta bort (CRUD) i funktionslager, funktionsuppsättningar och funktionslagerentiteter.

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,
)

Anpassad källdefinition

Du kan definiera din egen källinläsningslogik från alla datalagringar som har en anpassad källdefinition. För att använda den här funktionen, implementera en klass för användardefinierad funktion (UDF) för en källprocessor (CustomSourceTransformer i denna handledning). Den här klassen ska definiera en __init__(self, **kwargs) funktion och en process(self, start_time, end_time, **kwargs) funktion. Ordlistan kwargs anges som en del av funktionsuppsättningens specifikationsdefinition. Den här definitionen skickas sedan till UDF. Parametrarna start_time och end_time beräknas och skickas till UDF-funktionen.

Det här är exempelkod för källprocessorns UDF-klass:

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

Skapa en funktionsuppsättningsspecifikation med en anpassad källa och experimentera med den lokalt

Skapa nu en funktionsuppsättningsspecifikation med en anpassad källdefinition och använd den i utvecklingsmiljön för att experimentera med funktionsuppsättningen. Den självstudiekursanteckningsbok som är kopplad till Serverless Spark Compute fungerar som utvecklingsmiljö.

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

Definiera sedan ett funktionsfönster och visa funktionsvärdena i det här funktionsfönstret.

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
    )
)

Exportera som en funktionsuppsättningsspecifikation

För att registrera funktionsspecifikationen i funktionsarkivet sparar du specifikationen i ett visst format. Granska specifikationen för den genererade funktionsuppsättningen transactions_custom_source . Öppna den här filen från filträdet för att visa specifikationen: featurestore/featuresets/transactions_custom_source/spec/FeaturesetSpec.yaml.

Specifikationen har följande element:

• features: En lista över funktioner och deras datatyper.
• index_columns: Anslutningsnycklarna som krävs för att komma åt värden från funktionsuppsättningen.

Mer information om specifikationen finns i guiderna Förstå topplista över entiteter i hanterad funktionsbutik och CLI (v2) funktionsuppsättningens YAML-schema.

Funktionsuppsättningens specifikationspersistence ger en annan fördel: funktionsuppsättningsspecifikationen kan källkontrolleras.

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

udf_featureset_spec.dump(feature_spec_folder)

Registrera transaktionsfunktionuppsättningen med funktionslagret

Använd den här koden för att registrera ett funktionsuppsättningsobjekt som har lästs in från en anpassad källa i funktionslagret. Du kan sedan återanvända den tillgången och enkelt dela den. Registrering av en funktionstillgångsuppsättning erbjuder förvaltade möjligheter, inklusive versionshantering och materialisering.

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())

Hämta den registrerade funktionsuppsättningen och skriv ut relaterad information.

# 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)

Testa funktionsgenerering från registrerad funktionsuppsättning

to_spark_dataframe() Använd funktionen för funktionsuppsättningen för att testa funktionsgenereringen från den registrerade funktionsuppsättningen och visa funktionerna. print-txn-fset-sample-values

df = transactions_fset_config.to_spark_dataframe()
display(df)

Du bör kunna hämta den registrerade funktionsuppsättningen som en Spark-dataram och sedan visa den. Nu kan du använda de här funktionerna för en punkt-i-tid-koppling med observationsdata och de efterföljande stegen i din maskininlärningspipeline.

Rensa

Om du har skapat en resursgrupp för självstudien kan du ta bort resursgruppen, som tar bort alla resurser som är associerade med den här självstudien. Annars kan du ta bort resurserna individuellt:

• Om du vill ta bort funktionsarkivet öppnar du resursgruppen i Azure portalen, väljer funktionsarkivet och tar bort det.
• Den användartilldelade hanterade identiteten (UAI) som tilldelats funktionsbutiken tas inte bort när vi raderar funktionsbutiken. Om du vill ta bort UAI följer du de här anvisningarna.
• Om du vill ta bort ett offlinearkiv av lagringskontotyp öppnar du resursgruppen i Azure portalen, väljer lagringen som du skapade och tar bort den.
• Om du vill ta bort en Azure Cache for Redis instans öppnar du resursgruppen i Azure-portalen, väljer den instans som du skapade och tar bort den.

Nästa steg

• Nätverksisolering med feature store
• funktionen Azure Machine Learning lagrar exempellagringsplatsen