Monitorar consultas contínuas

É possível monitorar consultas contínuas do BigQuery usando as seguintes ferramentas do BigQuery:

Devido à natureza de longa duração das consultas contínuas do BigQuery, métricas que geralmente são geradas após a conclusão de uma consulta SQL podem estar ausentes ou imprecisas.

Usar visualizações INFORMATION_SCHEMA

É possível usar várias visualizações INFORMATION_SCHEMA para monitorar consultas contínuas e reservas de consulta contínuas.

Ver detalhes do job

Você pode usar a JOBS visualização para receber metadados dos jobs de consulta contínuos.

A consulta a seguir retorna os metadados de todas as consultas contínuas ativas. Os metadados incluem o carimbo de data/hora da marca-d'água de saída, que representa o ponto até o qual a consulta contínua processou dados.

  1. No Cloud de Confiance console, acesse a página BigQuery.

    Acessar o BigQuery

  2. No Editor de consultas, execute esta consulta:

    SELECT
  start_time,
  job_id,
  user_email,
  query,
  state,
  reservation_id,
  continuous_query_info.output_watermark
FROM `PROJECT_ID.region-REGION.INFORMATION_SCHEMA.JOBS`
WHERE
  creation_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 day)
  AND continuous IS TRUE
  AND state = "RUNNING"
ORDER BY
  start_time DESC

    Substitua:

Ver detalhes da atribuição de reserva

Você pode usar as ASSIGNMENTS e RESERVATIONS visualizações para receber detalhes da atribuição de reserva de consulta contínua.

Retorne detalhes de atribuição de reserva para consultas contínuas:

  1. No Cloud de Confiance console, acesse a página BigQuery.

    Acessar o BigQuery

  2. No Editor de consultas, execute esta consulta:

    SELECT
  reservation.reservation_name,
  reservation.slot_capacity
FROM
  `ADMIN_PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.ASSIGNMENTS`
    AS assignment
INNER JOIN
  `ADMIN_PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.RESERVATIONS`
    AS reservation
  ON (assignment.reservation_name = reservation.reservation_name)
WHERE
  assignment.assignee_id = 'PROJECT_ID'
  AND job_type = 'CONTINUOUS';

    Substitua:

    • ADMIN_PROJECT_ID: o ID do projeto de administração que contém a reserva.
    • LOCATION: o local da reserva.
    • PROJECT_ID: o ID do projeto atribuído à reserva. Somente informações sobre consultas contínuas em execução neste projeto são retornadas.

Ver informações de consumo de slots

Você pode usar as ASSIGNMENTS, RESERVATIONS e JOBS_TIMELINE visualizações para receber informações de consumo do slot de consulta contínua.

Retorna informações de consumo de slots para consultas contínuas:

  1. No Cloud de Confiance console, acesse a página BigQuery.

    Acessar o BigQuery

  2. No Editor de consultas, execute esta consulta:

    SELECT
  jobs.period_start,
  reservation.reservation_name,
  reservation.slot_capacity,
  SUM(jobs.period_slot_ms) / 1000 AS consumed_total_slots
FROM
  `ADMIN_PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.ASSIGNMENTS`
    AS assignment
INNER JOIN
  `ADMIN_PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.RESERVATIONS`
    AS reservation
  ON (assignment.reservation_name = reservation.reservation_name)
INNER JOIN
  `PROJECT_ID.region-LOCATION.INFORMATION_SCHEMA.JOBS_TIMELINE` AS jobs
  ON (
    UPPER(CONCAT('ADMIN_PROJECT_ID:LOCATION.', assignment.reservation_name))
    = UPPER(jobs.reservation_id))
WHERE
  assignment.assignee_id = 'PROJECT_ID'
  AND assignment.job_type = 'CONTINUOUS'
  AND jobs.period_start
    BETWEEN TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
    AND CURRENT_TIMESTAMP()
GROUP BY 1, 2, 3
ORDER BY jobs.period_start DESC;

    Substitua:

    • ADMIN_PROJECT_ID: o ID do projeto de administração que contém a reserva.
    • LOCATION: o local da reserva.
    • PROJECT_ID: o ID do projeto atribuído à reserva. Somente informações sobre consultas contínuas em execução neste projeto são retornadas.

Também é possível monitorar reservas de consulta contínua usando outras ferramentas, como Metrics Explorer e gráficos de recursos administrativos. Para mais informações, consulte Monitore as reservas do BigQuery.

Usar o gráfico de execução de consulta

Você pode usar o gráfico de execução de consulta para receber insights de desempenho e estatísticas gerais para uma consulta contínua. Para mais informações, consulte Visualizar insights de desempenho da consulta.

Visualizar o histórico de tarefas

Você pode conferir os detalhes do job de consulta contínua no seu histórico profissional pessoal ou no histórico de job do projeto. Para mais informações, consulte Ver detalhes do job.

A lista histórica de jobs é classificada pela hora de início do job. Isso significa que as consultas contínuas que estão em execução há um tempo pode não estar perto do início da lista.

Usar o explorador de jobs

No explorador de jobs, filtre seus jobs para mostrar consultas contínuas definindo o filtro Categoria de job como Consulta contínua.

Usar o Cloud Monitoring

É possível conferir métricas específicas para consultas contínuas do BigQuery usando o Cloud Monitoring. Para mais informações, consulte Criar painéis, gráficos e alertas e leia sobre as métricas disponíveis para visualização.

Alertar sobre consultas com falha

Em vez de verificar rotineiramente se as consultas contínuas falharam, pode ser útil criar um alerta para notificar você sobre a falha. Uma maneira de fazer isso é criar uma métrica personalizada com base em registros do Cloud Logging com um filtro para seus jobs e uma política de alertas do Cloud Monitoring com base nessa métrica:

  1. Ao criar uma consulta contínua, use um prefixo de ID de job personalizado. Várias consultas contínuas podem compartilhar o mesmo prefixo. Por exemplo, você pode usar o prefixo prod- para indicar uma consulta de produção.

  2. No Cloud de Confiance console, acesse a página Métricas com base em registros.

    Acessar "Métricas com base em registros"

  3. Clique em Criar métrica. O painel Criar métrica de registros é exibido.

  4. Em Tipo de métrica, selecione Contador.

  5. Na seção Detalhes, atribua um nome à métrica. Por exemplo, CUSTOM_JOB_ID_PREFIX-metric.

  6. Na seção Seleção de filtro, insira o seguinte no editor Criar filtro:

    resource.type = "bigquery_project"
protoPayload.resourceName : "projects/PROJECT_ID/jobs/CUSTOM_JOB_ID_PREFIX"
severity = ERROR

    Substitua:

  7. Clique em Criar métrica.

  8. No menu de navegação, clique em Métricas com base em registros. A métrica que você acabou de criar aparece na lista de métricas definidas pelo usuário.

  9. Na linha da métrica, clique em Mais ações e em Criar alerta com base na métrica.

  10. Clique em Próxima. Não é necessário mudar as configurações padrão na página Modo de configuração da política.

  11. Clique em Próxima. Não é necessário mudar as configurações padrão na página Configurar acionador de alerta.

  12. Selecione os canais de notificação e insira um nome para a política de alertas.

  13. Clique em Criar política.

Para testar o alerta, execute uma consulta contínua com o prefixo de ID de job personalizado selecionado e cancele-o. Pode levar alguns minutos para que o alerta chegue ao seu canal de notificação.

Tentar novamente consultas com falha

Tentar novamente uma consulta contínua com falha pode ajudar a evitar situações em que um pipeline contínuo fica inativo por um período prolongado ou exige intervenção humana para ser reiniciado. As considerações importantes ao tentar novamente uma consulta contínua com falha incluem o seguinte:

  • Se o reprocessamento de alguma quantidade de dados processados pela consulta anterior antes da falha é tolerável.
  • Como limitar as tentativas ou usar a espera exponencial.

Uma possível abordagem para automatizar a repetição de consultas é a seguinte:

  1. Crie um coletor do Cloud Logging com base em um filtro de inclusão que corresponda aos seguintes critérios para encaminhar registros a um tópico do Pub/Sub:

    resource.type = "bigquery_project"
protoPayload.resourceName : "projects/PROJECT_ID/jobs/CUSTOM_JOB_ID_PREFIX"
severity = ERROR

    Substitua:

  2. Crie uma função do Cloud Run que seja acionada em resposta ao Pub/Sub que recebe registros que correspondem ao filtro.

    A função do Cloud Run pode aceitar o payload de dados da mensagem do Pub/Sub e tentar iniciar uma nova consulta contínua usando a mesma sintaxe SQL da consulta com falha, mas no início, logo após a interrupção do job anterior.

Por exemplo, você pode usar uma função semelhante à seguinte:

Python

Antes de testar esta amostra, siga as instruções de configuração do Python no Guia de início rápido do BigQuery: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API BigQuery em Python.

Para autenticar no BigQuery, configure o Application Default Credentials. Para mais informações, acesse Configurar a autenticação para bibliotecas de cliente.

Antes de executar exemplos de código, defina a variável de ambiente GOOGLE_CLOUD_UNIVERSE_DOMAIN como s3nsapis.fr. 

import base64
import json
import logging
import re
import uuid

import google.auth
import google.auth.transport.requests
import requests


def retry_continuous_query(event, context):
    logging.info("Cloud Function started.")

    if "data" not in event:
        logging.info("No data in Pub/Sub message.")
        return

    try:
        # Decode and parse the Pub/Sub message data
        log_entry = json.loads(base64.b64decode(event["data"]).decode("utf-8"))

        # Extract the SQL query and other necessary data
        proto_payload = log_entry.get("protoPayload", {})
        metadata = proto_payload.get("metadata", {})
        job_change = metadata.get("jobChange", {})
        job = job_change.get("job", {})
        job_config = job.get("jobConfig", {})
        query_config = job_config.get("queryConfig", {})
        sql_query = query_config.get("query")
        job_stats = job.get("jobStats", {})
        end_timestamp = job_stats.get("endTime")
        failed_job_id = job.get("jobName")

        # Check if required fields are missing
        if not all([sql_query, failed_job_id, end_timestamp]):
            logging.error("Required fields missing from log entry.")
            return

        logging.info(f"Retrying failed job: {failed_job_id}")

        # Adjust the timestamp in the SQL query
        timestamp_match = re.search(
            r"\s*TIMESTAMP\(('.*?')\)(\s*\+ INTERVAL 1 MICROSECOND)?", sql_query
        )

        if timestamp_match:
            original_timestamp = timestamp_match.group(1)
            new_timestamp = f"'{end_timestamp}'"
            sql_query = sql_query.replace(original_timestamp, new_timestamp)
        elif "CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE" in sql_query:
            new_timestamp = f"TIMESTAMP('{end_timestamp}') + INTERVAL 1 MICROSECOND"
            sql_query = sql_query.replace(
                "CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE", new_timestamp
            )

        # Get access token
        credentials, project = google.auth.default(
            scopes=["https://www.googleapis.com/auth/cloud-platform"]
        )
        request = google.auth.transport.requests.Request()
        credentials.refresh(request)
        access_token = credentials.token

        # API endpoint
        url = f"https://bigquery.googleapis.com/bigquery/v2/projects/{project}/jobs"

        # Request headers
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json",
        }

        # Generate a random UUID
        random_suffix = str(uuid.uuid4())[:8]  # Take the first 8 characters of the UUID

        # Combine the prefix and random suffix
        job_id = f"CUSTOM_JOB_ID_PREFIX{random_suffix}"

        # Request payload
        data = {
            "configuration": {
                "query": {
                    "query": sql_query,
                    "useLegacySql": False,
                    "continuous": True,
                    "connectionProperties": [
                        {"key": "service_account", "value": "SERVICE_ACCOUNT"}
                    ],
                    # ... other query parameters ...
                },
                "labels": {"bqux_job_id_prefix": "CUSTOM_JOB_ID_PREFIX"},
            },
            "jobReference": {
                "projectId": project,
                "jobId": job_id,  # Use the generated job ID here
            },
        }

        # Make the API request
        response = requests.post(url, headers=headers, json=data)

        # Handle the response
        if response.status_code == 200:
            logging.info("Query job successfully created.")
        else:
            logging.error(f"Error creating query job: {response.text}")

    except Exception as e:
        logging.error(
            f"Error processing log entry or retrying query: {e}", exc_info=True
        )

    logging.info("Cloud Function finished.")

A seguir