Feature Store Feature Processor SDK

変換関数を @feature_processor デコレータでデコレートして、Feature Store Feature Processor 定義を宣言します。SageMaker AI SDK for Python (Boto3) は、設定された入力データソースからデータを自動的にロードし、デコレーションされた変換関数を適用してから、変換されたデータをターゲット特徴量グループに取り込みます。デコレーションされた変換関数は @feature_processor デコレータの期待するシグネチャに従わなければなりません。@feature_processor デコレータの詳細については、「 Read the Docs」の「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 SDK は、次のコマンドを使用して SageMaker Python 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 が提供するデータソース定義の全リストについては、「Read the Docs」の「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 パイプラインのスケジュール済みの実行とイベントベースの実行」を参照)、その値は実行予定時刻に固定されます。SDK の実行 API を使用して同期実行中に引数をオーバーライドできます。これにより、データのバックフィルや、失敗した過去の実行の再実行などのユースケースに対応できます。Feature Processor が他の方法で実行されている場合、その値は現在の時刻です。

Spark コードの作成については、「 Spark SQL Programming Guide」を参照してください。

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

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