Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Cuaderno de ejemplo de detección de fraudes con el almacén de características

El código de ejemplo de esta página hace referencia al cuaderno de ejemplo: Detección de fraude con Amazon SageMaker Feature Store. Le recomendamos que ejecute este bloc de notas en Studio Classic, Notebook Instances o JupyterLab porque el código de esta guía es conceptual y no es completamente funcional si se copia.

Utilice lo siguiente para clonar el amazon-sagemaker-examples GitHub repositorio aws/, que contiene el cuaderno de ejemplo.

Ahora que tienes los cuadernos de ejemplo de SageMaker IA, navega hasta el amazon-sagemaker-examples/sagemaker-featurestore directorio y abre el cuaderno de ejemplo de Fraud Detection with Amazon SageMaker Feature Store.

Paso 1: configuración de la sesión del Almacén de características

Para empezar a usar Feature Store, crea una sesión de SageMaker IA, una sesión de Boto3 y una sesión de Feature Store. Además, configure el bucket de S3 que desea utilizar para sus características. Este es su almacenamiento sin conexión. El siguiente código usa el segmento predeterminado de SageMaker IA y le agrega un prefijo personalizado.

import boto3
import sagemaker
from sagemaker.session import Session

sagemaker_session = sagemaker.Session()
region = sagemaker_session.boto_region_name
boto_session = boto3.Session(region_name=region)
role = sagemaker.get_execution_role()
default_bucket = sagemaker_session.default_bucket()
prefix = 'sagemaker-featurestore'
offline_feature_store_bucket = 's3://{}/{}'.format(default_bucket, prefix)

sagemaker_client = boto_session.client(service_name='sagemaker', region_name=region)
featurestore_runtime = boto_session.client(service_name='sagemaker-featurestore-runtime', region_name=region)

feature_store_session = Session(
    boto_session=boto_session,
    sagemaker_client=sagemaker_client,
    sagemaker_featurestore_runtime_client=featurestore_runtime
)

Paso 2: Cargar el conjuntos de datos y datos de partición a grupos de características

Cargue los datos en marcos de datos para cada una de sus características. Utilizará estos marcos de datos después de configurar el grupo de características. En el ejemplo de detección de fraude, puede ver estos pasos en el siguiente código.

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import io

s3_client = boto3.client(service_name='s3', region_name=region)

fraud_detection_bucket_name = 'sagemaker-featurestore-fraud-detection'
identity_file_key = 'sampled_identity.csv'
transaction_file_key = 'sampled_transactions.csv'

identity_data_object = s3_client.get_object(Bucket=fraud_detection_bucket_name, Key=identity_file_key)
transaction_data_object = s3_client.get_object(Bucket=fraud_detection_bucket_name, Key=transaction_file_key)

identity_data = pd.read_csv(io.BytesIO(identity_data_object['Body'].read()))
transaction_data = pd.read_csv(io.BytesIO(transaction_data_object['Body'].read()))

identity_data = identity_data.round(5)
transaction_data = transaction_data.round(5)

identity_data = identity_data.fillna(0)
transaction_data = transaction_data.fillna(0)

# Feature transformations for this dataset are applied before ingestion into FeatureStore.
# One hot encode card4, card6
encoded_card_bank = pd.get_dummies(transaction_data['card4'], prefix = 'card_bank')
encoded_card_type = pd.get_dummies(transaction_data['card6'], prefix = 'card_type')

transformed_transaction_data = pd.concat([transaction_data, encoded_card_type, encoded_card_bank], axis=1)
transformed_transaction_data = transformed_transaction_data.rename(columns={"card_bank_american express": "card_bank_american_express"})

Paso 3: Configurar grupos de características

Al configurar los grupos de características, debe personalizar los nombres de las características con un nombre único y configurar cada grupo de características con la clase FeatureGroup.

from sagemaker.feature_store.feature_group import FeatureGroup
feature_group_name = "some string for a name"
feature_group = FeatureGroup(name=feature_group_name, sagemaker_session=feature_store_session)

Por ejemplo, en el ejemplo de detección de fraude, los dos grupos de características son identity y transaction. En el siguiente código, puede ver cómo se personalizan los nombres con una marca temporal y, a continuación, cada grupo se configura pasando el nombre y la sesión.

import time
from time import gmtime, strftime, sleep
from sagemaker.feature_store.feature_group import FeatureGroup

identity_feature_group_name = 'identity-feature-group-' + strftime('%d-%H-%M-%S', gmtime())
transaction_feature_group_name = 'transaction-feature-group-' + strftime('%d-%H-%M-%S', gmtime())

identity_feature_group = FeatureGroup(name=identity_feature_group_name, sagemaker_session=feature_store_session)
transaction_feature_group = FeatureGroup(name=transaction_feature_group_name, sagemaker_session=feature_store_session)

Paso 4: Configurar el identificador de registros y las características de hora del evento

En este paso, especificará un nombre de identificador de registro y un nombre de característica de hora del evento. Este nombre se asigna a la columna de las características correspondientes de los datos. Por ejemplo, en el ejemplo de detección de fraude, la columna de interés es TransactionID. EventTime se pueden anexar a los datos cuando no hay una marca temporal disponible. En el siguiente código, puede ver cómo se configuran estas variables y, a continuación, EventTime se anexa a los datos de ambas características.

record_identifier_name = "TransactionID"
event_time_feature_name = "EventTime"
current_time_sec = int(round(time.time()))
identity_data[event_time_feature_name] = pd.Series([current_time_sec]*len(identity_data), dtype="float64")
transformed_transaction_data[event_time_feature_name] = pd.Series([current_time_sec]*len(transaction_data), dtype="float64")

Paso 5: Cargar definiciones de características

Ahora puede cargar las definiciones de las características pasando un marco de datos que contenga los datos de las características. En el siguiente código del ejemplo de detección de fraude, la característica de identidad y la característica de transacción se cargan mediante load_feature_definitions, y esta función detecta automáticamente el tipo de datos de cada columna de datos. Los desarrolladores que utilizan un esquema en lugar de la detección automática pueden consultar el ejemplo de código de exportar grupos de características desde Data Wrangler que muestra cómo cargar el esquema, asignarlo y agregarlo como una FeatureDefinition que puede usar para crear el FeatureGroup. Este ejemplo también incluye una AWS SDK for Python (Boto3) implementación, que puede usar en lugar del SDK de SageMaker Python.

identity_feature_group.load_feature_definitions(data_frame=identity_data); # output is suppressed
transaction_feature_group.load_feature_definitions(data_frame=transformed_transaction_data); # output is suppressed

Paso 6: Crear un grupo de características

En este paso, utilizará la función create para crear el grupo de características. El ejemplo de código siguiente muestra todos los parámetros disponibles. El almacenamiento en línea no se crea de forma predeterminada, por lo que debe establecer esto en True si desea habilitarlo. El s3_uri es la ubicación del bucket de S3 del almacenamiento sin conexión.

# create a FeatureGroup
feature_group.create(
    description = "Some info about the feature group",
    feature_group_name = feature_group_name,
    record_identifier_name = record_identifier_name,
    event_time_feature_name = event_time_feature_name,
    feature_definitions = feature_definitions,
    role_arn = role,
    s3_uri = offline_feature_store_bucket,
    enable_online_store = True,
    online_store_kms_key_id = None,
    offline_store_kms_key_id = None,
    disable_glue_table_creation = False,
    data_catalog_config = None,
    tags = ["tag1","tag2"])

El siguiente código del ejemplo de detección de fraude muestra una llamada create mínima para cada uno de los dos grupos de características que se están creando.

identity_feature_group.create(
    s3_uri=offline_feature_store_bucket,
    record_identifier_name=record_identifier_name,
    event_time_feature_name=event_time_feature_name,
    role_arn=role,
    enable_online_store=True
)

transaction_feature_group.create(
    s3_uri=offline_feature_store_bucket,
    record_identifier_name=record_identifier_name,
    event_time_feature_name=event_time_feature_name,
    role_arn=role,
    enable_online_store=True
)

Al crear un grupo de características, se tarda un tiempo en cargar los datos y hay que esperar a que se cree el grupo de características para poder usarlo. Puede comprobar el estado con los métodos siguientes.

status = feature_group.describe().get("FeatureGroupStatus")

Mientras se crea el grupo de características, recibe Creating como respuesta. Cuando este paso haya finalizado correctamente, la respuesta será Created. Otros estados posibles son CreateFailed, Deleting oDeleteFailed.

Paso 7: Trabajar con grupos de características

Ahora que ha configurado el grupo de características, puede realizar cualquiera de las siguientes tareas:

Describir un grupo de características

Puede recuperar información sobre su grupo de características con la función describe.

feature_group.describe()

Enumerar grupos de características

Puede enumerar todos sus grupos de características con la función list_feature_groups.

sagemaker_client.list_feature_groups()

Colocar registros en un grupo de características

Puede utilizar la función ingest para cargar los datos de sus características. Se pasa un marco de datos de características, se establece el número de trabajadores y se elige si se espera a que regrese o no. El siguiente ejemplo muestra el uso de la función ingest.

feature_group.ingest(
    data_frame=feature_data, max_workers=3, wait=True
)

Para cada grupo de características que tenga, ejecute la función ingest en los datos de las características que desee cargar.

Obtener un registro de un grupo de características

Puede utilizar la función get_record para recuperar los datos de una característica específica mediante su identificador de registro. En el siguiente ejemplo, se utiliza un identificador de ejemplo para recuperar el registro.

record_identifier_value = str(2990130)
featurestore_runtime.get_record(FeatureGroupName=transaction_feature_group_name, RecordIdentifierValueAsString=record_identifier_value)

Un ejemplo de respuesta del ejemplo de detección de fraude:

...
'Record': [{'FeatureName': 'TransactionID', 'ValueAsString': '2990130'},
  {'FeatureName': 'isFraud', 'ValueAsString': '0'},
  {'FeatureName': 'TransactionDT', 'ValueAsString': '152647'},
  {'FeatureName': 'TransactionAmt', 'ValueAsString': '75.0'},
  {'FeatureName': 'ProductCD', 'ValueAsString': 'H'},
  {'FeatureName': 'card1', 'ValueAsString': '4577'},
...

Generar comandos Hive DDL

La FeatureStore clase del SDK de SageMaker Python también proporciona la funcionalidad para generar comandos DDL de Hive. El esquema de la tabla se genera en función de las definiciones de las características. Las columnas reciben el nombre de la característica y el tipo de datos se infiere en función del tipo de característica.

print(feature_group.as_hive_ddl())

Ejemplo de salida:

CREATE EXTERNAL TABLE IF NOT EXISTS sagemaker_featurestore.identity-feature-group-27-19-33-00 (
  TransactionID INT
  id_01 FLOAT
  id_02 FLOAT
  id_03 FLOAT
  id_04 FLOAT
 ...

Crear un conjunto de datos de entrenamiento

Feature Store crea automáticamente un catálogo de AWS Glue datos al crear grupos de entidades y, si lo desea, puede desactivarlo. A continuación, se describe cómo crear un único conjunto de datos de entrenamiento con valores de características de grupos de características de identidades y de transacciones creados anteriormente en este tema. Además, a continuación se describe cómo ejecutar una consulta de Amazon Athena para unir los datos almacenados en el almacenamiento sin conexión de los grupos de características de identidades y transacciones.

Para empezar, cree una consulta de Athena utilizando athena_query() tanto para los grupos de características de identidades como de transacciones. El `table_name` es la AWS Glue tabla que Feature Store genera automáticamente.

identity_query = identity_feature_group.athena_query()
transaction_query = transaction_feature_group.athena_query()

identity_table = identity_query.table_name
transaction_table = transaction_query.table_name

Escribir y ejecutar una consulta de Athena

Escribirá la consulta con SQL en estos grupos de características y, a continuación, ejecutará la consulta con el comando .run() y especificará la ubicación del bucket de Amazon S3 donde guardará el conjunto de datos.

# Athena query
query_string = 'SELECT * FROM "'+transaction_table+'" LEFT JOIN "'+identity_table+'" ON "'+transaction_table+'".transactionid = "'+identity_table+'".transactionid'

# run Athena query. The output is loaded to a Pandas dataframe.
dataset = pd.DataFrame()
identity_query.run(query_string=query_string, output_location='s3://'+default_s3_bucket_name+'/query_results/')
identity_query.wait()
dataset = identity_query.as_dataframe()

A partir de aquí, puede entrenar un modelo con este conjunto de datos y, a continuación, realizar una inferencia.

Eliminar un grupo de características

Puede eliminar un grupo de características con la función delete.

feature_group.delete()

El siguiente ejemplo de código proviene del ejemplo de detección de fraude.

identity_feature_group.delete()
transaction_feature_group.delete()

Para obtener más información, consulte Delete a feature group API.