Crie consultas contínuas

Este documento descreve como executar uma consulta contínua no BigQuery.

As consultas contínuas do BigQuery são declarações SQL executadas continuamente. As consultas contínuas permitem-lhe analisar os dados recebidos no BigQuery em tempo real e, em seguida, exportar os resultados para o Bigtable, o Pub/Sub ou o Spanner, ou escrever os resultados numa tabela do BigQuery.

Escolha um tipo de conta

Pode criar e executar uma tarefa de consulta contínua através de uma conta de utilizador ou pode criar uma tarefa de consulta contínua através de uma conta de utilizador e, em seguida, executá-la através de uma conta de serviço. Tem de usar uma conta de serviço para executar uma consulta contínua que exporte resultados para um tópico do Pub/Sub.

Quando usa uma conta de utilizador, é executada uma consulta contínua durante um máximo de dois dias. Quando usa uma conta de serviço, é executada uma consulta contínua durante um máximo de 150 dias. Para mais informações, consulte o artigo Autorização.

Autorizações necessárias

Esta secção descreve as autorizações de que precisa para criar e executar uma consulta contínua. Em alternativa às funções de gestão de identidade e de acesso (IAM) mencionadas, pode obter as autorizações necessárias através de funções personalizadas.

Autorizações quando usa uma conta de utilizador

Esta secção fornece informações sobre as funções e as autorizações necessárias para criar e executar uma consulta contínua através de uma conta de utilizador.

Para criar uma tarefa no BigQuery, a conta de utilizador tem de ter a autorização de IAM bigquery.jobs.create. Cada uma das seguintes funções de IAM concede a autorização bigquery.jobs.create:

• Utilizador do BigQuery (roles/bigquery.user)
• Utilizador de tarefas do BigQuery (roles/bigquery.jobUser)
• Administrador do BigQuery (roles/bigquery.admin)

Para exportar dados de uma tabela do BigQuery, a conta de utilizador tem de ter a autorização de IAM bigquery.tables.export . Cada uma das seguintes funções de IAM concede a autorização bigquery.tables.export:

• Visualizador de dados do BigQuery (roles/bigquery.dataViewer)
• Editor de dados do BigQuery (roles/bigquery.dataEditor)
• Proprietário dos dados do BigQuery (roles/bigquery.dataOwner)
• Administrador do BigQuery (roles/bigquery.admin)

Para atualizar dados numa tabela do BigQuery, a conta de utilizador tem de ter a autorização de IAM bigquery.tables.updateData. Cada uma das seguintes funções de IAM concede a autorização bigquery.tables.updateData:

• Editor de dados do BigQuery (roles/bigquery.dataEditor)
• Proprietário dos dados do BigQuery (roles/bigquery.dataOwner)
• Administrador do BigQuery (roles/bigquery.admin)

Se a conta de utilizador tiver de ativar as APIs necessárias para o seu exemplo de utilização de consulta contínua, a conta de utilizador tem de ter a função Administrador de utilização de serviços (roles/serviceusage.serviceUsageAdmin).

Autorizações ao usar uma conta de serviço

Esta secção fornece informações sobre as funções e as autorizações necessárias para a conta de utilizador que cria a consulta contínua e a conta de serviço que executa a consulta contínua.

Autorizações da conta de utilizador

Para criar uma tarefa no BigQuery, a conta de utilizador tem de ter a autorização de IAM bigquery.jobs.create. Cada uma das seguintes funções do IAM concede a autorização bigquery.jobs.create:

• Utilizador do BigQuery (roles/bigquery.user)
• Utilizador de tarefas do BigQuery (roles/bigquery.jobUser)
• Administrador do BigQuery (roles/bigquery.admin)

Para enviar uma tarefa que é executada através de uma conta de serviço, a conta de utilizador tem de ter a função Utilizador da conta de serviço (roles/iam.serviceAccountUser). Se estiver a usar a mesma conta de utilizador para criar a conta de serviço, a conta de utilizador tem de ter a função Administrador da conta de serviço (roles/iam.serviceAccountAdmin). Para obter informações sobre como limitar o acesso de um utilizador a uma única conta de serviço, em vez de a todas as contas de serviço num projeto, consulte o artigo Conceda uma única função.

Se a conta de utilizador tiver de ativar as APIs necessárias para o seu exemplo de utilização de consulta contínua, a conta de utilizador tem de ter a função Administrador de utilização de serviços (roles/serviceusage.serviceUsageAdmin).

Autorizações da conta de serviço

Para exportar dados de uma tabela do BigQuery, a conta de serviço tem de ter a autorização de IAM bigquery.tables.export. Cada uma das seguintes funções de IAM concede a autorização bigquery.tables.export:

• Visualizador de dados do BigQuery (roles/bigquery.dataViewer)
• Editor de dados do BigQuery (roles/bigquery.dataEditor)
• Proprietário dos dados do BigQuery (roles/bigquery.dataOwner)
• Administrador do BigQuery (roles/bigquery.admin)

• Editor de dados do BigQuery (roles/bigquery.dataEditor)
• Proprietário dos dados do BigQuery (roles/bigquery.dataOwner)
• Administrador do BigQuery (roles/bigquery.admin)

Antes de começar

1. In the Trusted Cloud console, on the project selector page, select or create a Trusted Cloud project.

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

   Go to project selector

2. Verify that billing is enabled for your Trusted Cloud project.

3. Enable the BigQuery API.

   Roles required to enable APIs

   To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

   Enable the API

Crie uma reserva

Crie uma reserva da edição Enterprise ou Enterprise Plus e, de seguida, crie uma atribuição de reserva com um CONTINUOUS tipo de tarefa. Esta reserva pode usar o dimensionamento automático. Existem limitações de reservas que se aplicam a atribuições de reservas para consultas contínuas.

Exporte para o Pub/Sub

São necessários APIs, autorizações da IAM e Trusted Cloud by S3NS recursos adicionais para exportar dados para o Pub/Sub. Para mais informações, consulte o artigo Exporte para o Pub/Sub.

Incorporar atributos personalizados como metadados em mensagens do Pub/Sub

Pode usar atributos do Pub/Sub para fornecer informações adicionais sobre a mensagem, como a respetiva prioridade, origem, destino ou metadados adicionais. Também pode usar atributos para filtrar mensagens na subscrição.

Num resultado de consulta contínua, se uma coluna for denominada _ATTRIBUTES, os respetivos valores são copiados para os atributos de mensagens do Pub/Sub. Os campos fornecidos em _ATTRIBUTES são usados como chaves de atributos.

A coluna _ATTRIBUTES tem de ser do tipo JSON, no formato ARRAY<STRUCT<STRING, STRING>> ou STRUCT<STRING>.

Para ver um exemplo, consulte o artigo sobre como exportar dados para um tópico do Pub/Sub.

Exporte para o Bigtable

São necessárias APIs, autorizações da IAM e Trusted Cloud recursos adicionais para exportar dados para o Bigtable. Para mais informações, consulte Exportar para o Bigtable.

Exporte para o Spanner

São necessárias APIs, autorizações da IAM e Trusted Cloud recursos adicionais para exportar dados para o Spanner. Para mais informações, consulte Exporte para o Spanner (ETL inverso).

Escreva dados numa tabela do BigQuery

Pode escrever dados numa tabela do BigQuery através de uma declaração INSERT.

Use funções de IA

São necessárias APIs, autorizações de IAM e Trusted Cloud recursos adicionais para usar uma função de IA suportada numa consulta contínua. Para mais informações, consulte um dos seguintes tópicos, com base no seu exemplo de utilização:

• Gere texto com a função ML.GENERATE_TEXT
• Gere incorporações de texto através da função ML.GENERATE_EMBEDDING
• Compreenda o texto com a função ML.UNDERSTAND_TEXT
• Traduza texto com a função ML.TRANSLATE

Quando usa uma função de IA numa consulta contínua, considere se o resultado da consulta vai permanecer dentro da quota da função. Se exceder a quota, pode ter de processar separadamente os registos que não são processados.

Especifique um ponto de partida

Tem de usar a função APPENDS na cláusula FROM de uma consulta contínua para especificar os dados mais antigos a processar. Por exemplo, APPENDS(TABLE my_table, CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)diz ao BigQuery para processar dados que foram adicionados à tabela my_table, no máximo, 10 minutos antes do início da consulta contínua. Os dados adicionados ao my_table são processados à medida que chegam. Não existe um atraso imposto no processamento de dados. Não forneça um argumento end_timestamp à função APPENDS quando a usar numa consulta contínua.

O exemplo seguinte mostra como iniciar uma consulta contínua a partir de um ponto específico no tempo usando a função APPENDS, quando consulta uma tabela do BigQuery que está a receber informações de viagens de táxi em streaming:

EXPORT DATA
  OPTIONS (format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides') AS (
  SELECT
    TO_JSON_STRING(STRUCT(ride_id,
        timestamp,
        latitude,
        longitude)) AS message
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE
    ride_status = 'enroute');

Especifique um ponto de partida anterior à hora atual

Se quiser processar dados de antes do momento atual, pode usar a função APPENDS para especificar um ponto de partida anterior para a consulta. O ponto de partida que especificar tem de estar dentro do período de viagem no tempo para a tabela a partir da qual está a selecionar. Por predefinição, o período de deslocação no tempo abrange os últimos sete dias.

Para incluir dados que estejam fora do período de viagem no tempo, use uma consulta padrão para inserir ou exportar dados até um determinado momento e, em seguida, inicie uma consulta contínua a partir desse momento.

Exemplo

O exemplo seguinte mostra como carregar dados mais antigos de uma tabela do BigQuery que está a receber informações de viagens de táxi em streaming até um determinado momento para uma tabela e, em seguida, iniciar uma consulta contínua a partir do ponto de corte dos dados mais antigos.

1. Execute uma consulta padrão para preencher dados até um determinado momento:

   INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
   SELECT
     timestamp,
     meter_reading,
     ride_status,
     passenger_count,
     ST_Distance(
       ST_GeogPoint(pickup_longitude, pickup_latitude),
       ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
       SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
   FROM `myproject.real_time_taxi_streaming.taxirides`
     -- Include all data inserted into the table up to this point in time.
     -- This timestamp must be within the time travel window.
     FOR SYSTEM_TIME AS OF '2025-01-01 00:00:00 UTC'
   WHERE
     ride_status = 'dropoff';

2. Execute uma consulta contínua a partir do momento em que a consulta parou:

   INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
   SELECT
     timestamp,
     meter_reading,
     ride_status,
     passenger_count,
     ST_Distance(
       ST_GeogPoint(pickup_longitude, pickup_latitude),
       ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
       SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
   FROM
     APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
       -- Configure the APPENDS TVF start_timestamp to start processing
       -- data right where the batch query left off.
       -- This timestamp must be within the time travel window.
       TIMESTAMP '2025-01-01 00:00:00 UTC' + INTERVAL 1 MICROSECOND)
   WHERE
     ride_status = 'dropoff';

Execute uma consulta contínua através de uma conta de utilizador

Esta secção descreve como executar uma consulta contínua através de uma conta de utilizador. Depois de a consulta contínua estar em execução, pode fechar a Trusted Cloud consola, a janela do terminal ou a aplicação sem interromper a execução da consulta. Uma consulta contínua executada por uma conta de utilizador é executada durante um máximo de dois dias e, em seguida, é interrompida automaticamente. Para continuar a processar novos dados recebidos, inicie uma nova consulta contínua e especifique um ponto de partida. Para automatizar este processo, consulte o artigo sobre como repetir consultas com falhas.

Siga estes passos para executar uma consulta contínua:

Execute uma consulta contínua através de uma conta de serviço

Esta secção descreve como executar uma consulta contínua através de uma conta de serviço. Depois de a consulta contínua estar em execução, pode fechar a Trusted Cloud consola, a janela do terminal ou a aplicação sem interromper a execução da consulta. Uma consulta contínua executada através de uma conta de serviço pode ser executada durante um máximo de 150 dias e, em seguida, é interrompida automaticamente. Para continuar a processar novos dados recebidos, inicie uma nova consulta contínua e especifique um ponto de partida. Para automatizar este processo, consulte o artigo sobre como repetir consultas com falhas.

Siga estes passos para usar uma conta de serviço para executar uma consulta contínua:

Crie um ID de tarefa personalizado

A cada tarefa de consulta é atribuído um ID da tarefa que pode usar para pesquisar e gerir a tarefa. Por predefinição, os IDs das tarefas são gerados aleatoriamente. Para facilitar a pesquisa do ID da tarefa de uma consulta contínua através do histórico de tarefas ou do explorador de tarefas, pode atribuir um prefixo do ID da tarefa personalizado:

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

   Aceda ao BigQuery

2. No editor de consultas, clique em Mais.

3. Na secção Escolher modo de consulta, escolha Consulta contínua.

4. Clique em Confirm.

5. No editor de consultas, clique em Mais > Definições da consulta.

6. Na secção Prefixo do ID da tarefa personalizado, introduza um prefixo do nome personalizado.

7. Clique em Guardar.

Exemplos

Os exemplos de SQL seguintes mostram exemplos de utilização comuns para consultas contínuas.

Exporte dados para um tópico do Pub/Sub

O exemplo seguinte mostra uma consulta contínua que filtra dados de uma tabela do BigQuery que está a receber informações de viagens de táxi em streaming e publica os dados num tópico do Pub/Sub em tempo real com atributos de mensagens:

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides')
AS (
  SELECT
    TO_JSON_STRING(
      STRUCT(
        ride_id,
        timestamp,
        latitude,
        longitude)) AS message,
    TO_JSON(
      STRUCT(
        CAST(passenger_count AS STRING) AS passenger_count)) AS _ATTRIBUTES
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxi_rides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
);

Exporte dados para uma tabela do Bigtable

O exemplo seguinte mostra uma consulta contínua que filtra dados de uma tabela do BigQuery que está a receber informações de viagens de táxi em streaming e exporta os dados para uma tabela do Bigtable em tempo real:

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_BIGTABLE',
    truncate = TRUE,
    overwrite = TRUE,
    uri = 'https://bigtable.googleapis.com/projects/myproject/instances/mybigtableinstance/tables/taxi-real-time-rides')
AS (
  SELECT
    CAST(CONCAT(ride_id, timestamp, latitude, longitude) AS STRING) AS rowkey,
    STRUCT(
      timestamp,
      latitude,
      longitude,
      meter_reading,
      ride_status,
      passenger_count) AS features
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
);

Exporte dados para uma tabela do Spanner

O exemplo seguinte mostra uma consulta contínua que filtra dados de uma tabela do BigQuery que está a receber informações de viagens de táxi com streaming e, em seguida, exporta os dados para uma tabela do Spanner em tempo real:

EXPORT DATA
 OPTIONS (
   format = 'CLOUD_SPANNER',
   uri = 'https://spanner.googleapis.com/projects/myproject/instances/myspannerinstance/databases/taxi-real-time-rides',
   spanner_options ="""{
      "table": "rides",
      -- To ensure data is written to Spanner in the correct sequence
      -- during a continuous export, use the change_timestamp_column
      -- option. This should be mapped to a timestamp column from your
      -- BigQuery data. If your source data lacks a timestamp, the 
      -- _CHANGE_TIMESTAMP pseudocolumn provided by the APPENDS function 
      -- will be automatically mapped to the "change_timestamp" column.
      "change_timestamp_column": "change_timestamp"
   }"""
  )
  AS (
  SELECT
    ride_id,
    latitude,
    longitude,
    meter_reading,
    ride_status,
    passenger_count
  FROM APPENDS(
        TABLE `myproject.real_time_taxi_streaming.taxirides`,
        -- Configure the APPENDS TVF start_timestamp to specify when you want to
        -- start processing data using your continuous query.
        -- This example starts processing at 10 minutes before the current time.
        CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
  );

Escreva dados numa tabela do BigQuery

O exemplo seguinte mostra uma consulta contínua que filtra e transforma dados de uma tabela do BigQuery que está a receber informações de viagens de táxi com streaming e, em seguida, escreve os dados noutra tabela do BigQuery em tempo real. Isto disponibiliza os dados para análise mais detalhada.

INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
SELECT
  timestamp,
  meter_reading,
  ride_status,
  passenger_count,
  ST_Distance(
    ST_GeogPoint(pickup_longitude, pickup_latitude),
    ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
    SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
FROM
  APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
    -- Configure the APPENDS TVF start_timestamp to specify when you want to
    -- start processing data using your continuous query.
    -- This example starts processing at 10 minutes before the current time.
    CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
WHERE
  ride_status = 'dropoff';

Trate dados usando um modelo da Vertex AI

O exemplo seguinte mostra uma consulta contínua que usa um modelo da Vertex AI para gerar um anúncio para passageiros de táxi com base na respetiva latitude e longitude atuais e, em seguida, exporta os resultados para um tópico do Pub/Sub em tempo real:

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides')
AS (
  SELECT
    TO_JSON_STRING(
      STRUCT(
        ride_id,
        timestamp,
        latitude,
        longitude,
        prompt,
        ml_generate_text_llm_result)) AS message
  FROM
    ML.GENERATE_TEXT(
      MODEL `myproject.real_time_taxi_streaming.taxi_ml_generate_model`,
      (
        SELECT
          timestamp,
          ride_id,
          latitude,
          longitude,
          CONCAT(
            'Generate an ad based on the current latitude of ',
            latitude,
            ' and longitude of ',
            longitude) AS prompt
        FROM
          APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
            -- Configure the APPENDS TVF start_timestamp to specify when you
            -- want to start processing data using your continuous query.
            -- This example starts processing at 10 minutes before the current time.
            CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
        WHERE ride_status = 'enroute'
      ),
      STRUCT(
        50 AS max_output_tokens,
        1.0 AS temperature,
        40 AS top_k,
        1.0 AS top_p,
        TRUE AS flatten_json_output))
      AS ml_output
);

Modifique o SQL de uma consulta contínua

Não pode atualizar o SQL usado numa consulta contínua enquanto a tarefa de consulta contínua estiver em execução. Tem de cancelar a tarefa de consulta contínua, modificar o SQL e, em seguida, iniciar uma nova tarefa de consulta contínua a partir do ponto em que parou a tarefa de consulta contínua original.

Siga estes passos para modificar o SQL usado numa consulta contínua:

1. Veja os detalhes da tarefa para a tarefa de consulta contínua que quer atualizar e tome nota do ID da tarefa.
2. Se possível, pause a recolha de dados a montante. Se não o conseguir fazer, pode ocorrer alguma duplicação de dados quando a consulta contínua for reiniciada.
3. Cancele a consulta contínua que quer modificar.

4. Obtenha o valor end_time para a tarefa de consulta contínua original através da vista JOBS:INFORMATION_SCHEMA

   SELECT end_time
   FROM `PROJECT_ID.region-REGION`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
   WHERE
     EXTRACT(DATE FROM creation_time) = current_date()
   AND error_result.reason = 'stopped'
   AND job_id = 'JOB_ID';

   Substitua o seguinte:

   • PROJECT_ID: o ID do seu projeto.
   • REGION: a região usada pelo seu projeto.
   • JOB_ID: o ID da tarefa de consulta contínua que identificou no passo 1.

5. Modifique a declaração SQL da consulta contínua para iniciar a consulta contínua a partir de um momento específico, usando o valor end_time que recuperou no passo 5 como ponto de partida.

6. Modifique a declaração SQL de consulta contínua para refletir as alterações necessárias.

7. Execute a consulta contínua modificada.

Cancele uma consulta contínua

Pode cancelar uma tarefa de consulta contínua tal como qualquer outra tarefa. A consulta pode demorar até um minuto a parar de ser executada após o cancelamento da tarefa.

Se cancelar e, em seguida, reiniciar uma consulta, a consulta reiniciada comporta-se como uma nova consulta independente. A consulta reiniciada não começa a processar dados no ponto em que a tarefa anterior parou e não pode fazer referência aos resultados da consulta anterior. Veja como iniciar uma consulta contínua a partir de um determinado momento.

Monitorize consultas e resolva erros

Uma consulta contínua pode ser interrompida devido a fatores como inconsistências de dados, alterações ao esquema, interrupções temporárias do serviço ou manutenção. Embora o BigQuery processe alguns erros transitórios, as práticas recomendadas para melhorar a capacidade de recuperação de tarefas incluem o seguinte:

• Monitorize consultas contínuas.
• Alerta sobre consultas com falhas.
• Tentar novamente consultas com falhas.

O que se segue?

• Monitorize consultas contínuas