Feature Store Feature Processor SDK - Amazon SageMaker

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Feature Store Feature Processor SDK

Dichiara una definizione del processore di funzionalità Feature Store decorando le funzioni di trasformazione con il decoratore @feature_processor. SageMaker SDKfor Python (Boto3) carica automaticamente i dati dalle fonti di dati di input configurate, applica la funzione di trasformazione decorata e quindi inserisce i dati trasformati in un gruppo di funzionalità di destinazione. Le funzioni di trasformazione decorate devono essere conformi alla firma prevista del decoratore @feature_processor. Per ulteriori informazioni sul @feature_processor decoratore, consulta @feature_processor Decorator nell'Amazon SageMaker Feature Store Leggi i documenti.

Con il @feature_processor decoratore, la funzione di trasformazione viene eseguita in un ambiente di runtime Spark in cui gli argomenti di input forniti alla funzione e il suo valore restituito sono Spark. DataFrames Il numero di parametri di input nella funzione di trasformazione deve corrispondere al numero di input configurati nel decoratore @feature_processor.

Per ulteriori informazioni sul @feature_processor decoratore, consulta il Feature Processor Feature Store SDK for Python (Boto3).

Il codice seguente è un esempio di base su come usare il decoratore @feature_processor. Per esempi di utilizzo più specifici, consulta Esempio di codice di elaborazione della funzionalità per casi d'uso comuni.

Il Feature Processor SDK può essere installato da SageMaker Python SDK e dai suoi extra usando il seguente comando. 

pip install sagemaker[feature-processor]

Negli esempi seguenti, us-east-1 è la Regione della risorsa, 111122223333 è l'ID dell'account proprietario della risorsa e your-feature-group-name è il nome del gruppo di funzionalità.

Di seguito è riportata una definizione di base del feature processor, in cui il @feature_processor decoratore configura un CSV input da Amazon S3 da caricare e fornire alla funzione di trasformazione (ad esempiotransform) e lo prepara per l'inserimento in un gruppo di funzionalità. Viene eseguito dall'ultima riga.

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

I parametri @feature_processor includono:

  • inputs (List[str]): un elenco di fonti di dati utilizzate nel processore di funzionalità Feature Store. Se le tue fonti di dati sono gruppi di funzionalità o sono archiviate in Amazon S3, potresti essere in grado di utilizzare le definizioni delle fonti di dati fornite da Feature Store per il processore di funzionalità. Per un elenco completo delle definizioni delle fonti di dati fornite dal Feature Store, consulta la fonte dei dati del Feature Processor in Amazon SageMaker Feature Store Leggi i documenti.

  • output(str): Il gruppo ARN di funzionalità che acquisisce l'output della funzione decorata.

  • target_stores (Optional[List[str]]): un elenco di archivi (ad esempio, OnlineStore o OfflineStore) da inserire nell'output. Se non specificato, i dati vengono inseriti in tutti gli archivi abilitati del gruppo di funzionalità di output.

  • parameters (Dict[str, Any]): un dizionario da fornire alla funzione di trasformazione.

  • enable_ingestion (bool): flag che indica se gli output della funzione di trasformazione vengono inseriti nel gruppo di funzionalità di output. Questo flag è utile durante la fase di sviluppo. Se non specificato, l'inserimento è abilitato.

I parametri opzionali della funzione con wrap (forniti come argomento se presenti nella firma della funzione) includono:

  • params (Dict[str, Any]): il dizionario definito nei parametri @feature_processor. Contiene anche parametri configurati dal sistema a cui è possibile fare riferimento con la chiave system, come il parametro scheduled_time.

  • spark(SparkSession): Un riferimento all' SparkSession istanza inizializzata per l'applicazione Spark.

Il seguente codice è un esempio di utilizzo dei parametri 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()

Il parametro di sistema scheduled_time (fornito nell'argomento params della funzione) è un valore importante per supportare ogni nuovo tentativo di esecuzione. Il valore può aiutare a identificare in modo univoco l'esecuzione del processore di funzionalità e può essere utilizzato come punto di riferimento per gli input basati su intervalli di date (ad esempio, caricando solo i dati delle ultime 24 ore) per garantire l'intervallo di input indipendente dal tempo di esecuzione effettivo del codice. Se il processore di funzionalità viene eseguito secondo un programma (consulta Esecuzioni pianificate e basate su eventi per le pipeline del Processore di funzionalità), il relativo valore viene fissato all'ora in cui è pianificata l'esecuzione. L'argomento può essere sovrascritto durante l'esecuzione sincrona utilizzando il comando execute API per supportare casi d'uso come il SDK backfill dei dati o la riesecuzione di un'esecuzione precedente mancata. Il valore corrisponde all'ora corrente se il processore di funzionalità viene eseguito in altro modo.

Per informazioni sulla creazione del codice Spark, consulta la Spark Programming Guide. SQL

Per ulteriori esempi di codice per casi d'uso comuni, consulta Esempio di codice di elaborazione della funzionalità per casi d'uso comuni.

Si noti che le funzioni di trasformazione decorate con @feature_processor non restituiscono alcun valore. Per testare programmaticamente la tua funzione, puoi rimuovere o applicare patch al decoratore @feature_processor in modo che funga da collegamento diretto alla funzione con wrap. Per maggiori dettagli sul @feature_processor decoratore, consulta Amazon SageMaker Feature Store Python SDK.