Importa valores de atributos por lotes

La transferencia por lotes permite transferir valores de atributos de forma masiva desde una fuente de datos válida. En una solicitud de transferencia por lotes, puedes importar valores para hasta 100 funciones de un tipo de entidad. Ten en cuenta que solo puedes ejecutar un trabajo de transferencia por lotes para evitar colisiones por tipo de entidad.

En una solicitud de transferencia por lotes, especifique la ubicación de sus datos de origen y cómo se asigna a los atributos del almacén de atributos. Debido a que cada solicitud de transferencia por lotes está destinada a un solo tipo de entidad, sus datos de origen también deben ser para un solo tipo de entidad.

Una vez que la importación se haya completado de forma correcta, los valores de los atributos estarán disponibles en las operaciones de lectura posteriores.

Rendimiento de los trabajos de importación

Vertex AI Feature Store (Legacy) proporciona transferencias de alta capacidad de procesamiento, pero la latencia mínima puede tardar unos minutos. Cada solicitud a Vertex AI Feature Store (Legacy) inicia un trabajo para completar la tarea. Un trabajo de importación tarda unos minutos en completarse, incluso si estás importando un solo registro.

Si deseas realizar ajustes en el rendimiento de un trabajo, cambia las siguientes dos variables:

  • La cantidad de nodos de entrega en línea del almacén de atributos.
  • La cantidad de trabajadores que se usaron para el trabajo de transferencia. Los trabajadores procesan y escriben datos en el almacén de atributos.

La cantidad recomendada de trabajadores es un trabajador por cada 10 nodos de entrega en línea en el almacén de atributos. Puedes aumentarla si la carga de la entrega en línea es baja. Puedes especificar un máximo de 100 trabajadores. Si deseas obtener más orientación, consulta Supervisa y ajusta los recursos según corresponda para optimizar la transferencia por lotes.

Si el clúster de entrega en línea no está suficientemente aprovisionado, el trabajo de transferencia puede fallar. En caso de falla, vuelve a intentar la solicitud de importación cuando la carga de la entrega en línea sea baja, o aumenta el recuento de nodos de tu almacén de atributos y vuelve a intentar la solicitud.

Si el almacén de atributos no tiene una tienda en línea (cero nodos de entrega en línea), el trabajo de transferencia solo escribe en la tienda sin conexión y el rendimiento del trabajo depende solo de la cantidad de trabajadores de transferencia.

Coherencia de los datos

Se pueden generar inconsistencias si los datos de origen se modifican durante la importación. Asegúrate de que las modificaciones de los datos de origen se hayan completado antes de iniciar un trabajo de transferencia. Además, los valores de atributos duplicados pueden hacer que se entreguen valores diferentes entre las solicitudes en línea y por lotes. Asegúrate de tener un valor de atributo para cada par de ID de la entidad y marca de tiempo.

Si una operación de importación falla, la tienda de atributos puede tener solo datos parciales, lo que puede provocar que se muestren valores incoherentes entre las solicitudes de entrega en línea y por lotes. Para evitar esta incoherencia, vuelve a intentar la misma solicitud de importación y espera hasta que la solicitud se complete de forma correcta.

Valores nulos y arrays vacíos

Durante la transferencia, Vertex AI Feature Store (Legacy) considera valores escalares nulos o arrays vacíos como valores vacíos. Estos incluyen valores vacíos en una columna CSV. Vertex AI Feature Store (Legacy) no admite valores nulos no escalares, como un valor null en una matriz.

Durante la entrega en línea y por lotes, Vertex AI Feature Store (Legacy) muestra el último valor no nulo o no vacío de la función. Si un valor histórico de la función no está disponible, Vertex AI Feature Store (Legacy) muestra null.

Valores NaN

Vertex AI Feature Store (Legacy) admite valores NaN (no un número) en Double y DoubleArray. Durante la transferencia, puedes ingresar NaN en el archivo CSV de entrada de entrega para representar un valor NaN. Durante la entrega en línea y la entrega por lotes, Vertex AI Feature Store (Legacy) muestra NaN para los valores NaN.

batchImport

Importa valores de forma masiva en un almacén de atributos para una o más atributos de un solo tipo de entidad.

IU web

  1. En la sección de Vertex AI de la consola de Google Cloud, ve a la página Funciones.

    Ve a la página Atributos

  2. Selecciona una región de la lista desplegable Región.
  3. En la tabla de atributos, consulta la columna Tipo de entidad y busca el tipo de entidad que contiene los atributos para los que deseas transferir valores.
  4. Haz clic en el nombre del tipo de entidad.
  5. En la barra de acciones, haz clic en Transferir valores.
  6. En Fuente de datos, selecciona una de las siguientes opciones:
    • Archivo CSV de Cloud Storage: Selecciona esta opción para transferir datos de varios archivos CSV de Cloud Storage. Especifica la ruta de acceso y el nombre del archivo CSV. Para especificar archivos adicionales, haz clic en Agregar otro archivo.
    • Archivo AVRO de Cloud Storage: Selecciona esta opción para transferir datos de un archivo AVRO desde Cloud Storage. Especifica la ruta y el nombre del archivo AVRO.
    • Tabla de BigQuery: selecciona esta opción para transferir datos desde una tabla de BigQuery o una vista de BigQuery. Explora y selecciona una tabla o vista para usar, que tiene el siguiente formato: PROJECT_ID.DATASET_ID.TABLE_ID
  7. Haz clic en Continuar.
  8. En Asignar columna a atributos, especifica qué columnas de tus datos de origen se asignan a entidades y atributos en tu almacén de atributos.
    1. Especifica el nombre de la columna en los datos de origen que contiene los ID de las entidades.
    2. Para la marca de tiempo, especifica una columna de marca de tiempo en los datos de origen o especifica una sola marca de tiempo asociada con todos los valores de atributos que transfieras.
    3. En la lista de atributos, ingresa el nombre de la columna de datos de origen que se asignará a cada atributo. De forma predeterminada, Vertex AI Feature Store (Legacy) supone que el nombre del atributo y el nombre de la columna coinciden.
  9. Haz clic en Transferir.

REST

Para importar los valores de los atributos existentes, envía una solicitud POST mediante el método featurestores.entityTypes.importFeatureValues. Ten en cuenta que, si los nombres de las columnas de datos de origen y los ID de los atributos de destino son diferentes, incluye el parámetro sourceField.

Antes de usar cualquiera de los datos de solicitud a continuación, realiza los siguientes reemplazos:

  • LOCATION_ID: Región en la que se crea el featurestore. Por ejemplo, us-central1
  • PROJECT_ID: El ID del proyecto.
  • FEATURESTORE_ID: ID del featurestore.
  • ENTITY_TYPE_ID: Es el ID del tipo de entidad.
  • ENTITY_SOURCE_COLUMN_ID: Es el ID de la columna de origen que contiene los ID de la entidad.
  • FEATURE_TIME_ID: ID de la columna de origen que contiene las marcas de tiempo de los valores de los atributos.
  • FEATURE_ID: Es el ID de un atributo existente en la tienda de atributos para la cual se importarán valores.
  • FEATURE_SOURCE_COLUMN_ID: Es el ID de la columna de origen que contiene los valores de los atributos de las entidades.
  • SOURCE_DATA_DETAILS: La ubicación de los datos de origen, que también indica el formato, como "bigquerySource": { "inputUri": "bq://test.dataset.sourcetable" } para una tabla de BigQuery o una vista de BigQuery.
  • WORKER_COUNT: Es la cantidad de trabajadores que se usarán para escribir datos en la tienda de atributos.

Método HTTP y URL: 

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

Cuerpo JSON de la solicitud:

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

Para enviar tu solicitud, elige una de estas opciones:

curl

Guarda el cuerpo de la solicitud en un archivo llamado request.json y ejecuta el siguiente comando:

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

Guarda el cuerpo de la solicitud en un archivo llamado request.json y ejecuta el siguiente comando:

$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

Deberías ver un resultado similar al siguiente. Puedes usar el OPERATION_ID en la respuesta para obtener el estado de la operación.

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

Si deseas obtener información para instalar o actualizar el SDK de Vertex AI para Python, consulta Instala el SDK de Vertex AI para Python. Si deseas obtener más información, consulta la documentación de referencia de la API de Python. 

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

La biblioteca cliente de Vertex AI se incluye cuando instalas el SDK de Vertex AI para Python. Si deseas obtener información sobre cómo instalar el SDK de Vertex AI para Python, consulta Instala el SDK de Vertex AI para Python. Si deseas obtener más información, consulta la documentación de referencia del SDK de AI de Vertex para la API de Python. 

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

Antes de probar este ejemplo, sigue las instrucciones de configuración para Java incluidas en la guía de inicio rápido de Vertex AI sobre cómo usar bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Vertex AI Java.

Para autenticarte en Vertex AI, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 


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

Antes de probar este ejemplo, sigue las instrucciones de configuración para Node.js incluidas en la guía de inicio rápido de Vertex AI sobre cómo usar bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Vertex AI Node.js.

Para autenticarte en Vertex AI, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

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

Visualiza trabajos de importación

Usa la consola de Google Cloud para ver los trabajos de importación por lotes en un proyecto de Google Cloud.

IU web

  1. En la sección de Vertex AI de la consola de Google Cloud, ve a la página Funciones.

    Ve a la página Atributos

  2. Selecciona una región de la lista desplegable Región.
  3. En la barra de acciones, haz clic en Ver trabajos de transferencia para enumerar los trabajos de transferencia de todos los almacenes de atributos.
  4. Haz clic en el ID de un trabajo de importación para ver sus detalles, como sus fuente de datos, cantidad de entidades de importación y cantidad de valores de atributos importados.

Reemplazar los datos existentes en un almacén de atributos

Puedes volver a importar valores para reemplazar los valores de atributos existentes si ambos tienen las mismas marcas de tiempo. No es necesario que borres primero los valores de los atributos existentes. Por ejemplo, puedes confiar en un dato de origen subyacente que se modificó recientemente. Para mantener el almacén de atributos coherente con esos datos subyacentes, vuelve a importar los valores de tus atributos. Si tienes marcas de tiempo que no coinciden, los valores importados se consideran únicos y los valores antiguos continúan existiendo (no se reemplazan).

Para garantizar la coherencia entre las solicitudes de entrega en línea y por lotes, espera a que se complete el trabajo de importación antes de realizar cualquier solicitud de entrega.

Datos históricos de reabastecimiento

Si reabasteces datos, en los que importas valores de atributos anteriores, inhabilita la entrega en línea para el trabajo de importación. La entrega en línea sirve para entregar solo los valores de función más recientes, que el reabastecimiento no incluye. Inhabilitar la entrega en línea es útil porque borras cualquier carga en los nodos de entrega en línea y aumentas la capacidad de procesamiento de tu trabajo de importación, lo que puede disminuir su tiempo de finalización.

Puedes inhabilitar la entrega en línea para trabajos de importación cuando usas la API o las bibliotecas cliente. Para obtener más información, consulta el campo disableOnlineServing del método importFeatureValue.

¿Qué sigue?