流式导入

流式导入使您可以对特征值进行实时更新。如果获取最新可用数据以进行在线传送具有高优先级，则此方法非常有用。例如，您可以导入流式事件数据，Vertex AI Feature Store（旧版）会在几秒钟内提供这些数据以用于在线传送场景。

如果您必须回填数据或者您要批量计算特征值，请使用批量导入。与流式导入请求相比，批量导入请求可以处理更大的载荷，但需要更长时间才能完成。

如需了解您可以导入的最早特征值时间戳，请参阅配额和限制中的 Vertex AI Feature Store（旧版）。您无法导入时间戳表示未来日期或时间的特征值。

用例示例

在线零售组织可能会根据用户的当前活动提供个性化购物体验。当用户浏览网站时，您可以将用户的活动捕获到特征存储区中，然后很快地传送所有这些信息用于在线预测。这种实时导入和传送功能可帮助您在购物会话期间向客户显示有用且相关的商品推荐。

在线存储节点用量

将特征值写入在线存储区会使用特征存储区的 CPU 资源（在线存储节点）。请监控您的 CPU 使用率，以确保需求没有超出供应，否则可能导致传送错误。为避免这些错误，我们建议使用率约为 70%。如果您经常超过该值，则可以更新特征存储区以增加节点数或使用自动扩缩功能。如需了解详情，请参阅管理特征存储区。

流式导入

将值写入特定特征。特征值必须包含在导入请求中。您无法直接从数据源流式传输数据。

如果您要写入最近创建的特征，请先等待几分钟，因为新特征可能尚未传播。否则，您可能会看到 resource not found 错误。

每次写入只能导入一个实体的特征值。对于任何特定项目和区域，您最多可以同时写入 10 个不同实体类型的多个实体的特征值。此限制包括对一个给定项目和区域中所有特征存储区的流式导入请求。如果您超出此限制，Vertex AI Feature Store（旧版）可能不会将所有数据写入离线存储区。如果发生这种情况，Vertex AI Feature Store（旧版）会在 Logs Explorer 中记录错误。如需了解详情，请参阅监控流式导入的离线存储写入错误。

POST https://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 \
     "https://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 "https://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（旧版）常见问题。