Google verwendet KI-Technologie, um Inhalte in Ihre bevorzugte Sprache zu übersetzen. KI-Übersetzungen können Fehler enthalten.

Kontinuierliche Abfragen überwachen

Sie können kontinuierliche Abfragen in BigQuery mit den folgenden BigQuery-Tools überwachen:

Da eine kontinuierliche BigQuery-Abfrage sehr lange andauert, fehlen Messwerte, die normalerweise nach Abschluss einer SQL-Abfrage generiert werden, eventuell nicht oder sind ungenau.

`INFORMATION_SCHEMA`-Ansichten verwenden

Sie können eine Reihe von INFORMATION_SCHEMA-Ansichten verwenden, um kontinuierliche Abfragen und kontinuierliche Abfragereservierungen zu überwachen.

Auftragsdetails aufrufen

Mit der Ansicht JOBS können Sie Metadaten des kontinuierlichen Abfragejobs abrufen.

Die folgende Abfrage gibt die Metadaten für alle aktiven kontinuierlichen Abfragen zurück. Die Metadaten enthalten den Zeitstempel des Ausgabewasserzeichens, der den Punkt darstellt, bis zu dem die kontinuierliche Abfrage Daten erfolgreich verarbeitet hat.

Rufen Sie in der Google Cloud Console die Seite BigQuery auf.

BigQuery aufrufen

Führen Sie im Abfrageeditor folgende Abfrage aus:

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

Ersetzen Sie Folgendes:

PROJECT_ID: die ID des Projekts.
REGION: ist ein beliebiger Dataset-Regionsname. Beispiel: region-us

Details zur Reservierungszuweisung ansehen

Mit den Ansichten ASSIGNMENTS und RESERVATIONS können Sie Details zur Zuweisung von kontinuierlichen Abfragereservierungen abrufen.

Geben Sie die Details der Reservierungszuweisung für kontinuierliche Abfragen zurück:

Rufen Sie in der Google Cloud Console die Seite BigQuery auf.

BigQuery aufrufen

Führen Sie im Abfrageeditor folgende Abfrage aus:

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';

Ersetzen Sie dabei Folgendes:

ADMIN_PROJECT_ID: Die ID des Administrationsprojekts, dem die Reservierung gehört.
LOCATION: Der Standort der Reservierung.
PROJECT_ID: die ID des Projekts, das der Reservierung zugewiesen ist. Es werden nur Informationen zu kontinuierlichen Abfragen zurückgegeben, die in diesem Projekt ausgeführt werden.

Informationen zum Slot-Verbrauch aufrufen

Sie können die Ansichten ASSIGNMENTS, RESERVATIONS und JOBS_TIMELINE verwenden, um Informationen zum Slot-Verbrauch für kontinuierliche Abfragen zu erhalten.

Informationen zum Slot-Verbrauch für kontinuierliche Abfragen zurückgeben:

Rufen Sie in der Google Cloud Console die Seite BigQuery auf.

BigQuery aufrufen

Führen Sie im Abfrageeditor folgende Abfrage aus:

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;

Ersetzen Sie dabei Folgendes:

ADMIN_PROJECT_ID: Die ID des Administrationsprojekts, dem die Reservierung gehört.
LOCATION: Der Standort der Reservierung.
PROJECT_ID: die ID des Projekts, das der Reservierung zugewiesen ist. Es werden nur Informationen zu kontinuierlichen Abfragen zurückgegeben, die in diesem Projekt ausgeführt werden.

Sie können kontinuierliche Abfragereservierungen auch mit anderen Tools wie Metrics Explorer und administrativen Ressourcendiagrammen überwachen. Weitere Informationen finden Sie unter BigQuery Reservations überwachen.

Ausführungsgrafik für Abfragen verwenden

Mit dem Diagramm zur Abfrageausführung können Sie Leistungsinformationen und allgemeine Statistiken für eine kontinuierliche Abfrage abrufen. Weitere Informationen finden Sie unter Statistiken zur Abfrageleistung ansehen.

Jobverlauf ansehen

Details zu kontinuierlichen Abfragejobs können Sie in Ihrem persönlichen Jobverlauf oder im Jobverlauf des Projekts einsehen. Weitere Informationen finden Sie unter Jobdetails ansehen.

Die Verlaufsliste der Jobs wird nach der Startzeit des Jobs sortiert. Daher befinden sich kontinuierliche Abfragen, die bereits seit einer Weile ausgeführt werden, möglicherweise nicht nahe am Anfang der Liste.

Job-Explorer verwenden

Filtern Sie Ihre Jobs im Job-Explorer, um kontinuierliche Abfragen anzuzeigen. Setzen Sie dazu den Filter Jobkategorie auf Kontinuierliche Abfrage.

Cloud Monitoring verwenden

Mit Cloud Monitoring können Sie Messwerte für kontinuierliche BigQuery-Abfragen aufrufen. Weitere Informationen finden Sie unter Dashboards, Diagramme und Benachrichtigungen erstellen und unter Für die Visualisierung verfügbare Messwerte.

Benachrichtigung bei fehlgeschlagenen Abfragen

Anstatt regelmäßig zu prüfen, ob Ihre kontinuierlichen Abfragen fehlgeschlagen sind, kann es hilfreich sein, eine Benachrichtigung zu erstellen, die Sie über Fehler informiert. Eine Möglichkeit besteht darin, einen benutzerdefinierten logbasierten Cloud Logging-Messwert mit einem Filter für Ihre Jobs und eine Cloud Monitoring-Benachrichtigungsrichtlinie basierend auf diesem Messwert zu erstellen:

Verwenden Sie beim Erstellen einer kontinuierlichen Abfrage ein benutzerdefiniertes Job-ID-Präfix. Mehrere kontinuierliche Abfragen können dasselbe Präfix haben. Sie können beispielsweise das Präfix prod- verwenden, um eine Produktionsabfrage anzugeben.
Rufen Sie in der Google Cloud Console die Seite Logbasierte Messwerte auf.

Zu „Logbasierte Messwerte”
Klicken Sie auf Messwert erstellen. Der Bereich Log-Messwert erstellen wird angezeigt.
Wählen Sie als Messwerttyp die Option Zähler aus.
Geben Sie im Abschnitt Details einen Namen für den Messwert ein. Beispiel: CUSTOM_JOB_ID_PREFIX-metric
Geben Sie im Abschnitt Filterauswahl im Editor Filter erstellen Folgendes ein:
```
resource.type = "bigquery_project"
protoPayload.resourceName : "projects/PROJECT_ID/jobs/CUSTOM_JOB_ID_PREFIX"
severity = ERROR
```
Ersetzen Sie Folgendes:
- PROJECT_ID: Name Ihres Projekts
- CUSTOM_JOB_ID_PREFIX: der Name des Präfixes für die benutzerdefinierte Job-ID, das Sie für Ihre kontinuierliche Abfrage festgelegt haben.
Klicken Sie auf Messwert erstellen.
Klicken Sie im Navigationsmenü auf Logbasierte Messwerte. Der gerade erstellte Messwert wird in der Liste der benutzerdefinierten Messwerte angezeigt.
Klicken Sie in der Zeile des Messwerts auf Weitere Aktionen und dann auf Benachrichtigung mit dem Messwert erstellen.
Klicken Sie auf Weiter. Sie müssen die Standardeinstellungen auf der Seite Richtlinienkonfigurationsmodus nicht ändern.
Klicken Sie auf Weiter. Sie müssen die Standardeinstellungen auf der Seite Benachrichtigungstrigger konfigurieren nicht ändern.
Wählen Sie die Benachrichtigungskanäle aus und geben Sie einen Namen für die Benachrichtigungsrichtlinie ein.
Klicken Sie auf Richtlinie erstellen.

Sie können Ihre Benachrichtigung testen, indem Sie eine kontinuierliche Abfrage mit dem von Ihnen ausgewählten Präfix für die benutzerdefinierte Job-ID ausführen und dann abbrechen. Es kann einige Minuten dauern, bis die Benachrichtigung Ihren Benachrichtigungskanal erreicht.

Fehlgeschlagene Abfragen wiederholen

Wenn Sie eine fehlgeschlagene kontinuierliche Abfrage wiederholen, können Sie Situationen vermeiden, in denen eine kontinuierliche Pipeline über einen längeren Zeitraum ausfällt oder manuell neu gestartet werden muss. Wichtige Punkte, die Sie beachten sollten, wenn Sie eine fehlgeschlagene kontinuierliche Abfrage wiederholen:

Ob es akzeptabel ist, einen Teil der Daten, die von der vorherigen Abfrage vor dem Fehlschlagen verarbeitet wurden, noch einmal zu verarbeiten.
Wie Sie Wiederholungsversuche einschränken oder exponentiellen Backoff verwenden.

Ein möglicher Ansatz zur Automatisierung der Wiederholung von Abfragen ist folgender:

Erstellen Sie eine Cloud Logging-Senke basierend auf einem „Einschließen“-Filter, der die folgenden Kriterien erfüllt, um Logs an ein Pub/Sub-Thema weiterzuleiten:
```
resource.type = "bigquery_project"
protoPayload.resourceName : "projects/PROJECT_ID/jobs/CUSTOM_JOB_ID_PREFIX"
severity = ERROR
```
Ersetzen Sie Folgendes:
- PROJECT_ID: Name Ihres Projekts
- CUSTOM_JOB_ID_PREFIX: der Name des Präfixes für die benutzerdefinierte Job-ID, das Sie für Ihre kontinuierliche Abfrage festgelegt haben.
Erstellen Sie eine Cloud Run-Funktion, die ausgelöst wird, wenn Pub/Sub Logs empfängt, die Ihrem Filter entsprechen.

Die Cloud Run-Funktion könnte die Datennutzlast aus der Pub/Sub-Nachricht akzeptieren und versuchen, eine neue kontinuierliche Abfrage mit derselben SQL-Syntax wie die fehlgeschlagene Abfrage zu starten, aber am Anfang direkt nach dem Beenden des vorherigen Jobs.

Sie können beispielsweise eine Funktion wie die folgende verwenden:

Python

Bevor Sie dieses Beispiel anwenden, folgen Sie den Schritten zur Einrichtung von Python in der BigQuery-Kurzanleitung zur Verwendung von Clientbibliotheken. Weitere Angaben finden Sie in der Referenzdokumentation zur BigQuery Python API.

Richten Sie zur Authentifizierung bei BigQuery die Standardanmeldedaten für Anwendungen ein. Weitere Informationen finden Sie unter Authentifizierung für Clientbibliotheken einrichten.

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.")

Nächste Schritte

Kontinuierliche Abfragen erstellen und ausführen

Kontinuierliche Abfragen überwachen

INFORMATION_SCHEMA-Ansichten verwenden

Auftragsdetails aufrufen

Details zur Reservierungszuweisung ansehen

Informationen zum Slot-Verbrauch aufrufen

Ausführungsgrafik für Abfragen verwenden

Jobverlauf ansehen

Job-Explorer verwenden

Cloud Monitoring verwenden

Benachrichtigung bei fehlgeschlagenen Abfragen

Fehlgeschlagene Abfragen wiederholen

Python

Nächste Schritte

`INFORMATION_SCHEMA`-Ansichten verwenden