Feature Store 機能プロセッサ SDK

変換関数を @feature_processor デコレータでデコレートして、Feature Store Feature Processor 定義を宣言します。for SageMaker SDKPython (Boto3) は、設定された入力データソースからデータを自動的にロードし、装飾された変換関数を適用し、変換されたデータをターゲット特徴量グループに取り込みます。デコレーションされた変換関数は @feature_processor デコレータの期待するシグネチャに従わなければなりません。@feature_processor デコレータの詳細については、「Amazon SageMaker Feature Store ドキュメントを読む」の「@feature_processor Decorator」を参照してください。

@feature_processor デコレータを使用すると、変換関数は Spark ランタイム環境で実行され、関数に指定された入力引数とその戻り値は Spark です DataFrames。変換関数の入力パラメータの数は、@feature_processor デコレータに設定された入力の数と一致する必要があります。

@feature_processor デコレータの詳細については、Feature Processor Feature Store SDK for Python (Boto3) を参照してください。

次のコードは @feature_processor デコレータの使用方法の基本的な例です。より具体的な使用例については、「一般的なユースケース向けの特徴量処理コードの例」を参照してください。

Feature Processor は、次のコマンドを使用して SageMaker Python SDKとその追加からインストールSDKできます。


pip install sagemaker[feature-processor]

次の例では、us-east-1 はリソースのリージョン、111122223333 はリソース所有者のアカウント ID、your-feature-group-name は特徴量グループ名です。

以下は基本的な特徴プロセッサ定義です。デ@feature_processorコレータは Amazon S3 からCSVの入力をロードして変換関数 ( などtransform) に提供し、特徴グループへの取り込みに備えます。最後の行ではこの関数が実行されます。


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

@feature_processor パラメータには以下が含まれます。

inputs (List[str]): Feature Store Feature Processor.で使用されるデータソースのリスト。データソースが特徴量グループであるか、Amazon S3 に保存されている場合は、Feature Store が提供する特徴量プロセッサのデータソース定義を使用できる場合があります。Feature Store が提供するデータソース定義の完全なリストについては、Amazon SageMaker Feature Store の「Feature Processor Data Source」を参照してください。
output (str): デコレーションARNされた関数の出力を取り込む特徴グループの。
target_stores (オプション [List[str]]): 出力に取り込むストアのリスト (例えば、OnlineStore または OfflineStore)。指定しない場合、データは出力特徴量グループで有効になっているすべてのストアに取り込まれます。
parameters (Dict[str, Any]): 変換関数に提供されるディクショナリ。
enable_ingestion (bool): 変換関数の出力を出力特徴量グループに取り込むかどうかを示すフラグ。このフラグは開発段階で役立ちます。指定しない場合、取り込みは有効になります。

オプションのラップ関数パラメータ (関数シグネチャで指定されている場合は引数として提供) には以下が含まれます。

params (Dict[str, Any]): @feature_processor パラメータで定義されているディクショナリ。また、scheduled_time パラメータなど、キー system で参照できるシステム設定パラメータも含まれています。
spark (SparkSession): Spark アプリケーション用に初期化された SparkSession インスタンスへの参照。

以下のコードは、params パラメータと 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()

scheduled_time システムパラメータ (関数の params 引数で指定) は、実行それぞれの再試行をサポートする重要な値です。この値は、Feature Processor の実行を一意に識別するのに役立ち、コードの実際の実行時間とは無関係に入力範囲を確保するために、日付範囲ベースの入力 (過去 24 時間分のデータのみをロードするなど) の基準点として使用できます。Feature Processor がスケジュールに従って実行されている場合 (「Feature Processor パイプラインのスケジュールされた実行とイベントベースの実行」を参照)、その値は実行予定時刻に固定されます。引数は、データのバックフィルや過去の実行ミスの再実行などのユースケースAPIをサポートするために、 SDKの実行を使用して同期実行中に上書きできます。Feature Processor が他の方法で実行されている場合、その値は現在の時刻です。

Spark コードの作成については、「Spark SQL プログラミングガイド」を参照してください。

一般的なユースケースのその他のコードサンプルについては、「一般的なユースケース向けの特徴量処理コードの例」を参照してください。

@feature_processor でデコレーションされた変換関数は値を返さないことに注意してください。関数をプログラムでテストするには、@feature_processor デコレータを削除するかモンキーパッチを適用して、ラップ関数へのパススルーとして機能させることができます。@feature_processor デコレータの詳細については、「Amazon SageMaker Feature Store PythonSDK」を参照してください。

ブラウザで JavaScript が無効になっているか、使用できません。

AWS ドキュメントを使用するには、JavaScript を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

機能の処理

Feature Store Feature Processor のリモートでの実行