Es posible que parte de la información de esta página (o toda) no se aplique a Cloud de Confiance de S3NS. Consulta Diferencias con Google Cloud para obtener más información.

Google utiliza tecnología de IA para traducir contenido a tu idioma preferido. Las traducciones realizadas con IA pueden contener errores.

Supervisa consultas continuas

Puedes supervisar las consultas continuas de BigQuery con las siguientes herramientas de BigQuery:

Debido a la naturaleza de larga duración de una consulta continua de BigQuery, las métricas que se suelen generar cuando se completa una consulta en SQL pueden estar ausentes o imprecisas.

Usa vistas `INFORMATION_SCHEMA`

Puedes usar varias vistas INFORMATION_SCHEMA para supervisar las consultas continuas y las reservas de consultas continuas.

Ver detalles del trabajo

Puedes usar la JOBS vista para obtener metadatos de trabajos de consulta continuas.

La siguiente consulta muestra los metadatos de todas las consultas continuas activas. Los metadatos incluyen la marca de tiempo de la marca de agua de resultado, que representa el punto hasta el cual la consulta continua procesó los datos de forma correcta.

En la Cloud de Confiance consola de, ve a la página BigQuery.

Ir a BigQuery

En el Editor de consultas, ejecute la siguiente 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

Reemplaza lo siguiente:

PROJECT_ID: el ID del proyecto.
REGION: cualquier nombre de región del conjunto de datos. Por ejemplo, region-us.

Visualiza los detalles de la asignación de reservas

Puedes usar las ASSIGNMENTS y las RESERVATIONS vistas para obtener detalles de asignación de reserva de consultas continuas.

Muestra los detalles de la asignación de reservas para las consultas continuas:

En la Cloud de Confiance consola de, ve a la página BigQuery.

Ir a BigQuery

En el Editor de consultas, ejecute la siguiente 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';

Reemplaza lo siguiente:

ADMIN_PROJECT_ID por el ID del proyecto de administración que posee la reserva.
LOCATION: la ubicación de la reserva.
PROJECT_ID: el ID del proyecto que se asigna a la reserva. Solo se muestra información sobre las consultas continuas que se ejecutan en este proyecto.

Visualiza información sobre el consumo de ranuras

Puedes usar las ASSIGNMENTS, RESERVATIONS y JOBS_TIMELINE vistas para obtener información del consumo de ranuras de consulta continua.

Muestra información sobre el consumo de ranuras de consultas continuas:

En la Cloud de Confiance consola de, ve a la página BigQuery.

Ir a BigQuery

En el Editor de consultas, ejecute la siguiente 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;

Reemplaza lo siguiente:

ADMIN_PROJECT_ID por el ID del proyecto de administración que posee la reserva.
LOCATION: la ubicación de la reserva.
PROJECT_ID: el ID del proyecto que se asigna a la reserva. Solo se muestra información sobre las consultas continuas que se ejecutan en este proyecto.

También puedes supervisar las reservas de consultas continuas con otras herramientas, como el Explorador de métricas y los gráficos de recursos administrativos. Para obtener más información, consulta Supervisa reservas de BigQuery.

Usa el gráfico de ejecución de consultas

Puedes usar el gráfico de ejecución de consultas para obtener estadísticas de rendimiento y estadísticas generales para una consulta continua. Para obtener más información, consulta Visualiza las estadísticas de rendimiento de las consultas.

Ver el historial de un trabajo

Puedes ver los detalles del trabajo de consulta continua en tu historial de trabajos personales o en el historial de trabajos del proyecto. Para obtener más información, consulta Visualiza los detalles del trabajo.

Ten en cuenta que la lista histórica de trabajos se ordena según la hora de inicio del trabajo, por lo que es posible que las consultas continuas que se han ejecutado por un tiempo no estén cerca del inicio de la lista.

Usa el explorador de trabajos

En el explorador de trabajos, filtra tus trabajos para mostrar consultas continuas. Para ello, configura el filtro Categoría del trabajo en Consulta continua.

Use Cloud Monitoring

Puedes ver métricas específicas de las consultas continuas de BigQuery con Cloud Monitoring. Para obtener más información, consulta Crea paneles, gráficos y alertas y lee sobre las métricas disponibles para la visualización.

Recibe alertas sobre consultas con errores

En lugar de verificar de forma rutinaria si tus consultas continuas fallaron, puede ser útil crear una alerta para notificarte sobre la falla. Una forma de hacerlo es crear una métrica personalizada basada en registros de Cloud Logging con un filtro para tus trabajos y una política de alertas de Cloud Monitoring basada en esa métrica:

Cuando crees una consulta continua, usa un prefijo de ID de trabajo personalizado. Varias consultas continuas pueden compartir el mismo prefijo. Por ejemplo, puedes usar el prefijo prod- para indicar una consulta de producción.
En la Cloud de Confiance consola de, ve a la página Métricas basadas en registros.

Ve a Métricas basadas en registros
Haz clic en Crear métrica. Aparecerá el panel Crear métrica de registros.
En Tipo de métrica, selecciona Contador.
En la sección Detalles, asígnale un nombre a tu métrica. Por ejemplo, CUSTOM_JOB_ID_PREFIX-metric.
En la sección Selección de filtros, ingresa lo siguiente en el editor Compilar filtro:
```
resource.type = "bigquery_project"
protoPayload.resourceName : "projects/PROJECT_ID/jobs/CUSTOM_JOB_ID_PREFIX"
severity = ERROR
```
Reemplaza lo siguiente:
- PROJECT_ID: nombre del proyecto.
- CUSTOM_JOB_ID_PREFIX: el nombre del prefijo de ID de trabajo personalizado que estableciste para tu consulta continua.
Haz clic en Crear métrica.
En el menú de navegación, haz clic en Métricas basadas en registros. La métrica que acabas de crear aparece en la lista de métricas definidas por el usuario.
En la fila de tu métrica, haz clic en Más acciones y, luego, en Crear alerta a partir de métricas.
Haz clic en Siguiente. No es necesario cambiar la configuración predeterminada en la página Modo de configuración de políticas.
Haz clic en Siguiente. No es necesario cambiar la configuración predeterminada en la página Configurar activador de alertas.
Selecciona tus canales de notificaciones y, luego, ingresa un nombre para la política de alertas.
Haz clic en Crear política.

Para probar tu alerta, ejecuta una consulta continua con el prefijo de ID de trabajo personalizado que seleccionaste y, luego, cancélala. Es posible que la alerta tarde unos minutos en llegar a tu canal de notificaciones.

Vuelve a intentar con las consultas que fallaron

Volver a intentar con una consulta continua con errores puede ayudar a evitar situaciones en las que una canalización continua esté inactiva durante un período prolongado o requiera la intervención humana para reiniciarse. Entre las cosas importantes que debes tener en cuenta cuando vuelves a intentar con una consulta continua con errores, se incluyen las siguientes:

Si es tolerable volver a procesar una cierta cantidad de datos procesados por la consulta anterior antes de que fallara.
Cómo controlar los reintentos limitantes o usar una retirada exponencial.

Un enfoque posible para automatizar el reintento de consultas es el siguiente:

Crea un receptor de Cloud Logging basado en un filtro de inclusión que coincida con los siguientes criterios para enrutar registros a un tema de Pub/Sub:
```
resource.type = "bigquery_project"
protoPayload.resourceName : "projects/PROJECT_ID/jobs/CUSTOM_JOB_ID_PREFIX"
severity = ERROR
```
Reemplaza lo siguiente:
- PROJECT_ID: nombre del proyecto.
- CUSTOM_JOB_ID_PREFIX: el nombre del prefijo de ID de trabajo personalizado que estableciste para tu consulta continua.
Crea una función de Cloud Run que se active en respuesta a que Pub/Sub reciba registros que coincidan con tu filtro.

La función de Cloud Run podría aceptar la carga útil de datos del mensaje de Pub/Sub y tratar de iniciar una nueva consulta continua con la misma sintaxis de SQL que la consulta con errores, pero al principio, justo después de que se detuvo el trabajo anterior.

Por ejemplo, puedes usar una función similar a la siguiente:

Python

Antes de probar este ejemplo, sigue las instrucciones de configuración para Python incluidas en la guía de inicio rápido de BigQuery sobre cómo usar bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de BigQuery para Python.

Para autenticarte en BigQuery, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para bibliotecas cliente.

Antes de ejecutar muestras de código, configura la variable de entorno 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.")

¿Qué sigue?

Obtén información para crear y ejecutar consultas continuas.

Supervisa consultas continuas

Usa vistas INFORMATION_SCHEMA

Ver detalles del trabajo

Visualiza los detalles de la asignación de reservas

Visualiza información sobre el consumo de ranuras

Usa el gráfico de ejecución de consultas

Ver el historial de un trabajo

Usa el explorador de trabajos

Use Cloud Monitoring

Recibe alertas sobre consultas con errores

Vuelve a intentar con las consultas que fallaron

Python

¿Qué sigue?

Usa vistas `INFORMATION_SCHEMA`