Compartir a través de

Tutorial 5: Desarrollo de un conjunto de características con un origen personalizado

  • Artículo

El almacén de características gestionado de Azure Machine Learning le permite detectar, crear y operacionalizar características. Las características sirven como tejido conectivo en el ciclo de vida del aprendizaje automático, empezando por la fase de creación de prototipos, donde experimenta con diversas características. Ese ciclo de vida continúa con la fase de operacionalización, donde se implementan los modelos y los pasos de inferencia buscan los datos de características. Para obtener más información sobre los almacenes de características, vea el documento de conceptos del almacén de características.

La parte 1 de esta serie de tutoriales mostró cómo crear una especificación de conjunto de características con transformaciones personalizadas, cómo permitir la materialización y ejecutar una reposición. La parte 2 mostró cómo experimentar con características en los flujos de experimentación y de entrenamiento. En la parte 3 se explicó la materialización recurrente del conjunto de características transactions y se mostró cómo ejecutar una canalización de inferencia por lotes en el modelo registrado. En la parte 4 se describe cómo ejecutar la inferencia por lotes.

En este tutorial, hará lo siguiente

  • Define la lógica para cargar datos desde un origen de datos personalizado.
  • Configura y registra un conjunto de características para consumir desde este origen de datos personalizado.
  • Pruebe el conjunto de características registrado.

Requisitos previos

Nota:

Este tutorial utiliza un cuaderno de Azure Machine Learning con Serverless Spark Compute.

  • Asegúrese de realizar los tutoriales anteriores de esta serie. En este tutorial se reutiliza el almacén de características y otros recursos creados en esos tutoriales anteriores.

Configurar

En este tutorial se usa el SDK principal del almacén de características de Python (azureml-featurestore). El SDK de Python se usa para las operaciones de creación, lectura, actualización y eliminación (CRUD), en almacenes de características, conjuntos de características y entidades de almacén de características.

No necesita instalar explícitamente estos recursos para este tutorial, porque en las instrucciones de instalación que se muestran aquí, el archivo conda.yml los cubre.

Configurar el cuaderno de Spark de Azure Machine Learning

Puede crear un cuaderno y ejecutar las instrucciones de este tutorial paso a paso. También puede abrir y ejecutar el cuaderno existente featurestore_sample/notebooks/sdk_only/5.Enable-online-store-run-inference.ipynb. Mantenga este tutorial abierto y consúltelo para obtener vínculos de documentación y explicaciones más detalladas.

  1. En el menú superior, en la lista desplegable Proceso, seleccione Proceso de Spark sin servidor en Spark sin servidor de Azure Machine Learning.

  2. Configuración de la sesión:

    1. Seleccione Configurar sesión en el panel de navegación superior.
    2. Seleccione la pestaña Paquetes de Python, s
    3. Seleccione Cargar archivo de Conda.
    4. Cargue el archivo conda.yml que cargó en el primer tutorial.
    5. De manera opcional, aumente el tiempo de espera de la sesión (tiempo de inactividad) para evitar repeticiones frecuentes de ejecución de requisitos previos.

Configuración del directorio raíz para los ejemplos

Esta celda de código configura el directorio raíz para las muestras. Necesita unos 10 minutos para instalar todas las dependencias e iniciar la sesión de Spark.

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

Inicializar el cliente CRUD del área de trabajo del almacén de características

Inicializa el MLClient para el área de trabajo del almacén de características para cubrir las operaciones de creación, lectura, actualización y eliminación (CRUD) en el área de trabajo del almacén de características.

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

Inicializar el cliente del SDK principal del almacén de características

Como se mencionó anteriormente, en este tutorial se usa el SDK principal del almacén de características de Python (azureml-featurestore). Este cliente de SDK inicializado se usa para las operaciones de creación, lectura, actualización y eliminación (CRUD), en almacenes de características, conjuntos de características y entidades de almacén de características.

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

Definición de origen personalizado

Puedes definir tu propia lógica de carga de origen desde cualquier almacenamiento de datos que tenga una definición de origen personalizada. Implementa una clase definida por el usuario (UDF) del procesador de origen (CustomSourceTransformer en este tutorial) para usar esta característica. Esta clase debe definir una función __init__(self, **kwargs) y una función process(self, start_time, end_time, **kwargs) . El diccionario kwargs se proporciona como parte de la definición de especificación del conjunto de características. Luego esta definición se pasa a la UDF. Los parámetros start_time y end_time se calculan y pasan a la función UDF.

Este es el código de ejemplo de la clase UDF del procesador de origen:

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

Crea una especificación de conjunto de características con un origen personalizado y experimenta con ella localmente.

Ahora, crea una especificación de conjunto de características con una definición de origen personalizada y úsala en el entorno de desarrollo para experimentar con el conjunto de características. El cuaderno del tutorial asociado a Proceso spark sin servidor actúa como entorno de desarrollo.

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

A continuación, define una ventana de características y muestra los valores de características en esta ventana de características.

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

Exporta como una especificación de conjunto de características.

Para registrar la especificación de conjunto de características con el almacén de características, primero guarda la especificación de conjunto de características en un formato específico. Revise la especificación del conjunto de características generado transactions_custom_source. Abre este archivo desde el árbol de archivos para ver la especificación: featurestore/featuresets/transactions_custom_source/spec/FeaturesetSpec.yaml.

La especificación contiene estos elementos:

  • features: Lista de características y sus tipos de datos.
  • index_columns: Claves de combinación necesarias para acceder a los valores del conjunto de características

Para obtener más información sobre la especificación, vea Descripción de las entidades de nivel superior en el almacén de características gestionado y el esquema YAML del conjunto de características de la CLI (v2).

Conservar la especificación del conjunto de características ofrece otra ventaja: la especificación del conjunto de características puede controlarse en origen.

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

udf_featureset_spec.dump(feature_spec_folder)

Registro del conjunto de características de transacciones con el almacén de características

Usa este código para registrar un recurso de conjunto de características cargado desde el origen personalizado con el almacén de características. Después, puede reutilizar ese recurso y compartirlo fácilmente. El registro de un recurso de conjunto de características ofrece funcionalidades administradas, incluido el control de versiones y la materialización.

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

Obtén el conjunto de características registrado e imprime la información relacionada.

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

Prueba de la generación de características del conjunto de características registrado

Usa la función to_spark_dataframe() del conjunto de características para probar la generación de características del conjunto de características registrado y mostrar las características. print-txn-fset-sample-values

df = transactions_fset_config.to_spark_dataframe()
display(df)

Deberías poder capturar correctamente el conjunto de características registrado como un dataframe de Spark y, a continuación, mostrarlo. Ahora puedes usar estas características para una combinación a un momento dado con datos de observación y los pasos posteriores de la canalización de aprendizaje automático.

Limpieza

Si has creado un grupo de recursos para el tutorial, puedes eliminar ese grupo de recursos para eliminar así todos los recursos asociados a este tutorial. De lo contrario, puede eliminar los recursos individualmente:

  • Para eliminar el almacén de características, ve al grupo de recursos en el Azure Portal, selecciona el almacén de características y elimínalo.
  • La identidad administrada asignada por el usuario (UAI) asignada al área de trabajo del almacén de características no se elimina al eliminar el almacén de características. Para eliminar la UAI, sigue estas instrucciones.
  • Para eliminar un almacenamiento sin conexión tipo cuenta de almacenamiento, abre al grupo de recursos en el Azure Portal, selecciona el almacenamiento que has creado y elimínalo.
  • Para eliminar una instancia de Azure Cache for Redis, abre el grupo de recursos en Azure Portal, selecciona la instancia que has creado y elimínala.

Pasos siguientes