Featurewerte für den Batchimport

Mit dem Batchimport können Sie Featurewerte gesammelt aus einer gültigen Datenquelle aufnehmen. Bei einer Batchimportanfrage können Sie für einen Entitätstyp Werte für bis zu 100 Features importieren. Beachten Sie, dass nur ein Batchimportjob pro Entitätstyp ausgeführt werden kann, um Konflikte zu vermeiden.

Geben Sie in einer Batchimportanfrage den Speicherort Ihrer Quelldaten und die Zuordnung zu Features in Ihrem Featurestore an. Da jede Batchimportanfrage für einen einzelnen Entitätstyp gilt, müssen Ihre Quelldaten auch für einen einzelnen Entitätstyp gelten.

Wenn der Import erfolgreich abgeschlossen wurde, stehen die Featurewerte für nachfolgende Lesevorgänge zur Verfügung.

Leistung von Importjobs

Vertex AI Feature Store (Legacy) bietet einen hohen Durchsatz beim Import, die minimale Latenz kann jedoch einige Minuten dauern. Bei jeder Anfrage an Vertex AI Feature Store (Legacy) wird ein Job gestartet, um die Arbeit abzuschließen. Ein Importjob dauert einige Minuten, selbst wenn Sie einen einzelnen Datensatz importieren.

Wenn Sie Anpassungen der Leistung eines Jobs vornehmen möchten, ändern Sie die folgenden beiden Variablen:

  • Die Anzahl der Onlinebereitstellungsknoten im Feature Store.
  • Die Anzahl der für den Importjob verwendeten Worker. Die Worker verarbeiten und schreiben Daten in den Feature Store.

Die empfohlene Anzahl von Workern ist ein Worker pro 10 Online-Bereitstellungsknoten im Feature Store. Sie können auch einen höheren Wert wählen, wenn die Online-Bereitstellungslast gering ist. Sie können maximal 100 Worker angeben. Weitere Informationen finden Sie unter Ressourcen entsprechend überwachen und abstimmen, um den Batchimport zu optimieren.

Wenn der Onlinebereitstellungscluster nicht ausreichend bereitgestellt wird, kann der Importjob fehlschlagen. Bei einem Fehler versuchen Sie die Importanfrage noch einmal, wenn die Onlinebereitstellungslast gering ist. Sie können auch die Knotenanzahl Ihres Feature Store erhöhen und die Anfrage wiederholen.

Wenn der Feature Store keinen Onlinespeicher hat (null Onlinebereitstellungsknoten), schreibt der Importjob nur in den Offlinespeicher und die Leistung des Jobs hängt ausschließlich von der Anzahl der Import-Worker ab.

Datenkonsistenz

Inkonsistenzen können auftreten, wenn die Quelldaten während des Imports geändert werden. Prüfen Sie, ob alle Änderungen an Quelldaten abgeschlossen sind, bevor Sie einen Importjob starten. Außerdem können doppelte Featurewerte dazu führen, dass zwischen Online- und Batchanfragen unterschiedliche Werte bereitgestellt werden. Achten Sie darauf, ob Sie für jede Entitäts-ID und jedes Zeitstempelpaar einen Featurewert haben.

Wenn ein Importvorgang fehlschlägt, hat der Featurestore möglicherweise nur unvollständige Daten, was dazu führen kann, dass Online- und Batchbereitstellungsanfragen unterschiedliche Werte zurückgeben. Um diese Inkonsistenz zu vermeiden, wiederholen Sie dieselbe Importanfrage und warten Sie, bis die Anfrage erfolgreich abgeschlossen wurde.

Nullwerte und leere Arrays

Beim Import betrachtet Vertex AI Feature Store (Legacy) Null-Skalarwerte oder leere Arrays als leere Werte. Dazu gehören leere Werte in einer CSV-Spalte. Vertex AI Feature Store (Legacy) unterstützt keine nicht skalaren Nullwerte wie einen null-Wert in einem Array.

Während der Online- und Batch-Bereitstellung gibt der Vertex AI Feature Store (Legacy) den neuesten nicht leeren oder nicht leeren Wert des Features zurück. Wenn kein historischer Wert der Funktion verfügbar ist, gibt Vertex AI Feature Store (Legacy) null zurück.

NaN-Werte

Vertex AI Feature Store (Legacy) unterstützt NaN-Werte (keine Zahl) in Double und DoubleArray. Während des Datenimports können Sie NaN in die Bereitstellungs-CSV-Datei eingeben, um einen NaN-Wert darzustellen. Während der Online- und Batch-Bereitstellung gibt der Vertex AI Feature Store (Legacy) NaN für NaN-Werte zurück.

Batchimport

Importieren Sie Werte für ein oder mehrere Features eines einzelnen Entitätstyps in einen Feature Store.

Web-UI

  1. Rufen Sie im Bereich "Vertex AI" der Google Cloud Console die Seite Features auf.

    Zur Seite „Features“

  2. Wählen Sie eine Region aus der Drop-down-Liste Region aus.
  3. Rufen Sie in der Tabelle "Features" die Spalte Entitätstyp auf und suchen Sie den Entitätstyp, der die Features enthält, für die Sie Werte importieren möchten.
  4. Klicken Sie auf den Namen des Entitätstyps.
  5. Klicken Sie in der Aktionsleiste auf Werte aufnehmen.
  6. Wählen Sie als Datenquelle eine der folgenden Optionen aus:
    • Cloud Storage-CSV-Datei: Wählen Sie diese Option aus, um Daten aus mehreren CSV-Dateien aus Cloud Storage zu importieren. Geben Sie den Pfad und den Namen der CSV-Datei an. Wenn Sie weitere Dateien angeben möchten, klicken Sie auf Weitere Datei hinzufügen.
    • Cloud Storage-AVRO-Datei: Wählen Sie diese Option aus, um Daten aus einer AVRO-Datei aus Cloud Storage zu importieren. Geben Sie den Pfad und den Namen der AVRO-Datei an.
    • BigQuery-Tabelle: Wählen Sie diese Option aus, um Daten aus einer BigQuery-Tabelle oder BigQuery-Ansicht zu importieren. Suchen Sie nach einer zu verwendenden Tabelle oder Ansicht, die folgendes Format hat: PROJECT_ID.DATASET_ID.TABLE_ID
  7. Klicken Sie auf Weiter.
  8. Geben Sie unter Spalte Features zuordnen an, welche Spalten in Ihren Quelldaten Entitäten und Features in Ihrem Feature Store zugeordnet werden.
    1. Geben Sie in Ihren Quelldaten den Namen der Spalte an, die Entitäts-IDs enthält.
    2. Geben Sie für den Zeitstempel eine Zeitstempelspalte in Ihren Quelldaten oder einen einzelnen Zeitstempel an, der mit allen zu importierenden Featurewerten verknüpft ist.
    3. Geben Sie in der Liste der Funktionen den Namen der den einzelnen Features zugeordneten Quelldatenspalte ein. Vertex AI Feature Store (Legacy) geht standardmäßig davon aus, dass Featurename und Spaltenname übereinstimmen.
  9. Klicken Sie auf Aufnehmen.

REST

Wenn Sie Featurewerte für vorhandene Features importieren möchten, senden Sie eine POST-Anfrage mit der Methode featurestores.entityTypes.importFeatureValues. Wenn die Namen der Quelldatenspalten und die Zielfeature-IDs unterschiedlich sind, fügen Sie den sourceField-Parameter ein.

Ersetzen Sie diese Werte in den folgenden Anfragedaten:

  • LOCATION_ID: Region, in der der Featurestore erstellt wird. Beispiel: us-central1.
  • PROJECT_ID: Ihre Projekt-ID.
  • FEATURESTORE_ID: ID des Featurestores.
  • ENTITY_TYPE_ID: ID des Entitätstyps.
  • ENTITY_SOURCE_COLUMN_ID: ID der Quellspalte mit den Entitäts-IDs.
  • FEATURE_TIME_ID: ID der Quellspalte, die die Feature-Zeitstempel für die Featurewerte enthält.
  • FEATURE_ID: ID eines vorhandenen Features im Featurestore, für das Werte importiert werden sollen.
  • FEATURE_SOURCE_COLUMN_ID: ID der Quellspalte mit Featurewerten für die Entitäten.
  • SOURCE_DATA_DETAILS: Der Speicherort der Quelldaten, der auch das Format angibt, z. B. "bigquerySource": { "inputUri": "bq://test.dataset.sourcetable" } für eine BigQuery-Tabelle oder BigQuery-Ansicht.
  • WORKER_COUNT: Die Anzahl der Worker, die zum Schreiben von Daten in den Featurestore verwendet werden sollen.

HTTP-Methode und URL:

POST https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/featurestores/FEATURESTORE_ID/entityTypes/ENTITY_TYPE_ID:importFeatureValues

JSON-Text der Anfrage:

{
  "entityIdField": "ENTITY_SOURCE_COLUMN_ID",
  "featureTimeField": "FEATURE_TIME_ID",
  SOURCE_DATA_DETAILS,
  "featureSpecs": [{
    "id": "FEATURE_ID",
    "sourceField": "FEATURE_SOURCE_COLUMN_ID"
  }],
  "workerCount": WORKER_COUNT
}

Wenn Sie die Anfrage senden möchten, wählen Sie eine der folgenden Optionen aus:

curl

Speichern Sie den Anfragetext in einer Datei mit dem Namen request.json und führen Sie den folgenden Befehl aus: 

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

PowerShell

Speichern Sie den Anfragetext in einer Datei mit dem Namen request.json und führen Sie den folgenden Befehl aus: 

$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_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/featurestores/FEATURESTORE_ID/entityTypes/ENTITY_TYPE_ID:importFeatureValues" | Select-Object -Expand Content

Die Ausgabe sieht in etwa so aus: Sie können OPERATION_ID in der Antwort verwenden, um den Status des Vorgangs abzurufen.

{
  "name": "projects/PROJECT_NUMBER/locations/LOCATION_ID/featurestores/FEATURESTORE_ID/entityTypes/ENTITY_TYPE_ID/operations/OPERATION_ID",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.aiplatform.v1.ImportFeatureValuesOperationMetadata",
    "genericMetadata": {
      "createTime": "2021-03-02T00:04:13.039166Z",
      "updateTime": "2021-03-02T00:04:13.039166Z"
    }
  }
}

Python

Informationen zur Installation des Vertex AI SDK for Python finden Sie unter Vertex AI SDK for Python installieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Python API. 

import datetime
from typing import List, Union

from google.cloud import aiplatform


def import_feature_values_sample(
    project: str,
    location: str,
    entity_type_id: str,
    featurestore_id: str,
    feature_ids: List[str],
    feature_time: Union[str, datetime.datetime],
    gcs_source_uris: Union[str, List[str]],
    gcs_source_type: str,
):

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

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

    my_entity_type.ingest_from_gcs(
        feature_ids=feature_ids,
        feature_time=feature_time,
        gcs_source_uris=gcs_source_uris,
        gcs_source_type=gcs_source_type,
    )

Python

Die Clientbibliothek für Vertex AI ist bei der Installation des Vertex AI SDK für Python enthalten. Informationen zur Installation des Vertex AI SDK für Python finden Sie unter Vertex AI SDK für Python installieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Vertex AI SDK for Python API. 

from google.cloud import aiplatform


def import_feature_values_sample(
    project: str,
    featurestore_id: str,
    entity_type_id: str,
    avro_gcs_uri: str,
    entity_id_field: str,
    feature_time_field: str,
    worker_count: int = 2,
    location: str = "us-central1",
    api_endpoint: str = "us-central1-aiplatform.googleapis.com",
    timeout: int = 300,
):
    # The AI Platform services require regional API endpoints, which need to be
    # in the same region or multi-region overlap with the Feature Store location.
    client_options = {"api_endpoint": api_endpoint}
    # Initialize client that will be used to create and send requests.
    # This client only needs to be created once, and can be reused for multiple requests.
    client = aiplatform.gapic.FeaturestoreServiceClient(client_options=client_options)
    entity_type = f"projects/{project}/locations/{location}/featurestores/{featurestore_id}/entityTypes/{entity_type_id}"
    avro_source = aiplatform.gapic.AvroSource(
        gcs_source=aiplatform.gapic.GcsSource(uris=[avro_gcs_uri])
    )
    feature_specs = [
        aiplatform.gapic.ImportFeatureValuesRequest.FeatureSpec(id="age"),
        aiplatform.gapic.ImportFeatureValuesRequest.FeatureSpec(id="gender"),
        aiplatform.gapic.ImportFeatureValuesRequest.FeatureSpec(id="liked_genres"),
    ]
    import_feature_values_request = aiplatform.gapic.ImportFeatureValuesRequest(
        entity_type=entity_type,
        avro_source=avro_source,
        feature_specs=feature_specs,
        entity_id_field=entity_id_field,
        feature_time_field=feature_time_field,
        worker_count=worker_count,
    )
    lro_response = client.import_feature_values(request=import_feature_values_request)
    print("Long running operation:", lro_response.operation.name)
    import_feature_values_response = lro_response.result(timeout=timeout)
    print("import_feature_values_response:", import_feature_values_response)

Java

Bevor Sie dieses Beispiel anwenden, folgen Sie den Java-Einrichtungsschritten in der Vertex AI-Kurzanleitung zur Verwendung von Clientbibliotheken. Weitere Informationen finden Sie in der Referenzdokumentation zur Vertex AI Java API.

Richten Sie zur Authentifizierung bei Vertex AI Standardanmeldedaten für Anwendungen ein. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten. 


import com.google.api.gax.longrunning.OperationFuture;
import com.google.cloud.aiplatform.v1.AvroSource;
import com.google.cloud.aiplatform.v1.EntityTypeName;
import com.google.cloud.aiplatform.v1.FeaturestoreServiceClient;
import com.google.cloud.aiplatform.v1.FeaturestoreServiceSettings;
import com.google.cloud.aiplatform.v1.GcsSource;
import com.google.cloud.aiplatform.v1.ImportFeatureValuesOperationMetadata;
import com.google.cloud.aiplatform.v1.ImportFeatureValuesRequest;
import com.google.cloud.aiplatform.v1.ImportFeatureValuesRequest.FeatureSpec;
import com.google.cloud.aiplatform.v1.ImportFeatureValuesResponse;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class ImportFeatureValuesSample {

  public static void main(String[] args)
      throws IOException, InterruptedException, ExecutionException, TimeoutException {
    // TODO(developer): Replace these variables before running the sample.
    String project = "YOUR_PROJECT_ID";
    String featurestoreId = "YOUR_FEATURESTORE_ID";
    String entityTypeId = "YOUR_ENTITY_TYPE_ID";
    String entityIdField = "YOUR_ENTITY_FIELD_ID";
    String featureTimeField = "YOUR_FEATURE_TIME_FIELD";
    String gcsSourceUri = "YOUR_GCS_SOURCE_URI";
    int workerCount = 2;
    String location = "us-central1";
    String endpoint = "us-central1-aiplatform.googleapis.com:443";
    int timeout = 300;
    importFeatureValuesSample(
        project,
        featurestoreId,
        entityTypeId,
        gcsSourceUri,
        entityIdField,
        featureTimeField,
        workerCount,
        location,
        endpoint,
        timeout);
  }

  static void importFeatureValuesSample(
      String project,
      String featurestoreId,
      String entityTypeId,
      String gcsSourceUri,
      String entityIdField,
      String featureTimeField,
      int workerCount,
      String location,
      String endpoint,
      int timeout)
      throws IOException, InterruptedException, ExecutionException, TimeoutException {
    FeaturestoreServiceSettings featurestoreServiceSettings =
        FeaturestoreServiceSettings.newBuilder().setEndpoint(endpoint).build();

    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests. After completing all of your requests, call
    // the "close" method on the client to safely clean up any remaining background resources.
    try (FeaturestoreServiceClient featurestoreServiceClient =
        FeaturestoreServiceClient.create(featurestoreServiceSettings)) {
      List<FeatureSpec> featureSpecs = new ArrayList<>();

      featureSpecs.add(FeatureSpec.newBuilder().setId("title").build());
      featureSpecs.add(FeatureSpec.newBuilder().setId("genres").build());
      featureSpecs.add(FeatureSpec.newBuilder().setId("average_rating").build());
      ImportFeatureValuesRequest importFeatureValuesRequest =
          ImportFeatureValuesRequest.newBuilder()
              .setEntityType(
                  EntityTypeName.of(project, location, featurestoreId, entityTypeId).toString())
              .setEntityIdField(entityIdField)
              .setFeatureTimeField(featureTimeField)
              .addAllFeatureSpecs(featureSpecs)
              .setWorkerCount(workerCount)
              .setAvroSource(
                  AvroSource.newBuilder()
                      .setGcsSource(GcsSource.newBuilder().addUris(gcsSourceUri)))
              .build();
      OperationFuture<ImportFeatureValuesResponse, ImportFeatureValuesOperationMetadata>
          importFeatureValuesFuture =
              featurestoreServiceClient.importFeatureValuesAsync(importFeatureValuesRequest);
      System.out.format(
          "Operation name: %s%n", importFeatureValuesFuture.getInitialFuture().get().getName());
      System.out.println("Waiting for operation to finish...");
      ImportFeatureValuesResponse importFeatureValuesResponse =
          importFeatureValuesFuture.get(timeout, TimeUnit.SECONDS);
      System.out.println("Import Feature Values Response");
      System.out.println(importFeatureValuesResponse);
      featurestoreServiceClient.close();
    }
  }
}

Node.js

Bevor Sie dieses Beispiel anwenden, folgen Sie den Node.js-Einrichtungsschritten in der Vertex AI-Kurzanleitung zur Verwendung von Clientbibliotheken. Weitere Informationen finden Sie in der Referenzdokumentation zur Vertex AI Node.js API.

Richten Sie zur Authentifizierung bei Vertex AI Standardanmeldedaten für Anwendungen ein. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten. 

/**
 * TODO(developer): Uncomment these variables before running the sample.\
 * (Not necessary if passing values as arguments)
 */

// const project = 'YOUR_PROJECT_ID';
// const featurestoreId = 'YOUR_FEATURESTORE_ID';
// const entityTypeId = 'YOUR_ENTITY_TYPE_ID';
// const avroGcsUri = 'AVRO_FILE_IN_THE_GCS_URI';
// const entityIdField = 'ENTITY_ID_FIELD_IN_AVRO';
// const featureTimeField = 'TIMESTAMP_FIELD_IN_AVRO';
// const workerCount = <NO_OF_WORKERS_FOR_INGESTION_JOB>;
// const location = 'YOUR_PROJECT_LOCATION';
// const apiEndpoint = 'YOUR_API_ENDPOINT';
// const timeout = <TIMEOUT_IN_MILLI_SECONDS>;

// Imports the Google Cloud Featurestore Service Client library
const {FeaturestoreServiceClient} = require('@google-cloud/aiplatform').v1;

// Specifies the location of the api endpoint
const clientOptions = {
  apiEndpoint: apiEndpoint,
};

// Instantiates a client
const featurestoreServiceClient = new FeaturestoreServiceClient(
  clientOptions
);

async function importFeatureValues() {
  // Configure the entityType resource
  const entityType = `projects/${project}/locations/${location}/featurestores/${featurestoreId}/entityTypes/${entityTypeId}`;

  const avroSource = {
    gcsSource: {
      uris: [avroGcsUri],
    },
  };

  const featureSpecs = [{id: 'age'}, {id: 'gender'}, {id: 'liked_genres'}];

  const request = {
    entityType: entityType,
    avroSource: avroSource,
    entityIdField: entityIdField,
    featureSpecs: featureSpecs,
    featureTimeField: featureTimeField,
    workerCount: Number(workerCount),
  };

  // Import Feature Values Request
  const [operation] = await featurestoreServiceClient.importFeatureValues(
    request,
    {timeout: Number(timeout)}
  );
  const [response] = await operation.promise();

  console.log('Import feature values response');
  console.log('Raw response:');
  console.log(JSON.stringify(response, null, 2));
}
importFeatureValues();

Importjobs ansehen

Verwenden Sie die Google Cloud Console, um Batchimportjobs in einem Google Cloud-Projekt anzuzeigen.

Web-UI

  1. Rufen Sie im Bereich "Vertex AI" der Google Cloud Console die Seite Features auf.

    Zur Seite „Features“

  2. Wählen Sie eine Region aus der Drop-down-Liste Region aus.
  3. Klicken Sie in der Aktionsleiste auf Aufnahmejobs ansehen, um Importjobs für alle Feature Stores aufzulisten.
  4. Klicken Sie auf die ID eines Importjobs, um dessen Details wie Datenquelle, Anzahl der Importentitäten und Anzahl der importierten Featurewerte anzuzeigen.

Vorhandene Daten in einem Feature Store überschreiben

Sie können Werte noch einmal importieren, um vorhandene Featurewerte zu überschreiben, wenn beide die gleichen Zeitstempel haben. Sie brauchen vorhandene Featurewerte nicht erst zu löschen. So können Sie sich beispielsweise auf eine kürzlich geänderte Datenquelle verlassen. Damit Ihr Feature Store mit diesen zugrunde liegenden Daten konsistent bleibt, importieren Sie die Featurewerte noch einmal. Wenn Zeitstempel nicht übereinstimmen, werden die importierten Werte als eindeutig angesehen und die alten Werte bleiben bestehen (sie werden nicht überschrieben).

Warten Sie, bis der Importjob abgeschlossen ist, bevor Sie Bereitstellungsanfragen stellen, um Konsistenz zwischen Online- und Batchbereitstellungsanfragen zu gewährleisten.

Backfill für Verlaufsdaten

Wenn Sie ein Backfill für Daten durchführen, bei dem Sie frühere Featurewerte aufnehmen, deaktivieren Sie die Onlinebereitstellung für Ihren Importjob. Bei der Onlinebereitstellung werden nur die neuesten Featurewerte bereitgestellt, die beim Backfill nicht enthalten sind. Die Deaktivierung der Onlinebereitstellung ist nützlich, da Sie Ihre Onlinebereitstellungsknoten nicht mehr belasten und den Durchsatz für Ihren Importjob erhöhen, was dessen Abschlusszeit verkürzen kann.

Sie können die Onlinebereitstellung für Importjobs deaktivieren, wenn Sie die API oder Clientbibliotheken verwenden. Weitere Informationen finden Sie im Feld disableOnlineServing für die Methode importFeatureValue.

Nächste Schritte