Criar um conjunto de dados a partir de seus grupos de atributos

Depois que um grupo de atributos do Feature Store for criado em um armazenamento offline, você poderá optar por usar os seguintes métodos para obter seus dados:

Depois de criar grupos de atributos para seu armazenamento offline e preenchê-los com dados, você pode criar um conjunto de dados executando consultas ou usando o SDK para juntar dados armazenados no armazenamento offline de diferentes grupos de atributos. Você também pode juntar os grupos de atributos a um único dataframe de pandas. Você pode usar o Amazon Athena para escrever e executar consultas SQL.

Usar o Amazon SageMaker Python SDK para obter seus dados dos seus grupos de atributos

Você pode usar as APIs do Feature Store para criar um conjunto de dados dos seus grupos de atributos. Cientistas de dados criam conjuntos de dados de ML para treinamento recuperando dados de atributos de ML de um ou mais grupos de atributos no armazenamento offline. Use a função create_dataset() para criar o conjunto de dados. Você pode usar o SDK para fazer o seguinte:

Por padrão, o Feature Store não inclui registros que você excluiu do conjunto de dados. Ele também não inclui registros duplicados. Um registro duplicado tem o ID do registro e o valor do timestamp na coluna de hora do evento.

Antes de usar o SDK para criar um conjunto de dados, você deve iniciar uma sessão do SageMaker. Use o código a seguir para iniciar a sessão.

import boto3
from sagemaker.session import Session
from sagemaker.feature_store.feature_store import FeatureStore

region = boto3.Session().region_name
boto_session = boto3.Session(region_name=region)

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

feature_store = FeatureStore(feature_store_session) 

O código a seguir mostra um exemplo de criação de um conjunto de dados a partir de vários grupos de atributos. O trecho de código a seguir usa os exemplos de grupos de atributos "base_fg_name","first_fg_name" e "second_fg_name", que podem não existir ou ter o mesmo esquema no seu Feature Store. É recomendável substituir esses grupos de atributos por grupos de atributos que existem no seu Feature Store. Para obter informações sobre como criar um grupo atributos, consulte Etapa 3: criar grupos de atributos.

from sagemaker.feature_store.feature_group import FeatureGroup

s3_bucket_name = "offline-store-sdk-test" 

base_fg_name = "base_fg_name"
base_fg = FeatureGroup(name=base_fg_name, sagemaker_session=feature_store_session)

first_fg_name = "first_fg_name"
first_fg = FeatureGroup(name=first_fg_name, sagemaker_session=feature_store_session)

second_fg_name = "second_fg_name"
second_fg = FeatureGroup(name=second_fg_name, sagemaker_session=feature_store_session)

feature_store = FeatureStore(feature_store_session)
builder = feature_store.create_dataset(
    base=base_fg,
    output_path=f"s3://{amzn-s3-demo-bucket1}",
).with_feature_group(first_fg
).with_feature_group(second_fg, "base_id", ["base_feature_1"])         
        

O código a seguir mostra um exemplo de criação de um conjunto de dados a partir de vários grupos de atributos e dataframe do pandas.

base_data = [[1, 187512346.0, 123, 128],
             [2, 187512347.0, 168, 258],
             [3, 187512348.0, 125, 184],
             [1, 187512349.0, 195, 206]]
base_data_df = pd.DataFrame(
    base_data, 
    columns=["base_id", "base_time", "base_feature_1", "base_feature_2"]
)

builder = feature_store.create_dataset(
    base=base_data_df, 
    event_time_identifier_feature_name='base_time', 
    record_identifier_feature_name='base_id',
    output_path=f"s3://{s3_bucket_name}"
).with_feature_group(first_fg
).with_feature_group(second_fg, "base_id", ["base_feature_1"])            
        

As APIs do Feature Store fornecem métodos auxiliares para a função create_dataset. Você pode usá-las para fazer o seguinte:

O grupo de atributos de base é um conceito importante para junções. O grupo de atributos de base é o grupo de atributos que tem outros grupos de atributos ou o dataframe pandas juntados a ele. Para cada conjunto de dados

Você pode adicionar os seguintes métodos opcionais à função create_dataset para configurar como você criar o conjunto de dados:

Depois de configurar o conjunto de dados, especifique a saída usando um dos seguintes métodos:

Você pode recuperar dados que chegam após um período específico. O código a seguir recupera dados após um timestamp.

fg1 = FeatureGroup("example-feature-group-1")
feature_store.create_dataset(
    base=fg1, 
    output_path="s3://example-S3-path"
).with_number_of_records_from_query_results(5).to_csv_file()  

Também é possível recuperar dados de um período de tempo específico. Você pode usar o código a seguir para obter dados de um intervalo de tempo específico:

fg1 = FeatureGroup("fg1")
feature_store.create_dataset(
    base=fg1, 
    output_path="example-S3-path"
).with_event_time_range(
    datetime(2021, 11, 28, 23, 55, 59, 342380), 
    datetime(2020, 11, 28, 23, 55, 59, 342380)
).to_csv_file() #example time range specified in datetime functions

Talvez você queira juntar vários grupos de atributos a um dataframe do pandas em que os valores de hora do evento do grupo de atributos ocorram, no mais tardar, na hora do evento do dataframe. Use o código a seguir como modelo para ajudá-lo a realizar a junção.

fg1 = FeatureGroup("fg1")
fg2 = FeatureGroup("fg2")
events = [['2020-02-01T08:30:00Z', 6, 1],
          ['2020-02-02T10:15:30Z', 5, 2],
          ['2020-02-03T13:20:59Z', 1, 3],
          ['2021-01-01T00:00:00Z', 1, 4]]
df = pd.DataFrame(events, columns=['event_time', 'customer-id', 'title-id']) 
feature_store.create_dataset(
    base=df, 
    event_time_identifier_feature_name='event_time', 
    record_identifier_feature_name='customer_id',
    output_path="s3://example-S3-path"
).with_feature_group(fg1, "customer-id"
).with_feature_group(fg2, "title-id"
).point_in_time_accurate_join(
).to_csv_file()

Também é possível recuperar dados que chegam após um período de tempo específico. O código a seguir recupera dados após o horário especificado pelo timestamp no método as_of.

fg1 = FeatureGroup("fg1")
feature_store.create_dataset(
    base=fg1, 
    output_path="s3://example-s3-file-path"
).as_of(datetime(2021, 11, 28, 23, 55, 59, 342380)
).to_csv_file() # example datetime values

Exemplos de consultas do Amazon Athena

Você pode gravar consultas no Amazon Athena para criar um conjunto de dados a partir dos seus grupos de atributos. Também é possível gravar consultas que criam um conjunto de dados a partir de grupos de atributos e de um único dataframe do pandas.

Exploração interativa

Essa consulta seleciona os primeiros 1000 registros. 

SELECT *
FROM <FeatureGroup.DataCatalogConfig.DatabaseName>.<FeatureGroup.DataCatalogConfig.TableName>
LIMIT 1000

Snapshot mais recente sem duplicatas

Essa consulta seleciona os registros não duplicados mais recentes.

SELECT *
FROM
    (SELECT *,
         row_number()
        OVER (PARTITION BY <RecordIdentiferFeatureName>
    ORDER BY  <EventTimeFeatureName> desc, Api_Invocation_Time DESC, write_time DESC) AS row_num
    FROM <FeatureGroup.DataCatalogConfig.DatabaseName>.<FeatureGroup.DataCatalogConfig.TableName>)
WHERE row_num = 1;

Snapshot mais recente sem duplicatas e registros excluídos no armazenamento offline

Essa consulta filtra todos os registros excluídos e seleciona registros não duplicados do armazenamento offline. 

SELECT *
FROM
    (SELECT *,
         row_number()
        OVER (PARTITION BY <RecordIdentiferFeatureName>
    ORDER BY  <EventTimeFeatureName> desc, Api_Invocation_Time DESC, write_time DESC) AS row_num
    FROM <FeatureGroup.DataCatalogConfig.DatabaseName>.<FeatureGroup.DataCatalogConfig.TableName>)
WHERE row_num = 1 and 
NOT is_deleted;

Viagem no tempo sem duplicatas e registros excluídos no armazenamento offline

Essa consulta filtra todos os registros excluídos e seleciona registros não duplicados de um ponto no tempo particular.

SELECT *
FROM
    (SELECT *,
         row_number()
        OVER (PARTITION BY <RecordIdentiferFeatureName>
    ORDER BY  <EventTimeFeatureName> desc, Api_Invocation_Time DESC, write_time DESC) AS row_num
    FROM <FeatureGroup.DataCatalogConfig.DatabaseName>.<FeatureGroup.DataCatalogConfig.TableName>
    where <EventTimeFeatureName> <= timestamp '<timestamp>')
    -- replace timestamp '<timestamp>' with just <timestamp>  if EventTimeFeature is of type fractional
WHERE row_num = 1 and
NOT is_deleted