Kontinuierliche Abfragen erstellen

In diesem Dokument wird beschrieben, wie Sie eine kontinuierliche Abfrage in BigQuery ausführen.

Kontinuierliche BigQuery-Abfragen sind SQL-Anweisungen, die kontinuierlich ausgeführt werden und Daten nahezu in Echtzeit verarbeiten. Sie können die von einer kontinuierlichen Abfrage erzeugten Ausgabezeilen in die folgenden Ziele schreiben oder exportieren:

Kontotyp auswählen

Sie können einen Job für kontinuierliche Abfragen mit einem Nutzerkonto erstellen und ausführen. Sie haben auch die Möglichkeit, einen Job für kontinuierliche Abfragen mit einem Nutzerkonto zu erstellen und dann mit einem Dienstkonto auszuführen. Sie müssen ein Dienstkonto verwenden, um eine kontinuierliche Abfrage auszuführen, die Ergebnisse in ein Pub/Sub-Thema exportiert.

Wenn Sie ein Nutzerkonto verwenden, wird eine kontinuierliche Abfrage bis zu zwei Tage lang ausgeführt. Wenn Sie ein Dienstkonto verwenden, wird eine kontinuierliche Abfrage bis zu 150 Tage lang ausgeführt. Weitere Informationen finden Sie unter Autorisierung.

Erforderliche Berechtigungen

In diesem Abschnitt werden die Berechtigungen beschrieben, die Sie zum Erstellen und Ausführen einer kontinuierlichen Abfrage benötigen. Alternativ zu den genannten IAM-Rollen (Identity and Access Management) können Sie die erforderlichen Berechtigungen auch über benutzerdefinierte Rollen erhalten.

Berechtigungen bei Verwendung eines Nutzerkontos

Dieser Abschnitt enthält Informationen zu den Rollen und Berechtigungen, die zum Erstellen und Ausführen einer kontinuierlichen Abfrage mit einem Nutzerkonto erforderlich sind.

Zum Erstellen eines Jobs in BigQuery muss das Nutzerkonto die IAM-Berechtigung bigquery.jobs.create haben. Jede der folgenden IAM-Rollen gewährt die Berechtigung bigquery.jobs.create:

Zum Exportieren von Daten aus einer BigQuery-Tabelle muss das Nutzerkonto die IAM-Berechtigung bigquery.tables.export haben. Jede der folgenden IAM-Rollen gewährt die Berechtigung bigquery.tables.export:

Zum Schreiben von Daten in eine BigQuery-Tabelle muss das Nutzerkonto die IAM-Berechtigung bigquery.tables.updateData haben. Jede der folgenden IAM-Rollen gewährt die Berechtigung bigquery.tables.updateData:

Wenn das Nutzerkonto die APIs aktivieren muss, die für Ihren Anwendungsfall mit kontinuierlichen Abfragen erforderlich sind, muss das Nutzerkonto die Rolle Service Usage Admin (roles/serviceusage.serviceUsageAdmin) haben.

Wenn Sie Daten in verwaltete Apache Iceberg-Tabellen schreiben möchten, muss das Nutzerkonto zusätzlich zu den BigQuery-Berechtigungen die Rolle BigQuery Connection User (roles/bigquery.connectionUser) haben, um die mit der verwalteten Iceberg-Tabelle verknüpfte Google Cloud -Ressourcenverbindung zu verwenden. Außerdem muss das mit dieser Verbindung verknüpfte Dienstkonto die entsprechenden Berechtigungen (z. B. roles/storage.objectUser und roles/storage.legacyBucketReader) für den zugrunde liegenden Cloud Storage-Bucket haben. Weitere Informationen finden Sie unter Verwaltete Apache Iceberg-Tabellen erstellen und verwenden.

Berechtigungen bei Verwendung eines Dienstkontos

Dieser Abschnitt enthält Informationen zu den Rollen und Berechtigungen, die für das Nutzerkonto erforderlich sind, mit dem die kontinuierliche Abfrage erstellt wird, sowie zum Dienstkonto, das die kontinuierliche Abfrage ausführt.

Nutzerkontoberechtigungen

Zum Erstellen eines Jobs in BigQuery muss das Nutzerkonto die IAM-Berechtigung bigquery.jobs.create haben. Jede der folgenden IAM-Rollen gewährt die Berechtigung bigquery.jobs.create:

Wenn Sie einen Job einreichen möchten, der mit einem Dienstkonto ausgeführt wird, muss das Nutzerkonto die Rolle Dienstkontonutzer (roles/iam.serviceAccountUser) haben. Wenn Sie dasselbe Nutzerkonto zum Erstellen des Dienstkontos verwenden, muss das Nutzerkonto die Rolle Service Account Admin (roles/iam.serviceAccountAdmin) haben. Informationen dazu, wie Sie den Zugriff eines Nutzers auf ein einzelnes Dienstkonto anstelle aller Dienstkonten in einem Projekt beschränken, finden Sie unter Einzelne Rolle zuweisen.

Wenn das Nutzerkonto die APIs aktivieren muss, die für Ihren Anwendungsfall mit kontinuierlichen Abfragen erforderlich sind, muss das Nutzerkonto die Rolle Service Usage Admin (roles/serviceusage.serviceUsageAdmin) haben.

Dienstkontoberechtigungen

Zum Exportieren von Daten aus einer BigQuery-Tabelle muss das Dienstkonto die IAM-Berechtigung bigquery.tables.export haben. Jede der folgenden IAM-Rollen gewährt die Berechtigung bigquery.tables.export:

Zum Schreiben von Daten in eine BigQuery-Tabelle muss das Dienstkonto die IAM-Berechtigung bigquery.tables.updateData haben. Jede der folgenden IAM-Rollen gewährt die Berechtigung bigquery.tables.updateData:

Wenn Sie Daten in verwaltete Apache Iceberg-Tabellen schreiben möchten, muss das Dienstkonto zusätzlich zu den BigQuery-Berechtigungen auch die Rolle BigQuery-Verbindungsnutzer (roles/bigquery.connectionUser) haben, um die mit der verwalteten Iceberg-Tabelle verknüpfte Google Cloud -Ressourcenverbindung zu verwenden. Außerdem muss das mit dieser Verbindung verknüpfte Dienstkonto die entsprechenden Berechtigungen (z. B. roles/storage.objectUser und roles/storage.legacyBucketReader) für den zugrunde liegenden Cloud Storage-Bucket haben. Weitere Informationen finden Sie unter Verwaltete Apache Iceberg-Tabellen erstellen und verwenden.

Hinweis

  1. Wählen Sie in der Google Cloud Console auf der Seite für die Projektauswahl ein Projekt von Google Cloud aus oder erstellen Sie eines.

    Erforderliche Rollen zum Auswählen oder Erstellen eines Projekts

    • Projekt auswählen: Für die Auswahl eines Projekts ist keine bestimmte IAM-Rolle erforderlich. Sie können jedes Projekt auswählen, für das Ihnen eine Rolle zugewiesen wurde.
    • Projekt erstellen: Zum Erstellen eines Projekts benötigen Sie die Rolle „Project Creator“ (roles/resourcemanager.projectCreator), die die Berechtigung resourcemanager.projects.create enthält. Informationen zum Zuweisen von Rollen

    Zur Projektauswahl

  2. Prüfen Sie, ob die Abrechnung für Ihr Google Cloud -Projekt aktiviert ist.

  3. Aktivieren Sie die BigQuery API, falls sie noch nicht aktiviert ist.

    Erforderliche Rollen zum Aktivieren von APIs

    Zum Aktivieren von APIs benötigen Sie die Berechtigung serviceusage.services.enable. Wenn Sie das Projekt erstellt haben, haben Sie diese Berechtigung wahrscheinlich bereits über die Inhaberrolle (roles/owner). Andernfalls können Sie diese Berechtigung über die Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin) erhalten. Informationen zum Zuweisen von Rollen

    API aktivieren

Reservierung erstellen

Erstellen Sie eine Reservierung für die Enterprise- oder Enterprise Plus-Version und dann eine Reservierungszuweisung mit dem Jobtyp CONTINUOUS. Diese Reservierung kann Autoscaling und die Freigabe inaktiver Slots verwenden. Für Reservierungszuweisungen für kontinuierliche Abfragen gelten Reservierungseinschränkungen.

Nach Pub/Sub exportieren

Für den Export von Daten nach Pub/Sub sind zusätzliche APIs, IAM-Berechtigungen und Google Cloud -Ressourcen erforderlich. Weitere Informationen finden Sie unter In Pub/Sub exportieren.

Mutationen mit CHANGES verarbeiten

Wenn Sie Daten nach Pub/Sub exportieren, können Sie die Änderungsverlaufsfunktion CHANGES verwenden. Die Funktion CHANGES verarbeitet alle Zeilen, die sich in der Quelltabelle geändert haben, einschließlich Anhängen und Mutationen.

Benutzerdefinierte Attribute als Metadaten in Pub/Sub-Nachrichten einbetten

Sie können Pub/Sub-Attribute verwenden, um zusätzliche Informationen zur Nachricht bereitzustellen, z. B. ihre Priorität, ihren Ursprung, ihr Ziel oder zusätzliche Metadaten. Sie können Attribute auch verwenden, um Nachrichten im Abo zu filtern.

Wenn eine Spalte in einem Ergebnis einer kontinuierlichen Abfrage den Namen _ATTRIBUTES hat, werden ihre Werte in die Pub/Sub-Nachrichtenattribute kopiert. Die in _ATTRIBUTES angegebenen Felder werden als Attributschlüssel verwendet.

Die Spalte _ATTRIBUTES muss vom Typ JSON sein und das Format ARRAY<STRUCT<STRING, STRING>> oder STRUCT<STRING> haben.

Ein Beispiel finden Sie unter Daten in ein Pub/Sub-Thema exportieren.

Nach Bigtable exportieren

Für den Export von Daten nach Bigtable sind zusätzliche APIs, IAM-Berechtigungen und Google Cloud-Ressourcen erforderlich. Weitere Informationen finden Sie unter In Bigtable exportieren.

In Cloud Spanner exportieren

Für den Export von Daten nach Cloud Spanner sind zusätzliche APIs, IAM-Berechtigungen und Google Cloud-Ressourcen erforderlich. Weitere Informationen finden Sie unter Nach Spanner exportieren (umgekehrte ETL).

Daten in eine BigQuery-Tabelle oder in von Apache Iceberg verwaltete Tabellen schreiben

Sie können Daten mit einer INSERT-Anweisung in eine BigQuery-Tabelle oder eine verwaltete Iceberg-Tabelle schreiben.

Wenn Sie die Ausgabe einer kontinuierlichen Abfrage in eine verwaltete Iceberg-Tabelle schreiben, gilt Folgendes:

  • Die verwaltete Iceberg-Zieltabelle muss bereits vorhanden sein, bevor Sie die kontinuierliche Abfrage erstellen. Eine Anleitung zum Erstellen einer verwalteten Iceberg-Tabelle finden Sie unter Verwaltete Apache Iceberg-Tabellen erstellen und verwenden.
  • Die Abfragesyntax ist mit der Syntax zum Schreiben in eine Standard-BigQuery-Tabelle identisch. Sie können die Zieltabelle in der INSERT INTO-Klausel angeben und müssen in der SQL-Anweisung keine Verbindungs- oder Bucket-Optionen angeben.
  • Das Nutzer- oder Dienstkonto, mit dem die kontinuierliche Abfrage ausgeführt wird, muss die erforderlichen Berechtigungen für die Zieltabelle, die Google Cloud -Ressourcenverbindung und den zugrunde liegenden Cloud Storage-Bucket haben. Weitere Informationen finden Sie unter Erforderliche Berechtigungen.

Ein Beispiel finden Sie unter Daten in eine BigQuery-Tabelle oder in von Apache Iceberg verwaltete Tabellen schreiben.

KI-Funktionen verwenden

Für die Verwendung einer unterstützten KI-Funktion in einer kontinuierlichen Abfrage sind zusätzliche APIs, IAM-Berechtigungen und Google Cloud-Ressourcen erforderlich. Weitere Informationen finden Sie je nach Anwendungsfall in einem der folgenden Themen:

Wenn Sie eine KI-Funktion in einer kontinuierlichen Abfrage verwenden, sollten Sie darauf achten, dass die Abfrageausgabe innerhalb des Kontingents für die Funktion bleibt. Wenn Sie das Kontingent überschreiten, müssen Sie die nicht verarbeiteten Datensätze möglicherweise separat verarbeiten.

Startzeit für die kontinuierliche Abfrage angeben

Sie müssen die APPENDS-Funktion oder die CHANGES-Funktion (bei Exporten nach Pub/Sub) in der FROM-Klausel einer kontinuierlichen Abfrage verwenden, um die frühesten zu verarbeitenden Daten anzugeben. Zum Beispiel APPENDS(TABLE my_table, start_timestamp).

Das Argument start_timestamp definiert den Zeitpunkt, zu dem die kontinuierliche Abfrage mit der Datenverarbeitung beginnt. Beispiel: APPENDS(TABLE my_table, CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) weist BigQuery an, Daten zu verarbeiten, die der Tabelle my_table maximal 10 Minuten vor dem Start der kontinuierlichen Abfrage hinzugefügt wurden. Daten, die anschließend zu my_table hinzugefügt werden, werden verarbeitet, sobald sie eingehen. Es gibt keine Verzögerung bei der Datenverarbeitung.

Wenn Sie das Argument start_timestamp angeben, muss der Wert innerhalb des Zeitstempelfensters der Tabelle liegen. Bei Standardtabellen beträgt dieses standardmäßig sieben Tage. Wenn Sie start_timestamp auf NULL setzen, wird standardmäßig der Erstellungszeitpunkt der Tabelle verwendet. Der Wert NULL wird nicht empfohlen, da ein Fehler zurückgegeben wird, wenn die Tabelle vor dem Zeitreisenfenster erstellt wurde. Eine Abfrage mit NULL kann für eine neu erstellte Tabelle erfolgreich sein, aber später fehlschlagen, wenn der Erstellungszeitstempel der Tabelle außerhalb des siebentägigen Zeitfensters liegt.

Geben Sie kein end_timestamp-Argument für die APPENDS-Funktion an, wenn Sie sie in einer kontinuierlichen Abfrage verwenden.

Das folgende Beispiel zeigt, wie Sie eine kontinuierliche Abfrage ab einem bestimmten Zeitpunkt mit der Funktion APPENDS starten, wenn Sie eine BigQuery-Tabelle abfragen, die Informationen zu Streamingtaxifahrten empfängt:

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

Startpunkt vor dem Zeitreisefenster angeben

Wenn Sie Daten einbeziehen möchten, die außerhalb des Zeitreisenfensters von sieben Tagen liegen, verwenden Sie eine Standardabfrage, um Daten bis zu einem bestimmten Zeitpunkt nachzutragen, und starten Sie dann eine kontinuierliche Abfrage ab diesem Zeitpunkt. Der Zeitstempel für die Übergabe muss innerhalb des siebentägigen Zeitfensters liegen, damit die kontinuierliche Abfrage dort fortgesetzt werden kann, wo die Standardabfrage aufgehört hat.

Das folgende Beispiel zeigt, wie Sie ältere Daten aus einer BigQuery-Tabelle, die Informationen zu Streamingtaxifahrten empfängt, nachträglich einfügen und dann zu einer kontinuierlichen Abfrage wechseln.

  1. Führen Sie eine Standardabfrage aus, um Daten bis zu einem bestimmten Zeitpunkt nachzutragen:

    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 handoff point.
  -- This handoff 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. Führen Sie eine kontinuierliche Abfrage ab dem Zeitpunkt aus, an dem die Abfrage beendet wurde:

    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 + 1 microsecond.
    -- This timestamp must be within the time travel window.
    TIMESTAMP '2025-01-01 00:00:00 UTC' + INTERVAL 1 MICROSECOND)
WHERE
  ride_status = 'dropoff';

Kontinuierliche Abfrage mit einem Nutzerkonto ausführen

In diesem Abschnitt wird beschrieben, wie Sie eine kontinuierliche Abfrage mit einem Nutzerkonto ausführen. Sobald die kontinuierliche Abfrage ausgeführt wird, können Sie die Google Cloud Console, das Terminalfenster oder die Anwendung schließen, ohne die Abfrageausführung zu unterbrechen. Eine kontinuierliche Abfrage, die von einem Nutzerkonto ausgeführt wird, wird maximal zwei Tage lang ausgeführt und dann automatisch beendet. Wenn Sie neue eingehende Daten weiter verarbeiten möchten, starten Sie eine neue kontinuierliche Abfrage und geben Sie einen Startpunkt an. Informationen zum Automatisieren dieses Vorgangs finden Sie unter Fehlgeschlagene Abfragen wiederholen.

So führen Sie eine kontinuierliche Abfrage aus:

Console

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

    BigQuery aufrufen

  2. Klicken Sie im Abfrageeditor auf Bearbeiten > Abfragemodus > Kontinuierliche Abfrage. Klicken Sie auf Bestätigen, um diese Auswahl zu bestätigen.

  3. Optional: Wenn Sie festlegen möchten, wie lange die Abfrage ausgeführt wird, klicken Sie auf Bearbeiten > Abfrageeinstellungen und legen Sie das Job-Zeitlimit in Millisekunden fest.

  4. Geben Sie im Abfrageeditor die SQL-Anweisung für die kontinuierliche Abfrage ein. Die SQL-Anweisung darf nur unterstützte Vorgänge enthalten.

  5. Klicken Sie auf Ausführen.

bq

  1. Aktivieren Sie Cloud Shell in der Google Cloud Console.

    Cloud Shell aktivieren

    Unten in der Google Cloud Console wird eine Cloud Shell-Sitzung gestartet und eine Eingabeaufforderung angezeigt. Cloud Shell ist eine Shell-Umgebung, in der das Google Cloud CLI bereits installiert ist und Werte für Ihr aktuelles Projekt bereits festgelegt sind. Das Initialisieren der Sitzung kann einige Sekunden dauern.

  2. Führen Sie in Cloud Shell die kontinuierliche Abfrage mit dem bq query-Befehl und dem Flag --continuous aus:

    bq query --use_legacy_sql=false --continuous=true
'QUERY'

    Ersetzen Sie QUERY durch die SQL-Anweisung für die kontinuierliche Abfrage. Die SQL-Anweisung darf nur unterstützte Vorgänge enthalten. Mit dem Flag --job_timeout_ms können Sie festlegen, wie lange die Abfrage ausgeführt wird.

API

Führen Sie die kontinuierliche Abfrage aus, indem Sie die Methode jobs.insert aufrufen. Sie müssen das Feld continuous in der JobConfigurationQuery der Job-Ressource auf true setzen, die Sie übergeben. Optional können Sie festlegen, wie lange die Abfrage ausgeführt wird, indem Sie das Feld jobTimeoutMs festlegen.

curl --request POST \
  "https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs" \
  --header "Authorization: Bearer $(gcloud auth print-access-token)" \
  --header "Content-Type: application/json; charset=utf-8" \
  --data '{"configuration":{"query":{"query":"QUERY","useLegacySql":false,"continuous":true}}}' \
  --compressed

Ersetzen Sie Folgendes:

  • PROJECT_ID: Ihre Projekt-ID.
  • QUERY: die SQL-Anweisung für die kontinuierliche Abfrage. Die SQL-Anweisung darf nur unterstützte Vorgänge enthalten.

Kontinuierliche Abfrage mit einem Dienstkonto ausführen

In diesem Abschnitt wird beschrieben, wie Sie eine kontinuierliche Abfrage mit einem Dienstkonto ausführen. Sobald die kontinuierliche Abfrage ausgeführt wird, können Sie die Google Cloud Console, das Terminalfenster oder die Anwendung schließen, ohne die Abfrageausführung zu unterbrechen. Eine kontinuierliche Abfrage, die mit einem Dienstkonto ausgeführt wird, kann bis zu 150 Tage lang ausgeführt werden und wird dann automatisch beendet. Wenn Sie neue eingehende Daten weiter verarbeiten möchten, starten Sie eine neue kontinuierliche Abfrage und geben Sie einen Startpunkt an. Informationen zum Automatisieren dieses Vorgangs finden Sie unter Fehlgeschlagene Abfragen wiederholen.

Wenn Sie ein Dienstkonto verwenden, um eine kontinuierliche Abfrage über mehrere Projekte hinweg auszuführen, müssen Sie Ihre Organisationsrichtlinie aktualisieren, damit Dienstkonten in einem Projekt an Ressourcen in einem anderen Projekt angehängt werden können. Weitere Informationen finden Sie unter Ermöglichen, dass Dienstkonten projektübergreifend angehängt werden können.

So verwenden Sie ein Dienstkonto, um eine kontinuierliche Abfrage auszuführen:

Console

  1. Erstellen Sie ein Dienstkonto.
  2. Gewähren Sie die erforderlichen Berechtigungen für das Dienstkonto:

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

    BigQuery aufrufen

  4. Klicken Sie im Abfrageeditor auf Bearbeiten > Abfragemodus > Kontinuierliche Abfrage. Klicken Sie auf Bestätigen, um diese Auswahl zu bestätigen.

  5. Klicken Sie im Abfrageeditor auf Bearbeiten > Abfrageeinstellungen.

  6. Wählen Sie im Abschnitt Kontinuierliche Abfrage im Feld Dienstkonto das von Ihnen erstellte Dienstkonto aus.

  7. Optional: Wenn Sie festlegen möchten, wie lange die Abfrage ausgeführt wird, legen Sie das Job-Zeitlimit in Millisekunden fest.

  8. Klicken Sie auf Speichern.

  9. Geben Sie im Abfrageeditor die SQL-Anweisung für die kontinuierliche Abfrage ein. Die SQL-Anweisung darf nur unterstützte Vorgänge enthalten.

  10. Klicken Sie auf Ausführen.

bq

  1. Erstellen Sie ein Dienstkonto.
  2. Gewähren Sie die erforderlichen Berechtigungen für das Dienstkonto:

  3. Aktivieren Sie Cloud Shell in der Google Cloud Console.

    Cloud Shell aktivieren

    Unten in der Google Cloud Console wird eine Cloud Shell-Sitzung gestartet und eine Eingabeaufforderung angezeigt. Cloud Shell ist eine Shell-Umgebung, in der das Google Cloud CLI bereits installiert ist und Werte für Ihr aktuelles Projekt bereits festgelegt sind. Das Initialisieren der Sitzung kann einige Sekunden dauern.

  4. Führen Sie die kontinuierliche Abfrage in der Befehlszeile mit dem Befehl bq query und den folgenden Flags aus:

    • Legen Sie das Flag --continuous auf true fest, um die Abfrage kontinuierlich auszuführen.
    • Verwenden Sie das Flag --connection_property, um ein Dienstkonto anzugeben.
    • Optional: Setzen Sie das Flag --job_timeout_ms, um die Abfragelaufzeit zu begrenzen.
    bq query --project_id=PROJECT_ID --use_legacy_sql=false \
--continuous=true --connection_property=service_account=SERVICE_ACCOUNT_EMAIL \
'QUERY'

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Ihre Projekt-ID.
    • SERVICE_ACCOUNT_EMAIL: die E-Mail-Adresse des Dienstkontos. Sie finden die E‑Mail-Adresse des Dienstkontos in der Google Cloud Console auf der Seite Dienstkonten.
    • QUERY: die SQL-Anweisung für die kontinuierliche Abfrage. Die SQL-Anweisung darf nur unterstützte Vorgänge enthalten.

API

  1. Erstellen Sie ein Dienstkonto.
  2. Gewähren Sie die erforderlichen Berechtigungen für das Dienstkonto:

  3. Führen Sie die kontinuierliche Abfrage aus, indem Sie die Methode jobs.insert aufrufen. Legen Sie die folgenden Felder in der JobConfigurationQuery-Ressource der Job-Ressource fest, die Sie übergeben:

    • Legen Sie das Feld continuous auf true fest, um die Abfrage kontinuierlich zu machen.
    • Verwenden Sie das Feld connectionProperties, um ein Dienstkonto anzugeben.

    Optional können Sie festlegen, wie lange die Abfrage ausgeführt wird. Setzen Sie dazu das Feld jobTimeoutMs in der Ressource JobConfiguration.

    curl --request POST \
  "https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs" \
  --header "Authorization: Bearer $(gcloud auth print-access-token)" \
  --header "Content-Type: application/json; charset=utf-8" \
  --data '{"configuration":{"query":{"query":"QUERY","useLegacySql":false,"continuous":true,"connectionProperties":[{"key":"service_account","value":"SERVICE_ACCOUNT_EMAIL"}]}}}' \
  --compressed

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Ihre Projekt-ID.
    • QUERY: die SQL-Anweisung für die kontinuierliche Abfrage. Die SQL-Anweisung darf nur unterstützte Vorgänge enthalten.
    • SERVICE_ACCOUNT_EMAIL: die E-Mail-Adresse des Dienstkontos. Die E‑Mail-Adresse des Dienstkontos finden Sie in der Google Cloud Console auf der Seite Dienstkonten.

Benutzerdefinierte Job-ID erstellen

Jedem Abfragejob wird eine Job-ID zugewiesen, mit der Sie den Job suchen und verwalten können. Standardmäßig werden Job-IDs zufällig generiert. Damit Sie die Job-ID einer kontinuierlichen Abfrage im Jobverlauf oder im Jobs Explorer leichter finden, können Sie ein benutzerdefiniertes Präfix für die Job-ID zuweisen:

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

    BigQuery aufrufen

  2. Klicken Sie im Abfrageeditor auf Bearbeiten > Abfragemodus > Kontinuierliche Abfrage.

  3. Klicken Sie auf Bestätigen.

  4. Klicken Sie im Abfrageeditor auf Bearbeiten > Abfrageeinstellungen.

  5. Geben Sie im Abschnitt Präfix für benutzerdefinierte Job-ID ein benutzerdefiniertes Namenspräfix ein.

  6. Klicken Sie auf Speichern.

Zustandsorientierte Verarbeitung mit JOINs und Fensteraggregaten

Zustandsorientierte Vorgänge ermöglichen es kontinuierlichen Abfragen, komplexe Analysen durchzuführen, indem Informationen über mehrere Zeilen oder Zeitintervalle hinweg beibehalten werden. Zu diesen Vorgängen gehören JOINs und Windowing-Aggregationen.

Ausführliche Informationen zur Verwendung dieser zustandsorientierten Vorgänge finden Sie in den folgenden Themen:

  • Kontinuierliche AbfragenJOIN führen Echtzeitkorrelationen zwischen mehreren zeitorientierten Datenstreams durch.
  • Bei Windowing-Aggregationen werden Streamingdaten in konsistente Zeitintervalle gruppiert, um sie mit Aggregationsfunktionen zu analysieren.

Beispiele

Die folgenden SQL-Beispiele zeigen gängige Anwendungsfälle für kontinuierliche Abfragen.

Daten in ein Pub/Sub-Thema exportieren

Das folgende Beispiel zeigt eine kontinuierliche Abfrage, die Daten aus einer BigQuery-Tabelle filtert, die Informationen zu Streamingtaxifahrten empfängt, und die Daten für stornierte Fahrten in Echtzeit mit Nachrichtenattributen in einem Pub/Sub-Thema veröffentlicht:

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_comment AS STRING) AS passenger_comment))
  FROM
    CHANGES(TABLE `myproject.real_time_taxi_streaming.taxi_rides`,
      -- Configure the CHANGES 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 _CHANGE_TYPE = 'DELETE'
);

Daten in eine Bigtable-Tabelle exportieren

Das folgende Beispiel zeigt eine kontinuierliche Abfrage, die Daten aus einer BigQuery-Tabelle filtert, die Informationen zu Streamingtaxifahrten empfängt, und die Daten dann in Echtzeit in eine Bigtable-Tabelle exportiert:

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

Daten in eine Spanner-Tabelle exportieren

Das folgende Beispiel zeigt eine kontinuierliche Abfrage, die Daten aus einer BigQuery-Tabelle filtert, die Informationen zu Streamingtaxifahrten empfängt, und die Daten dann in Echtzeit in eine Spanner-Tabelle exportiert:

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,
    _CHANGE_TIMESTAMP as change_timestamp
  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'
  );

Daten in eine BigQuery-Tabelle oder in von Apache Iceberg verwaltete Tabellen schreiben

Das folgende Beispiel zeigt eine kontinuierliche Abfrage, die Daten aus einer BigQuery-Tabelle filtert und transformiert, die Informationen zu Streamingtaxifahrten empfängt, und die Daten dann in Echtzeit in eine andere BigQuery-Tabelle oder eine von Iceberg verwaltete Tabelle schreibt. Dadurch sind die Daten für weitere Analysen in BigQuery oder Ihrem Lakehouse mit offenem Format verfügbar.

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

Wenn Sie die Ausgabe einer kontinuierlichen Abfrage in eine verwaltete Iceberg-Tabelle schreiben möchten, ersetzen Sie myproject.real_time_taxi_streaming.transformed_taxirides durch den Namen Ihrer verwalteten Iceberg-Zieltabelle. Die Syntax ist identisch, da Verbindungsdetails und Lakehouse-Speichereinstellungen für die Zieltabelle selbst definiert sind.

Daten mit einem Gemini Enterprise Agent Platform-Modell verarbeiten

Das folgende Beispiel zeigt eine kontinuierliche Abfrage, die ein Agent Platform-Modell verwendet, um eine Anzeige für Taxigäste anhand des aktuellen Breiten- und Längengrads zu generieren. Anschließend werden die Ergebnisse in ein Pub/Sub-Thema in Echtzeit exportiert:

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,
        result)) AS message
  FROM
    AI.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))
      AS ml_output
);

JOINs und Fensteraggregationen ausführen

Das folgende Beispiel zeigt eine kontinuierliche Abfrage, die einen JOIN und Windowing-Aggregationen ausführt.

Angenommen, Sie möchten eine Tabelle mit Taxifahrten mit einer Tabelle mit Taxianfragen verknüpfen, um alle fünf Minuten die Taxiverfügbarkeit in den einzelnen Stadtteilen zu ermitteln. Mithilfe von Aggregatfunktionen können Sie das Taxinachfragevolumen pro Stadtteil und die minimale, maximale, durchschnittliche und Standardabweichungsentfernung eines Fahrgasts von einem Taxi erfassen, wenn er eine Fahrt angefordert hat.

INSERT INTO
 `real_time_taxi_streaming.neighborhood_taxi_health`
WITH potential_matches AS (
 SELECT
   requests._CHANGE_TIMESTAMP AS bq_changed_ts,
   requests.geohash,
   requests.latitude,
   requests.longitude,
   ST_DISTANCE(
     ST_GEOGPOINT(requests.longitude, requests.latitude),
     ST_GEOGPOINT(taxis.longitude, taxis.latitude)
   ) AS distance_in_meters
 FROM
   APPENDS(TABLE `real_time_taxi_streaming.ride_requests`,
     CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) AS requests
 INNER JOIN
   APPENDS(TABLE `real_time_taxi_streaming.taxirides`,
     CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) AS taxis
 ON requests.geohash = taxis.geohash
 WHERE
   taxis.ride_status = 'available'
   AND taxis._CHANGE_TIMESTAMP BETWEEN (requests._CHANGE_TIMESTAMP - INTERVAL 5 MINUTE) AND requests._CHANGE_TIMESTAMP
   AND ST_Dwithin(
     ST_GEOGPOINT(requests.longitude, requests.latitude),
     ST_GEOGPOINT(taxis.longitude, taxis.latitude),
     2000 -- Distance in meters
   )
)
SELECT
 window_end,
 geohash,
 ROUND(AVG(latitude), 6) AS avg_latitude,
 ROUND(AVG(longitude), 6) AS avg_longitude,
 COUNT(*) AS taxi_demand_volume,
 ROUND(AVG(distance_in_meters), 2) AS avg_proximity_meters,
 ROUND(MIN(distance_in_meters), 2) AS min_proximity_meters,
 ROUND(MAX(distance_in_meters), 2) AS max_proximity_meters,
 ROUND(STDDEV(distance_in_meters), 2) AS proximity_stddev
FROM
 TUMBLE(TABLE potential_matches, "bq_changed_ts", INTERVAL 5 MINUTE)
GROUP BY
 window_end,
 geohash;

SQL einer kontinuierlichen Abfrage ändern

Sie können die in einer kontinuierlichen Abfrage verwendete SQL nicht aktualisieren, während der Job für die kontinuierliche Abfrage ausgeführt wird. Sie müssen den kontinuierlichen Abfragejob abbrechen, die SQL-Anweisung ändern und dann einen neuen kontinuierlichen Abfragejob an der Stelle starten, an der Sie den ursprünglichen kontinuierlichen Abfragejob beendet haben.

So ändern Sie die SQL, die in einer kontinuierlichen Abfrage verwendet wird:

  1. Rufen Sie die Jobdetails für den Job mit kontinuierlicher Abfrage auf, den Sie aktualisieren möchten, und notieren Sie sich die Job-ID.
  2. Pausieren Sie nach Möglichkeit die Erfassung von Upstream-Daten. Wenn das nicht möglich ist, kann es beim Neustart der kontinuierlichen Abfrage zu einer Datenduplizierung kommen.
  3. Brechen Sie die kontinuierliche Abfrage ab, die Sie ändern möchten.

  4. Rufen Sie den Wert end_time für den ursprünglichen kontinuierlichen Abfragejob mithilfe der INFORMATION_SCHEMA-Ansicht JOBS ab:

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

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Ihre Projekt-ID.
    • REGION: die Region, die von Ihrem Projekt verwendet wird.
    • JOB_ID: die Job-ID der kontinuierlichen Abfrage, die Sie in Schritt 1 ermittelt haben.

  5. Ändern Sie die SQL-Anweisung für die kontinuierliche Abfrage so, dass die kontinuierliche Abfrage ab einem bestimmten Zeitpunkt gestartet wird. Verwenden Sie dabei den in Schritt 5 abgerufenen end_time-Wert als Ausgangspunkt.

  6. Ändern Sie die SQL-Anweisung für die kontinuierliche Abfrage so, dass sie die gewünschten Änderungen widerspiegelt.

  7. Führen Sie die geänderte kontinuierliche Abfrage aus.

Kontinuierliche Abfrage abbrechen

Sie können einen kontinuierlichen Abfragejob wie jeden anderen Job abbrechen. Es kann bis zu einer Minute dauern, bis die Abfrage nach dem Abbruch des Jobs beendet wird.

Wenn Sie eine Abfrage abbrechen und dann neu starten, verhält sich die neu gestartete Abfrage wie eine neue, unabhängige Abfrage. Die neu gestartete Abfrage beginnt nicht mit der Verarbeitung von Daten, bei denen der vorherige Job beendet wurde, und kann nicht auf die Ergebnisse der vorherigen Abfrage verweisen. Weitere Informationen finden Sie unter Kontinuierliche Abfrage ab einem bestimmten Zeitpunkt starten.

Abfragen überwachen und Fehler beheben

Eine kontinuierliche Abfrage kann durch Faktoren wie Dateninkonsistenzen, Schemaänderungen, vorübergehende Dienstunterbrechungen oder Wartungsarbeiten unterbrochen werden. BigQuery verarbeitet zwar einige vorübergehende Fehler, aber die folgenden Best Practices können die Jobstabilität verbessern:

Nächste Schritte