Associe e armazene dados no BigQuery

Quando adiciona um conetor do BigQuery à sua app Vertex AI Vision, todas as saídas do modelo da app associada são carregadas para a tabela de destino.

Pode criar a sua própria tabela do BigQuery e especificá-la quando adicionar um conetor do BigQuery à app ou permitir que a plataforma de apps do Vertex AI Vision crie automaticamente a tabela.

Aviso: os dados que carrega para o BigQuery são primeiro escritos no buffer de streaming antes de serem mantidos no armazenamento. Não pode usar o separador Pré-visualização para ver estas linhas no buffer de streaming. No entanto, ainda pode ler estas linhas com consultas SQL. Use o separador Detalhes da tabela do BigQuery para ver métricas sobre os dados do buffer de streaming.

Criação automática de tabelas

Se permitir que a plataforma de apps Vertex AI Vision crie automaticamente a tabela, pode especificar esta opção quando adicionar o nó do conetor do BigQuery.

As seguintes condições de conjunto de dados e tabela aplicam-se se quiser usar a criação automática de tabelas:

• Conjunto de dados: o nome do conjunto de dados criado automaticamente é visionai_dataset.
• Tabela: o nome da tabela criado automaticamente é visionai_dataset.APPLICATION_ID.

• Processamento de erros:

  • Se existir uma tabela com o mesmo nome no mesmo conjunto de dados, não é feita a criação automática.

Consola

1. Abra o separador Aplicações do painel de controlo do Vertex AI Vision.

   Aceda ao separador Aplicações

2. Selecione Ver app junto ao nome da sua aplicação na lista.

3. Na página do criador de aplicações, selecione BigQuery na secção Conetores.

4. Deixe o campo Caminho do BigQuery vazio.

   

5. Altere quaisquer outras definições.

REST e linha de comandos

Para permitir que a plataforma de apps infira um esquema de tabela, use o campo createDefaultTableIfNotExists de BigQueryConfig quando criar ou atualizar uma app.

Crie e especifique uma tabela manualmente

Se quiser gerir a tabela de saída manualmente, a tabela tem de ter o esquema necessário como um subconjunto do esquema da tabela.

Se a tabela existente tiver esquemas incompatíveis, a implementação é rejeitada.

Usar esquema predefinido

Se usar o esquema predefinido para tabelas de resultados do modelo, certifique-se de que a tabela contém apenas as seguintes colunas obrigatórias. Pode copiar diretamente o texto do esquema seguinte quando criar a tabela do BigQuery. Para obter informações mais detalhadas acerca da criação de uma tabela do BigQuery, consulte Crie e use tabelas. Para mais informações sobre a especificação do esquema quando cria uma tabela, consulte o artigo Especificar um esquema.

Use o texto seguinte para descrever o esquema quando criar uma tabela. Para ver informações sobre a utilização do tipo de coluna JSON ("type": "JSON"), consulte o artigo Trabalhar com dados JSON no SQL padrão. O tipo de coluna JSON é recomendado para a consulta de anotações. Também pode usar "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"
 }
]

Aviso: a coluna JSON do BigQuery pode não estar disponível na consola. Se não conseguir criar uma coluna JSON na consola, use o comando gcloud abaixo ou substitua a coluna "JSON" por uma coluna "STRING".

Google Cloud consola

1. Na Google Cloud consola, aceda à página BigQuery.

   Aceda ao BigQuery

2. Selecione o seu projeto.

3. Selecione mais opções .

4. Clique em Criar tabela.

5. Na secção "Esquema", ative a opção Editar como texto.

gcloud

O exemplo seguinte cria primeiro o ficheiro JSON de pedido e, em seguida, usa o comando gcloud alpha bq tables create.

1. Primeiro, crie o ficheiro JSON de pedido:

   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. Envie o comando gcloud. Faça as seguintes substituições:

   • TABLE_NAME: o ID da tabela ou o identificador totalmente qualificado da tabela.

   • DATASET: o ID do conjunto de dados do BigQuery.

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

Exemplo de linhas do BigQuery geradas por uma app Vertex AI Vision:

ingestion_time	aplicação	instância	nó	anotação
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"]}

Use um esquema personalizado

Se o esquema predefinido não funcionar para o seu exemplo de utilização, pode usar funções do Cloud Run para gerar linhas do BigQuery com um esquema definido pelo utilizador. Se usar um esquema personalizado, não existe nenhum pré-requisito para o esquema da tabela do BigQuery.

Gráfico de apps com o nó do BigQuery selecionado

O conetor do BigQuery pode ser associado a qualquer modelo que produza anotações baseadas em vídeo ou proto:

• Para a entrada de vídeo, o conetor do BigQuery só extrai os dados de metadados armazenados no cabeçalho da stream e carrega estes dados para o BigQuery como outras saídas de anotações de modelos. O vídeo em si não é armazenado.
• Se a sua stream não contiver metadados, nada é armazenado no BigQuery.

Consultar dados de tabelas

Com o esquema de tabela predefinido do BigQuery, pode realizar uma análise poderosa depois de a tabela ser preenchida com dados.

Consultas de exemplo

Pode usar as seguintes consultas de exemplo no BigQuery para obter informações a partir dos modelos do Vertex AI Vision.

Por exemplo, pode usar o BigQuery para desenhar uma curva baseada no tempo para o número máximo de pessoas detetadas por minuto usando dados do modelo de detetor de pessoas / veículos com a seguinte consulta:

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

Da mesma forma, pode usar o BigQuery e a funcionalidade de contagem de linhas de travessia do modelo de estatísticas de ocupação para criar uma consulta que conte o número total de veículos que passam a linha de travessia por 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

Execute a consulta

Depois de formatar a consulta SQL padrão da Google, pode usar a consola para executar a consulta:

Consola

1. Na Google Cloud consola, abra a página do BigQuery.

   Aceda ao BigQuery

2. Selecione Expandir junto ao nome do conjunto de dados e selecione o nome da tabela.

3. Na vista de detalhes da tabela, clique em Compor nova consulta.

   

4. Introduza uma consulta Google Standard SQL na área de texto do editor de consultas. Para ver consultas de exemplo, consulte as consultas de exemplo.

5. Opcional: para alterar a localização do tratamento de dados, clique em Mais e, de seguida, em Definições de consulta. Em Localização de processamento, clique em Selecionar automaticamente e escolha a localização dos seus dados. Por último, clique em Guardar para atualizar as definições de consulta.

6. Clique em Executar.

Esta ação cria uma tarefa de consulta que escreve o resultado numa tabela temporária.

Integração de funções do Cloud Run

Pode usar funções do Cloud Run que acionam o processamento de dados adicional com o carregamento do BigQuery personalizado. Para usar funções do Cloud Run para o carregamento personalizado do BigQuery, faça o seguinte:

• Quando usar a Google Cloud consola, selecione a função na nuvem correspondente no menu pendente de cada modelo associado.

  

• Quando usar a Vertex AI Vision API, adicione um par de chave-valor ao campo cloud_function_mapping de BigQueryConfig no nó do BigQuery. A chave é o nome do nó do BigQuery e o valor é o acionador HTTP da função de destino.

Para usar funções do Cloud Run com a ingestão personalizada do BigQuery, a função tem de cumprir os seguintes requisitos:

• A instância de funções do Cloud Run tem de ser criada antes de criar o nó do BigQuery.
• A Vertex AI Vision API espera receber uma anotação AppendRowsRequest devolvida pelas funções do Cloud Run.
• Tem de definir o campo proto_rows.writer_schema para todas as respostas CloudFunction; write_stream pode ser ignorado.

Exemplo de integração de funções do Cloud Run

O exemplo seguinte mostra como analisar a saída do nó de contagem de ocupação (OccupancyCountPredictionResult) e extrair dela um esquema da tabela ingestion_time, person_count e vehicle_count.

O resultado do exemplo seguinte é uma tabela do BigQuery com o esquema:

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

Use o seguinte código para criar esta tabela:

1. Defina um proto (por exemplo, test_table_schema.proto) para os campos da tabela que quer escrever:

   syntax = "proto3";
   
   package visionai.testing;
   
   message TestTableSchema {
     int64 ingestion_time = 1;
     int32 person_count = 2;
     int32 vehicle_count = 3;
   }
   

2. Compile o ficheiro proto para gerar o ficheiro Python de buffers de protocolo:

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

3. Importe o ficheiro Python gerado e escreva a função da nuvem.

   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. Para incluir as suas dependências nas funções do Cloud Run, também tem de carregar o ficheiro test_table_schema_pb2.py gerado e especificar requirements.txt semelhante ao seguinte:

   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. Implemente a função do Cloud e defina o acionador HTTP correspondente no ficheiro BigQueryConfig.