SDK do Processador de atributos do Feature Store

Declare uma definição de Processador de atributos do Feature Store decorando suas funções de transformação com o decorador @feature_processor. O SageMaker AI SDK for Python (Boto3) carrega automaticamente os dados das fontes de dados de entrada configuradas, aplica a função de transformação decorada e, em seguida, ingere os dados transformados em um grupo de recursos de destino. As funções de transformação decoradas devem estar de acordo com a assinatura esperada do decorador @feature_processor. Para obter mais informações sobre o @feature_processor decorador, consulte @feature_processor Decorator na Amazon SageMaker Feature Store Read the Docs.

Com o @feature_processor decorador, sua função de transformação é executada em um ambiente de execução do Spark, onde os argumentos de entrada fornecidos à sua função e seu valor de retorno são Spark. DataFrames O número de parâmetros de entrada em sua função de transformação deve corresponder ao número de entradas configuradas no decorador @feature_processor.

Para obter mais informações sobre o decorador @feature_processor, consulte o SDK do Processador de atributos do Feature Store para Python (Boto3).

O código a seguir é um exemplo básico de como usar o decorador @feature_processor. Para obter exemplos de casos de uso mais específicos, consulte Exemplo de código de Processamento de atributos para casos de uso comuns.

O Feature Processor SDK pode ser instalado a partir do SDK do SageMaker Python e seus extras usando o comando a seguir.

pip install sagemaker[feature-processor]

Nos exemplos a seguir, us-east-1 é a região do atributo, 111122223333 é o ID da conta do proprietário do atributo e your-feature-group-name é o nome do grupo de atributos.

Veja a seguir uma definição básica de Processador de atributos, em que o decorador @feature_processor configura uma entrada CSV do Amazon S3 para ser carregada e fornecida à sua função de transformação (por exemplo, transform) e a prepara para ingestão em um grupo de atributos. A última linha o executa.

from sagemaker.feature_store.feature_processor import CSVDataSource, feature_processor

CSV_DATA_SOURCE = CSVDataSource('s3://your-bucket/prefix-to-csv/')
OUTPUT_FG = 'arn:aws:sagemaker:us-east-1:111122223333:feature-group/your-feature-group-name'

@feature_processor(inputs=[CSV_DATA_SOURCE], output=OUTPUT_FG)
def transform(csv_input_df):
   return csv_input_df
   
transform()

O parâmetro @feature_processor inclui:

Os parâmetros opcionais da função encapsulada (fornecidos como argumento se fornecidos na assinatura da função) incluem:

O código a seguir é um exemplo do uso dos parâmetros params e spark.

from sagemaker.feature_store.feature_processor import CSVDataSource, feature_processor

CSV_DATA_SOURCE = CSVDataSource('s3://your-bucket/prefix-to-csv/')
OUTPUT_FG = 'arn:aws:sagemaker:us-east-1:111122223333:feature-group/your-feature-group-name' 

@feature_processor(inputs=[CSV_DATA_SOURCE], output=OUTPUT_FG)
def transform(csv_input_df, params, spark):
   
   scheduled_time = params['system']['scheduled_time']
   csv_input_df.createOrReplaceTempView('csv_input_df')
   return spark.sql(f'''
        SELECT *
        FROM csv_input_df
        WHERE date_add(event_time, 1) >= {scheduled_time}
   ''')
   
transform()

O parâmetro do sistema scheduled_time (fornecido no argumento params da sua função) é um valor importante para compatibilidade com a repetição de cada execução. O valor pode ajudar a identificar de forma exclusiva a execução do Processador de atributos e pode ser usado como um ponto de referência para entradas baseadas em intervalos de datas (por exemplo, carregando apenas os dados das últimas 24 horas) para garantir o intervalo de entrada independente do tempo real de execução do código. Se o Processador de atributos for executado com base em uma programação (consulte Execuções programadas e baseadas em eventos para pipelines do Processador de atributos), seu valor será fixado no horário programado para execução. O argumento pode ser substituído durante a execução síncrona usando a API de execução do SDK para dar compatibilidade com casos de uso como preenchimento de dados ou reexecução de uma execução anterior perdida. Seu valor é a hora atual se o Processador de atributos for executado de outra forma.

Para obter informações sobre a criação do código Spark, consulte o Guia do programa SQL do Spark.

Para obter mais exemplos de código para casos de uso comuns, consulte o Exemplo de código de Processamento de atributos para casos de uso comuns.

Observe que as funções de transformação decoradas com @feature_processor não retornam um valor. Para testar programaticamente sua função, você pode remover ou corrigir o decorador @feature_processor de forma que ele atue como uma passagem para a função agrupada. Para obter mais detalhes sobre o @feature_processor decorador, consulte Amazon SageMaker Feature Store Python SDK.