Connetti e archivia i dati in BigQuery

Quando aggiungi un connettore BigQuery all'app Vertex AI Vision, tutte le uscite del modello dell'app connesso vengono importate nella tabella di destinazione.

Puoi creare la tua tabella BigQuery e specificarla quando aggiungi un connettore BigQuery all'app oppure lasciare che sia la piattaforma di app Vertex AI Vision a creare automaticamente la tabella.

Creazione automatica di tabelle

Se consenti alla piattaforma di app Vertex AI Vision di creare automaticamente la tabella, puoi specificare questa opzione quando aggiungi il nodo del connettore BigQuery.

Se vuoi utilizzare la creazione automatica delle tabelle, si applicano le seguenti condizioni per i set di dati e le tabelle:

  • Set di dati: il nome del set di dati creato automaticamente è visionai_dataset.
  • Tabella: il nome della tabella creata automaticamente è visionai_dataset.APPLICATION_ID.

  • Gestione degli errori:

    • Se esiste già una tabella con lo stesso nome nello stesso set di dati, non viene eseguita alcuna creazione automatica.

Console

  1. Apri la scheda Applicazioni della dashboard di Vertex AI Vision.

    Vai alla scheda Applicazioni

  2. Seleziona Visualizza app accanto al nome dell'applicazione nell'elenco.

  3. Nella pagina di creazione dell'applicazione, seleziona BigQuery dalla sezione Connettori.

  4. Lascia vuoto il campo Percorso BigQuery.

    Il percorso della tabella non è stato specificato nell'interfaccia utente

  5. Modificare altre impostazioni.

REST e riga di comando

Per consentire alla piattaforma per app di dedurre uno schema di tabella, utilizza il campo createDefaultTableIfNotExists di BigQueryConfig quando crei o aggiorni un'app.

Creare e specificare manualmente una tabella

Se vuoi gestire manualmente la tabella di output, questa deve avere lo schema richiesto come sottoinsieme dello schema della tabella.

Se la tabella esistente ha schemi incompatibili, il deployment viene rifiutato.

Utilizza lo schema predefinito

Se utilizzi lo schema predefinito per le tabelle di output del modello, assicurati che la tabella contenga solo le seguenti colonne obbligatorie. Puoi copiare direttamente il seguente testo dello schema quando crei la tabella BigQuery. Per informazioni più dettagliate sulla creazione di una tabella BigQuery, consulta Creare e utilizzare le tabelle. Per ulteriori informazioni sulla specifica dello schema quando crei una tabella, consulta Specificare uno schema.

Utilizza il seguente testo per descrivere lo schema quando crei una tabella. Per informazioni sull'utilizzo del tipo di colonna JSON ("type": "JSON"), consulta Utilizzo dei dati JSON in Standard SQL. Il tipo di colonna JSON è consigliato per la query di annotazione. Puoi anche utilizzare "type" : "STRING".

[
  {
    "name": "ingestion_time",
    "type": "TIMESTAMP",
    "mode": "REQUIRED"
  },
 {
   "name": "application",
   "type": "STRING",
   "mode": "REQUIRED"
 },
 {
   "name": "instance",
   "type": "STRING",
   "mode": "REQUIRED"
 },
 {
   "name": "node",
   "type": "STRING",
   "mode": "REQUIRED"
 },
 {
   "name": "annotation",
   "type": "JSON",
   "mode": "REQUIRED"
 }
]

Console Google Cloud

  1. Nella console Google Cloud, vai alla pagina BigQuery.

    Vai a BigQuery

  2. Seleziona il progetto.

  3. Seleziona Altre opzioni .

  4. Fai clic su Crea tabella.

  5. Nella sezione "Schema", attiva Modifica come testo.

immagine schema predefinita

gcloud

L'esempio seguente crea prima il file JSON della richiesta, quindi utilizza il comando gcloud alpha bq tables create.

  1. Per prima cosa, crea il file JSON di richiesta:

    echo "{
\"schema\": [
    {
      \"name\": \"ingestion_time\",
      \"type\": \"TIMESTAMP\",
      \"mode\": \"REQUIRED\"
    },
    {
      \"name\": \"application\",
      \"type\": \"STRING\",
      \"mode\": \"REQUIRED\"
    },
    {
      \"name\": \"instance\",
      \"type\": \"STRING\",
      \"mode\": \"REQUIRED\"
    },
    {
      \"name\": \"node\",
      \"type\": \"STRING\",
      \"mode\": \"REQUIRED\"
    },
    {
      \"name\": \"annotation\",
      \"type\": \"JSON\",
      \"mode\": \"REQUIRED\"
    }
]
}
" >> bigquery_schema.json

  2. Invia il comando gcloud. Esegui le seguenti sostituzioni:

    • TABLE_NAME: l'ID della tabella o l'identificatore completamente qualificato della tabella.

    • DATASET: l'ID del set di dati BigQuery.

    gcloud alpha bq tables create TABLE_NAME \
--dataset=DATASET \
--schema-file=./bigquery_schema.json

Righe BigQuery di esempio generate da un'app Vertex AI Vision:

ingestion_time applicazione istanza nodo annotazione
2022-05-11 23:3211.911378 UTC my_application 5 just-one-node {"bytesFields": ["Ig1qdXN0LW9uZS1ub2RIGgE1Eg5teV9hcHBsaWNhdGlvbgjS+YnOzdj3Ag=="],"displayNames":["hello","world"],"ids":["12345","34567"]}
2022-05-11 23:3211.911338 UTC my_application 1 just-one-node {"bytesFields": ["Ig1qdXN0LW9uZS1ub2RIGgExEg5teV9hcHBsaWNhdGlvbgiq+YnOzdj3Ag=="],"displayNames":["hello","world"],"ids":["12345","34567"]}
2022-05-11 23:3211.911313 UTC my_application 4 just-one-node {"bytesFields": ["Ig1qdXN0LW9uZS1ub2RIGgE0Eg5teV9hcHBsaWNhdGlvbgiR+YnOzdj3Ag=="],"displayNames":["hello","world"],"ids":["12345","34567"]}
2022-05-11 23:3212.235327 UTC my_application 4 just-one-node {"bytesFields": ["Ig1qdXN0LW9uZS1ub2RIGgE0Eg5teV9hcHBsaWNhdGlvbgi/3J3Ozdj3Ag=="],"displayNames":["hello","world"],"ids":["12345","34567"]}

Utilizza uno schema personalizzato

Se lo schema predefinito non funziona per il tuo caso d'uso, puoi utilizzare le funzioni Cloud Run per generare righe BigQuery con uno schema definito dall'utente. Se utilizzi uno schema personalizzato, non è necessario alcun prerequisito per lo schema della tabella BigQuery.

Grafo dell'app con il nodo BigQuery selezionato

Grafo delle app collegato a BigQuery

Il connettore BigQuery può essere collegato a qualsiasi modello che generi annotazioni basate su video o proto:

  • Per l'input video, il connettore BigQuery estrae solo i dati dei metadati archiviati nell'intestazione dello stream e li importa in BigQuery come altre uscite di annotazione del modello. Il video stesso non viene archiviato.
  • Se lo stream non contiene nessun metadato, non verrà memorizzato nulla in BigQuery.

Esegui query sui dati della tabella

Con lo schema della tabella BigQuery predefinito, puoi eseguire analisi efficaci dopo aver compilato la tabella con i dati.

Query di esempio

Puoi utilizzare le seguenti query di esempio in BigQuery per ottenere informazioni dai modelli Vertex AI Vision.

Ad esempio, puoi utilizzare BigQuery per tracciare una curva basata sul tempo per il numero massimo di persone rilevate al minuto utilizzando i dati del modello di rilevamento di persone / veicoli con la seguente query:

WITH
 nested3 AS(
 WITH
   nested2 AS (
   WITH
     nested AS (
     SELECT
       t.ingestion_time AS ingestion_time,
       JSON_QUERY_ARRAY(t.annotation.stats["fullFrameCount"]) AS counts
     FROM
       `PROJECT_ID.DATASET_NAME.TABLE_NAME` AS t)
   SELECT
     ingestion_time,
     e
   FROM
     nested,
     UNNEST(nested.counts) AS e)
 SELECT
   STRING(TIMESTAMP_TRUNC(nested2.ingestion_time, MINUTE, "America/Los_Angeles"),"America/Los_Angeles") AS time,
   IFNULL(INT64(nested2.e["count"]), 0) AS person_count
 FROM
   nested2
 WHERE
   JSON_VALUE(nested2.e["entity"]["labelString"])="Person")
SELECT
 time,
 MAX(person_count)
FROM
 nested3
GROUP BY
 time

Analogamente, puoi utilizzare BigQuery e la funzionalità di conteggio delle linee di attraversamento del modello di analisi dell'occupazione per creare una query che conteggia il numero totale di veicoli che passano la linea di attraversamento al minuto:

WITH
 nested4 AS (
 WITH
   nested3 AS (
   WITH
     nested2 AS (
     WITH
       nested AS (
       SELECT
         t.ingestion_time AS ingestion_time,
         JSON_QUERY_ARRAY(t.annotation.stats["crossingLineCounts"]) AS lines
       FROM
         `PROJECT_ID.DATASET_NAME.TABLE_NAME` AS t)
     SELECT
       nested.ingestion_time,
       JSON_QUERY_ARRAY(line["positiveDirectionCounts"]) AS entities
     FROM
       nested,
       UNNEST(nested.lines) AS line
     WHERE
       JSON_VALUE(line.annotation.id) = "LINE_ANNOTATION_ID")
   SELECT
     ingestion_time,
     entity
   FROM
     nested2,
     UNNEST(nested2.entities) AS entity )
 SELECT
   STRING(TIMESTAMP_TRUNC(nested3.ingestion_time, MINUTE, "America/Los_Angeles"),"America/Los_Angeles") AS time,
   IFNULL(INT64(nested3.entity["count"]), 0) AS vehicle_count
 FROM
   nested3
 WHERE
   JSON_VALUE(nested3.entity["entity"]["labelString"])="Vehicle" )
SELECT
 time,
 SUM(vehicle_count)
FROM
 nested4
GROUP BY
 time

Esegui la query

Dopo aver formattato la query SQL standard di Google, puoi utilizzare la console per eseguirla:

Console

  1. Nella console Google Cloud, apri la pagina BigQuery.

    Vai a BigQuery

  2. Seleziona Espandi accanto al nome del set di dati e seleziona il nome della tabella.

  3. Nella visualizzazione dei dettagli della tabella, fai clic su Crea nuova query.

    Crea nuova query

  4. Inserisci una query SQL standard di Google nell'area di testo Editor query. Per esempi di query, consulta le query di esempio.

  5. (Facoltativo) Per cambiare la località di elaborazione dei dati, fai clic su Altro e poi su Impostazioni query. In Località di elaborazione, fai clic su Selezione automatica e scegli la località dei dati. Infine, fai clic su Salva per aggiornare le impostazioni della query.

  6. Fai clic su Esegui.

Viene creato un job di query che scrive l'output in una tabella temporanea.

Integrazione di Cloud Run Functions

Puoi utilizzare le funzioni Cloud Run per attivare un'ulteriore elaborazione dei dati con l'importazione di BigQuery personalizzata. Per utilizzare le funzioni Cloud Run per l'importazione BigQuery personalizzata, svolgi i seguenti passaggi:

  • Quando utilizzi la console Google Cloud, seleziona la funzione cloud corrispondente dal menu a discesa di ogni modello collegato.

    seleziona immagine funzione cloud

  • Quando utilizzi l'API Vertex AI Vision, aggiungi una coppia chiave-valore al campo cloud_function_mapping di BigQueryConfig nel nodo BigQuery. La chiave è il nome del nodo BigQuery e il valore è l'attivatore HTTP della funzione di destinazione.

Per utilizzare le funzioni Cloud Run con l'importazione BigQuery personalizzata, la funzione deve soddisfare i seguenti requisiti:

  • L'istanza delle funzioni Cloud Run deve essere creata prima di creare il nodo BigQuery.
  • L'API Vertex AI Vision si aspetta di ricevere un'annotazione AppendRowsRequest restituita dalle funzioni Cloud Run.
  • Devi impostare il campo proto_rows.writer_schema per tutte le risposte CloudFunction. write_stream può essere ignorato.

Esempio di integrazione di Cloud Run Functions

L'esempio seguente mostra come analizzare l'output del nodo di conteggio persone (OccupancyCountPredictionResult) e ricavarne uno schema di tabella ingestion_time, person_count e vehicle_count.

Il risultato del seguente esempio è una tabella BigQuery con lo schema:

[
  {
    "name": "ingestion_time",
    "type": "TIMESTAMP",
    "mode": "REQUIRED"
  },
  {
    "name": "person_count",
    "type": "INTEGER",
    "mode": "NULLABLE"
  },
      {
    "name": "vehicle_count",
    "type": "INTEGER",
    "mode": "NULLABLE"
  },
]

Utilizza il seguente codice per creare questa tabella:

  1. Definisci un proto (ad es. test_table_schema.proto) per i campi della tabella che vuoi scrivere:

    syntax = "proto3";

package visionai.testing;

message TestTableSchema {
  int64 ingestion_time = 1;
  int32 person_count = 2;
  int32 vehicle_count = 3;
}

  2. Compila il file proto per generare il file Python del buffer del protocollo:

    protoc -I=./ --python_out=./ ./test_table_schema.proto

  3. Importa il file Python generato e scrivi la funzione Cloud.

    Python

    import base64
import sys

from flask import jsonify
import functions_framework
from google.protobuf import descriptor_pb2
from google.protobuf.json_format import MessageToDict
import test_table_schema_pb2

def table_schema():
  schema = descriptor_pb2.DescriptorProto()
  test_table_schema_pb2.DESCRIPTOR.message_types_by_name[
      'TestTableSchema'].CopyToProto(schema)
  return schema

def bigquery_append_row_request(row):
  append_row_request = {}
  append_row_request['protoRows'] = {
      'writerSchema': {
          'protoDescriptor': MessageToDict(table_schema())
      },
      'rows': {
          'serializedRows':
              base64.b64encode(row.SerializeToString()).decode('utf-8')
      }
  }
  return append_row_request

@functions_framework.http
def hello_http(request):
  request_json = request.get_json(silent=False)
  annotations = []
  payloads = []
  if request_json and 'annotations' in request_json:
    for annotation_with_timestamp in request_json['annotations']:
      row = test_table_schema_pb2.TestTableSchema()
      row.person_count = 0
      row.vehicle_count = 0
      if 'ingestionTimeMicros' in annotation_with_timestamp:
        row.ingestion_time = int(
            annotation_with_timestamp['ingestionTimeMicros'])
      if 'annotation' in annotation_with_timestamp:
        annotation = annotation_with_timestamp['annotation']
        if 'stats' in annotation:
          stats = annotation['stats']
          for count in stats['fullFrameCount']:
            if count['entity']['labelString'] == 'Person':
              if 'count' in count:
                row.person_count = count['count']
            elif count['entity']['labelString'] == 'Vehicle':
              if 'count' in count:
                row.vehicle_count = count['count']
      payloads.append(bigquery_append_row_request(row))
  for payload in payloads:
    annotations.append({'annotation': payload})
  return jsonify(annotations=annotations)

  4. Per includere le dipendenze nelle funzioni Cloud Run, devi anche caricare il file test_table_schema_pb2.py generato e specificare requirements.txt simile al seguente:

    functions-framework==3.*
click==7.1.2
cloudevents==1.2.0
deprecation==2.1.0
Flask==1.1.2
gunicorn==20.0.4
itsdangerous==1.1.0
Jinja2==2.11.2
MarkupSafe==1.1.1
pathtools==0.1.2
watchdog==1.0.2
Werkzeug==1.0.1
protobuf==3.12.2

  5. Esegui il deployment della funzione cloud e imposta l'attivatore HTTP corrispondente in BigQueryConfig.