Connetti e archivia i dati in BigQuery

Quando aggiungi un connettore BigQuery all'app Gemini Enterprise Agent Platform Vision, tutti gli output del modello dell'app connessa verranno inseriti nella tabella di destinazione.

Puoi creare una tabella BigQuery e specificarla quando aggiungi un connettore BigQuery all'app oppure lasciare che la piattaforma app Gemini Enterprise Agent Platform Vision crei automaticamente la tabella.

Attenzione :i dati che importi in BigQuery vengono prima scritti nel buffer di streaming prima di essere resi persistenti nello spazio di archiviazione. Non puoi utilizzare la scheda Anteprima per visualizzare queste righe nel buffer dei flussi. Tuttavia, puoi comunque leggere queste righe con le query SQL. Utilizza la scheda Dettagli della tabella BigQuery per visualizzare le metriche sui dati del buffer di streaming.

Creazione automatica delle tabelle

Se consenti alla piattaforma di app Gemini Enterprise Agent Platform 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 set di dati e 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 la tabella con lo stesso nome nello stesso set di dati, non viene creata automaticamente.

Console

1. Apri la scheda Applicazioni della dashboard Vision di Gemini Enterprise Agent Platform.

   Vai alla scheda Applicazioni

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

3. Nella pagina del generatore di applicazioni, seleziona BigQuery dalla sezione Connettori.

4. Lascia vuoto il campo Percorso BigQuery.

   

5. Modifica tutte le eventuali altre impostazioni.

REST & CMD LINE

Per consentire alla piattaforma 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.

Utilizzare 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, vedi Creare e utilizzare tabelle. Per ulteriori informazioni sulla specifica dello schema durante la creazione di una tabella, consulta Specifica di 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 SQL standard. 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"
 }
]

Avviso: la colonna JSON di BigQuery potrebbe non essere disponibile nella console. Se non riesci a creare una colonna JSON nella console, utilizza il comando gcloud riportato di seguito o sostituisci la colonna "JSON" con una colonna "STRING".

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", abilita Modifica come testo.

gcloud

L'esempio seguente crea prima il file JSON di richiesta, poi 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. Effettua le seguenti sostituzioni:

   • TABLE_NAME: l'ID della tabella o l'identificatore completo della tabella.

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

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

Esempio di righe BigQuery generate da un'app Gemini Enterprise Agent Platform 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"]}

Utilizzare 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 sono previsti prerequisiti per lo schema della tabella BigQuery.

Grafico dell'app con il nodo BigQuery selezionato

Il connettore BigQuery può essere collegato a qualsiasi modello che restituisce 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 altri output di annotazione del modello. Il video in sé non viene memorizzato.
• Se il tuo stream non contiene metadati, non verrà memorizzato nulla in BigQuery.

Esegui query sui dati della tabella

Con lo schema della tabella BigQuery predefinito, puoi eseguire analisi avanzate dopo che la tabella è stata compilata con i dati.

Query di esempio

Puoi utilizzare le seguenti query di esempio in BigQuery per ottenere informazioni dai modelli di visione di Gemini Enterprise Agent Platform.

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

Allo stesso modo, 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 attraversano 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.

   

4. Inserisci una query SQL standard Google nell'area di testo editor di query. Per query di esempio, consulta Query di esempio.

5. (Facoltativo) Per modificare la posizione del trattamento dei dati, fai clic su Modifica > Impostazioni query. Nella sezione Località di trattamento, 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.

In questo modo 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 l'elaborazione aggiuntiva dei dati con l'importazione BigQuery personalizzata. Per utilizzare le funzioni Cloud Run per l'importazione BigQuery personalizzata, procedi nel seguente modo:

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

  

• Quando utilizzi l'API Gemini Enterprise Agent Platform 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 è il trigger HTTP della funzione di destinazione.

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

• L'istanza di Cloud Run Functions deve essere creata prima di creare il nodo BigQuery.
• L'API Gemini Enterprise Agent Platform Vision prevede 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 del conteggio di occupazione (OccupancyCountPredictionResult) ed estrarre da esso 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 esempio 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 in Cloud Run Functions, 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 il trigger HTTP corrispondente in BigQueryConfig.