Valori delle funzionalità di importazione collettiva

L'importazione in batch ti consente di importare i valori delle caratteristiche in massa da un'origine dati valida. In una richiesta di importazione batch, puoi importare i valori di un massimo di 100 caratteristiche per un tipo di entità. Tieni presente che puoi eseguire un solo job di importazione in batch per ogni tipo di entità, per evitare collisioni.

In una richiesta di importazione in batch, specifica la posizione dei dati di origine e come vengono mappati alle caratteristiche nell'archivio di caratteristiche. Poiché ogni richiesta di importazione in batch riguarda un singolo tipo di entità, anche i dati di origine devono riguardare un singolo tipo di entità.

Una volta completata l'importazione, i valori delle caratteristiche sono disponibili per le successive operazioni di lettura.

• Per informazioni sui requisiti dei dati di origine, consulta Requisiti dei dati di origine.
• Per informazioni sul periodo di conservazione dei dati nel datastore offline da parte di Vertex AI Feature Store (legacy), consulta la sezione Vertex AI Feature Store (legacy) in Quote e limiti.
• Per informazioni sul timestamp del valore della caratteristica meno recente che puoi importare, consulta Vertex AI Feature Store (legacy) in Quote e limiti.
• Non puoi importare valori delle caratteristiche per i quali i timestamp indicano date o ore future.

Rendimento del job di importazione

Vertex AI Feature Store (legacy) offre un'importazione ad alta velocità effettiva, ma la latenza minima può richiedere alcuni minuti. Ogni richiesta a Vertex AI Feature Store (legacy) avvia un job per completare il lavoro. Un job di importazione richiede alcuni minuti per essere completato, anche se stai importando un singolo record.

Se vuoi apportare modifiche al rendimento di un job, modifica le seguenti due variabili:

• Il numero di nodi di pubblicazione online dell'archivio di caratteristiche.
• Il numero di worker utilizzati per il job di importazione. I worker elaborano e scrivono i dati nel featurestore.

Il numero consigliato di worker è uno per ogni 10 nodi di pubblicazione online nello feature store. Puoi aumentare il valore se il carico di pubblicazione online è basso. Puoi specificare un massimo di 100 worker. Per ulteriori indicazioni, consulta Monitorare e ottimizzare le risorse di conseguenza per ottimizzare l'importazione batch.

Se il cluster di pubblicazione online è sottoprovvigionato, il job di importazione potrebbe non riuscire. In caso di errore, riprova la richiesta di importazione quando il carico di pubblicazione online è basso oppure aumenta il numero di nodi dell'archivio di caratteristiche e riprova la richiesta.

Se lo store delle funzionalità non dispone di uno store online (zero nodi di servizio online), il job di importazione scrive solo nello store offline e il rendimento del job dipende esclusivamente dal numero di worker di importazione.

Coerenza dei dati

Possono essere introdotte incongruenze se i dati di origine vengono modificati durante l'importazione. Assicurati che tutte le modifiche ai dati di origine siano completate prima di avviare un job di importazione. Inoltre, i valori delle funzionalità duplicati possono comportare la pubblicazione di valori diversi tra le richieste online e batch. Assicurati di avere un valore di caratteristica per ogni coppia di ID entità e timestamp.

Se un'operazione di importazione non va a buon fine, l'archivio delle caratteristiche potrebbe contenere solo dati parziali, il che può comportare la restituzione di valori incoerenti tra le richieste di pubblicazione online e batch. Per evitare questa incoerenza, riprova la stessa richiesta di importazione e attendi il completamento dell'operazione.

Valori null e array vuoti

Durante l'importazione, Vertex AI Feature Store (legacy) considera i valori scalari null o gli array vuoti come valori vuoti. Questi includono valori vuoti in una colonna CSV. Vertex AI Feature Store (legacy) non supporta i valori null non scalari, ad esempio un valore null in un array.

Durante la distribuzione online e in recupero dati in batch, Vertex AI Feature Store (legacy) restituisce l'ultimo valore non nullo o non vuoto della caratteristica. Se un valore storico della funzionalità non è disponibile, Vertex AI Feature Store (legacy) restituisce null.

Valori NaN

Vertex AI Feature Store (legacy) supporta i valori NaN (Not a Number) in Double e DoubleArray. Durante l'importazione, puoi inserire NaN nel file CSV di input di pubblicazione per rappresentare un valore NaN. Durante la pubblicazione online e in recupero dati in batch, Vertex AI Feature Store (legacy) restituisce NaN per i valori NaN.

Importazione batch

Importa i valori in blocco in un archivio delle caratteristiche per una o più caratteristiche di un singolo tipo di entità.

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

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

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"

$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

{
  "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"
    }
  }
}

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

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

/**
 * 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();

Visualizzare i job di importazione

Utilizza la console Google Cloud per visualizzare i job di importazione batch in un progetto Google Cloud .

Sovrascrivere i dati esistenti in un featurestore

Puoi reimportare i valori per sovrascrivere i valori delle caratteristiche esistenti se entrambi hanno gli stessi timestamp. Non è necessario eliminare prima i valori delle caratteristiche esistenti. Ad esempio, potresti fare affidamento su dati di origine sottostanti che sono stati modificati di recente. Per mantenere la coerenza dell'archivio di caratteristiche con i dati sottostanti, importa di nuovo i valori delle caratteristiche. Se i timestamp non corrispondono, i valori importati vengono considerati unici e i valori precedenti continuano a esistere (non vengono sovrascritti).

Per garantire la coerenza tra le richieste di pubblicazione online e batch, attendi il completamento del job di importazione prima di effettuare qualsiasi richiesta di pubblicazione.

Esegui backfill dei dati storici

Se stai eseguendo il backfill dei dati, ovvero importando i valori passati delle funzionalità, disattiva la pubblicazione online per il job di importazione. La pubblicazione online serve solo per pubblicare gli ultimi valori delle caratteristiche, che non sono inclusi nel backfilling. La disattivazione della pubblicazione online è utile perché elimina qualsiasi carico sui nodi di pubblicazione online e aumenta la velocità effettiva del job di importazione, il che può ridurre il tempo di completamento.

Puoi disattivare la pubblicazione online per i job di importazione quando utilizzi l'API o le librerie client. Per ulteriori informazioni, consulta il campo disableOnlineServing per il metodo importFeatureValue.

Passaggi successivi

• Scopri come pubblicare le funzionalità tramite la distribuzione online o la distribuzione batch.
• Scopri di più su come monitorare i valori delle caratteristiche importate nel tempo.
• Visualizza la quota di job batch simultanei di Vertex AI Feature Store (legacy).
• Risolvi i problemi comuni di Vertex AI Feature Store (legacy).