ストリーミング インポート

ストリーミング インポートにより、特徴値をリアルタイムで更新できます。この方法は、オンライン サービングで最新の利用可能なデータを優先する場合に役立ちます。たとえば、ストリーミング イベントデータをインポートでき、数秒以内に Vertex AI Feature Store（レガシー）によってインポートしたデータをオンライン サービングのシナリオで使用できるようにします。

データをバックフィルする必要がある場合、または特徴値をバッチで計算する場合は、バッチ インポートを使用します。ストリーミング インポート リクエストと比較すると、バッチ インポート リクエストはより大きなペイロードを処理できますが、完了までに時間を要します。

インポートできる最も古い特徴値のタイムスタンプについては、割り当てと上限の Vertex AI Feature Store（従来版）をご覧ください。タイムスタンプが将来の日時を示している特徴値はインポートできません。

使用例

オンライン小売店組織は、ユーザーの現在のアクティビティを使用して、パーソナライズされたショッピング エクスペリエンスを提供できます。ユーザーがウェブサイト内を移動する際のアクティビティを featurestore にキャプチャし、その後直ちにキャプチャしたすべての情報をオンライン予測に提供できます。このリアルタイムのインポートとサービングにより、ショッピング セッション中に有用で関連性の高いレコメンデーションをお客様に示すことができます。

オンライン ストレージ ノードの使用量

オンライン ストアに特徴値を書き込むと、featurestore の CPU リソース（オンライン ストレージ ノード）が使用されます。CPU 使用率をモニタリングして、需要が供給を超えた状態（サービスエラーが発生する原因となる可能性があります）になっていないことを確認します。こうしたエラーを回避するため、使用率は 70% 以下にすることをおすすめします。この値を定期的に超える場合は、featurestore を更新してノード数を増やすか、自動スケーリングを使用できます。詳細については、featurestore の管理をご覧ください。

ストリーミング インポート

特定の特徴に値を書き込みます。特徴値はインポート リクエストの一部として含める必要があります。データソースから直接データをストリーミングすることはできません。

最近作成された特徴に書き込む場合は、新しい特徴がまだ伝播されていない可能性があるため、数分待ってから書き込んでください。そのようにしないと、resource not found エラーが表示される場合があります。

1 回の書き込みごとに 1 つのエンティティに対してのみ特徴値をインポートできます。特定のプロジェクトとリージョンの場合、最大 10 種類のエンティティ タイプ内で、複数のエンティティの特徴値を同時に書き込むことができます。この上限には、特定のプロジェクトとリージョン内のすべての featurestore へのストリーミング インポート リクエストが含まれます。この上限を超えると、Vertex AI Feature Store（レガシー）がすべてのデータをオフライン ストアに書き込めない可能性があります。この場合、Vertex AI Feature Store（従来版）はログ エクスプローラにエラーを記録します。詳細については、ストリーミング インポートのオフライン ストレージ書き込みエラーをモニタリングするをご覧ください。

POST http://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT/locations/LOCATION/featurestores/FEATURESTORE_ID/entityTypes/ENTITY_TYPE_ID:writeFeatureValues

{
  "payloads": [
    {
      "entityId": "ENTITY_ID",
      "featureValues": {
        "FEATURE_ID": {
          "VALUE_TYPE": VALUE,
          "metadata": {"generate_time": "TIME_STAMP"}
        }
      }
    }
  ]
}

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "http://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT/locations/LOCATION/featurestores/FEATURESTORE_ID/entityTypes/ENTITY_TYPE_ID:writeFeatureValues"

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "http://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT/locations/LOCATION/featurestores/FEATURESTORE_ID/entityTypes/ENTITY_TYPE_ID:writeFeatureValues" | Select-Object -Expand Content

from google.cloud import aiplatform

def write_feature_values_sample(
    project: str, location: str, entity_type_id: str, featurestore_id: str
):

    aiplatform.init(project=project, location=location)

    my_entity_type = aiplatform.featurestore.EntityType(
        entity_type_name=entity_type_id, featurestore_id=featurestore_id
    )

    my_data = {
        "movie_01": {
            "title": "The Shawshank Redemption",
            "average_rating": 4.7,
            "genre": "Drama",
        },
    }

    my_entity_type.write_feature_values(instances=my_data)

次のステップ

• ストリーミング インポートのオフライン ストレージ書き込みエラーをモニタリングする方法を学習する。
• オンライン サービングまたはバッチ サービングで特徴を処理する方法を学習する。
• Vertex AI Feature Store（従来版）に関する一般的な問題のトラブルシューティングを行う。