Crea query continue

Questo documento descrive come eseguire una query continua in BigQuery.

Le query continue di BigQuery sono istruzioni SQL che vengono eseguite continuamente ed elaborano i dati quasi in tempo reale. Puoi scrivere o esportare le righe di output prodotte da una query continua nelle seguenti destinazioni:

Scegli un tipo di account

Puoi creare ed eseguire un job di query continua utilizzando un account utente oppure puoi creare un job di query continua utilizzando un account utente ed eseguirlo utilizzando un service account. Devi utilizzare un account di servizio per eseguire una query continua che esporta i risultati in un argomento Pub/Sub.

Quando utilizzi un account utente, una query continua viene eseguita per un massimo di due giorni. Quando utilizzi un account di servizio, una query continua viene eseguita per un massimo di 150 giorni. Per saperne di più, consulta Autorizzazione.

Autorizzazioni obbligatorie

Questa sezione descrive le autorizzazioni necessarie per creare ed eseguire una query continua. In alternativa ai ruoli di Identity and Access Management (IAM) menzionati, puoi ottenere le autorizzazioni richieste tramite i ruoli personalizzati.

Autorizzazioni quando utilizzi un account utente

Questa sezione fornisce informazioni sui ruoli e sulle autorizzazioni necessari per creare ed eseguire una query continua utilizzando un account utente.

Per creare un job in BigQuery, l'account utente deve disporre dell'autorizzazione IAM bigquery.jobs.create. Ciascuno dei seguenti ruoli IAM concede l'autorizzazione bigquery.jobs.create:

Per esportare i dati da una tabella BigQuery, l'account utente deve disporre dell'autorizzazione IAM bigquery.tables.export. Ciascuno dei seguenti ruoli IAM concede l'autorizzazione bigquery.tables.export:

Per scrivere dati in una tabella BigQuery, l'account utente deve disporre dell'autorizzazione IAM bigquery.tables.updateData. Ciascuno dei seguenti ruoli IAM concede l'autorizzazione bigquery.tables.updateData:

Se l'account utente deve abilitare le API richieste per il tuo caso d'uso delle query continue, deve disporre del ruolo Amministratore utilizzo servizi (roles/serviceusage.serviceUsageAdmin).

Per scrivere dati nelle tabelle gestite Apache Iceberg, oltre alle autorizzazioni BigQuery, l'account utente deve disporre anche del ruolo BigQuery Connection User (roles/bigquery.connectionUser) per utilizzare la connessione alla risorsa Google Cloud associata alla tabella gestita Iceberg. Inoltre, il account di servizio associato a questa connessione deve disporre delle autorizzazioni appropriate (ad esempio roles/storage.objectUser e roles/storage.legacyBucketReader) sul bucket Cloud Storage sottostante. Per saperne di più, consulta Crea e utilizza tabelle gestite Apache Iceberg.

Autorizzazioni per l'utilizzo di un account di servizio

Questa sezione fornisce informazioni sui ruoli e sulle autorizzazioni richiesti dall'account utente che crea la query continua e dal account di servizio che la esegue.

Autorizzazioni dell'account utente

Per creare un job in BigQuery, l'account utente deve disporre dell'autorizzazione IAM bigquery.jobs.create. Ciascuno dei seguenti ruoli IAM concede l'autorizzazione bigquery.jobs.create:

Per inviare un job eseguito utilizzando un account di servizio, l'account utente deve disporre del ruolo Service Account User (roles/iam.serviceAccountUser). Se utilizzi lo stesso account utente per creare l'account di servizio, l'account utente deve disporre del ruolo Amministratore account di servizio (roles/iam.serviceAccountAdmin). Per informazioni su come limitare l'accesso di un utente a un singolo account di servizio, anziché a tutti i service account di un progetto, consulta Concedi un singolo ruolo.

Se l'account utente deve abilitare le API richieste per il tuo caso d'uso delle query continue, deve avere il ruolo Amministratore utilizzo servizi (roles/serviceusage.serviceUsageAdmin).

Autorizzazioni service account

Per esportare i dati da una tabella BigQuery, il account di servizio deve disporre dell'autorizzazione IAM bigquery.tables.export. Ciascuno dei seguenti ruoli IAM concede l'autorizzazione bigquery.tables.export:

Per scrivere dati in una tabella BigQuery, l'account di servizio deve disporre dell'autorizzazione IAM bigquery.tables.updateData. Ciascuno dei seguenti ruoli IAM concede l'autorizzazione bigquery.tables.updateData:

Per scrivere dati nelle tabelle gestite Apache Iceberg, oltre alle autorizzazioni BigQuery, il account di servizio deve disporre anche del ruolo BigQuery Connection User (roles/bigquery.connectionUser) per utilizzare la connessione alla risorsa Google Cloud associata alla tabella gestita Iceberg. Inoltre, il account di servizio associato a questa connessione deve disporre delle autorizzazioni appropriate (ad esempio roles/storage.objectUser e roles/storage.legacyBucketReader) sul bucket Cloud Storage sottostante. Per saperne di più, consulta Crea e utilizza tabelle gestite Apache Iceberg.

Prima di iniziare

  1. Nella console Google Cloud , nella pagina di selezione del progetto, seleziona o crea un progetto Google Cloud .

    Ruoli richiesti per selezionare o creare un progetto

    • Seleziona un progetto: la selezione di un progetto non richiede un ruolo IAM specifico. Puoi selezionare qualsiasi progetto per cui ti è stato concesso un ruolo.
    • Crea un progetto: per creare un progetto, devi disporre del ruolo Autore progetto (roles/resourcemanager.projectCreator), che contiene l'autorizzazione resourcemanager.projects.create. Scopri come concedere i ruoli.

    Vai al selettore di progetti

  2. Verifica che la fatturazione sia attivata per il tuo progetto Google Cloud .

  3. Abilita l'API BigQuery, se non è già abilitata.

    Ruoli richiesti per abilitare le API

    Per abilitare le API, devi disporre dell'autorizzazione serviceusage.services.enable. Se hai creato il progetto, probabilmente disponi già di questa autorizzazione tramite il ruolo Proprietario (roles/owner). In caso contrario, puoi ottenere questa autorizzazione tramite il ruolo Amministratore Service Usage (roles/serviceusage.serviceUsageAdmin). Scopri come concedere i ruoli.

    Abilitare l'API

Crea una prenotazione

Crea una prenotazione della versione Enterprise o Enterprise Plus, quindi crea un'assegnazione della prenotazione con un tipo di job CONTINUOUS. Questa prenotazione può utilizzare la scalabilità automatica e la condivisione degli slot inattivi. Esistono limitazioni delle prenotazioni che si applicano alle assegnazioni delle prenotazioni per le query continue.

Esporta in Pub/Sub

Per esportare i dati in Pub/Sub sono necessarie API, autorizzazioni IAM e risorse Google Cloud aggiuntive. Per saperne di più, consulta Esporta in Pub/Sub.

Elaborazione delle mutazioni con CHANGES

Quando esporti i dati in Pub/Sub, hai la possibilità di utilizzare la funzione di cronologia delle modifiche di CHANGES. La funzione CHANGES elabora tutte le righe che sono state modificate nella tabella di origine, incluse le aggiunte e le mutazioni.

Incorpora attributi personalizzati come metadati nei messaggi Pub/Sub

Puoi utilizzare gli attributi Pub/Sub per fornire informazioni aggiuntive sul messaggio, come priorità, origine, destinazione o metadati aggiuntivi. Puoi anche utilizzare gli attributi per filtrare i messaggi nella sottoscrizione.

All'interno di un risultato di una query continua, se una colonna è denominata _ATTRIBUTES, i suoi valori vengono copiati negli attributi del messaggio Pub/Sub. I campi forniti all'interno di _ATTRIBUTES vengono utilizzati come chiavi degli attributi.

La colonna _ATTRIBUTES deve essere di tipo JSON, nel formato ARRAY<STRUCT<STRING, STRING>> o STRUCT<STRING>.

Per un esempio, consulta Esporta i dati in un argomento Pub/Sub.

Esporta in Bigtable

Per esportare i dati in Bigtable sono necessarie API, autorizzazioni IAM e risorse Google Cloudaggiuntive. Per saperne di più, consulta Esporta in Bigtable.

Esporta in Spanner

Per esportare i dati in Spanner sono necessarie API, autorizzazioni IAM e risorse Google Cloudaggiuntive. Per saperne di più, consulta Esporta in Spanner (ETL inverso).

Scrivi i dati in una tabella BigQuery o in tabelle gestite di Apache Iceberg

Puoi scrivere dati in una tabella BigQuery o in una tabella gestita di Iceberg utilizzando un'istruzione INSERT.

Quando scrivi l'output di una query continua in una tabella Iceberg gestita:

  • La tabella gestita Iceberg di destinazione deve già esistere prima di creare la query continua. Per istruzioni sulla creazione di una tabella gestita Iceberg, consulta Crea e utilizza tabelle gestite Apache Iceberg.
  • La sintassi della query è identica a quella per la scrittura in una tabella BigQuery standard. Puoi specificare la tabella di destinazione nella clausola INSERT INTO e non è necessario specificare le opzioni di connessione o bucket nell'istruzione SQL.
  • L'account utente o il account di servizio che esegue la query continua deve disporre delle autorizzazioni richieste sulla tabella di destinazione, sulla connessione alla risorsa Google Cloud e sul bucket Cloud Storage sottostante. Per saperne di più, consulta Autorizzazioni richieste.

Per un esempio, consulta Scrivi dati in una tabella BigQuery o in tabelle gestite di Apache Iceberg.

Utilizza le funzioni AI

Per utilizzare una funzione AI supportata in una query continua sono necessarie API, autorizzazioni IAM e risorse Google Cloudaggiuntive. Per saperne di più, consulta uno dei seguenti argomenti, a seconda del tuo caso d'uso:

Quando utilizzi una funzione AI in una query continua, valuta se l'output della query rimarrà entro la quota della funzione. Se superi la quota, potresti dover gestire separatamente i record che non vengono elaborati.

Specifica un'ora di inizio per la query continua

Devi utilizzare la funzione APPENDS o la funzione CHANGES nel caso di esportazioni in Pub/Sub, nella clausola FROM di una query continua per specificare i dati meno recenti da elaborare. Ad esempio, APPENDS(TABLE my_table, start_timestamp).

L'argomento start_timestamp definisce il momento in cui la query continua inizia a elaborare i dati. Ad esempio, APPENDS(TABLE my_table, CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) indica a BigQuery di elaborare i dati aggiunti alla tabella my_table al massimo 10 minuti prima dell'inizio della query continua. I dati aggiunti successivamente a my_table vengono elaborati man mano che arrivano. Non è previsto alcun ritardo nel trattamento dei dati.

Quando specifichi l'argomento start_timestamp, il valore deve rientrare nella finestra di spostamento cronologico della tabella, che per impostazione predefinita è di sette giorni per le tabelle standard. Se imposti start_timestamp su NULL, il valore predefinito è l'ora di creazione della tabella. L'utilizzo di un valore NULL non è consigliato perché viene restituito un errore se la tabella è stata creata prima della finestra di tempo di viaggio. Una query che utilizza NULL potrebbe riuscire per una tabella appena creata, ma non riuscire in un secondo momento se il timestamp di creazione della tabella non rientra nella finestra di sette giorni.

Non fornire un argomento end_timestamp alla funzione APPENDS quando la utilizzi in una query continua.

L'esempio seguente mostra come avviare una query continua da un determinato momento utilizzando la funzione APPENDS, quando esegui una query su una tabella BigQuery che riceve informazioni sulle corse in taxi in 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');

Specifica un punto di partenza precedente alla finestra di viaggio nel tempo

Per includere dati che non rientrano nella finestra di cronologia di sette giorni, utilizza una query standard per eseguire il backfill dei dati fino a un determinato momento, quindi avvia una query continua da quel momento. Il timestamp di "passaggio" deve rientrare nella finestra di sette giorni in modo che la query continua possa riprendere da dove si era interrotta la query standard.

L'esempio seguente mostra come eseguire il backfill dei dati meno recenti da una tabella BigQuery che riceve informazioni sulle corse in taxi in modalità flusso e poi passare a una query continua.

  1. Esegui una query standard per eseguire il backfill dei dati fino a un momento specifico:

    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. Esegui una query continua dal momento in cui la query è stata interrotta:

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

Esegui una query continua utilizzando un account utente

Questa sezione descrive come eseguire una query continua utilizzando un account utente. Una volta avviata la query continua, puoi chiudere la console Google Cloud , la finestra del terminale o l'applicazione senza interrompere l'esecuzione della query. Una query continua eseguita da un account utente viene eseguita per un massimo di due giorni, dopodiché si interrompe automaticamente. Per continuare a elaborare i nuovi dati in arrivo, avvia una nuova query continua e specifica un punto di partenza. Per automatizzare questo processo, consulta la sezione Riprova le query non riuscite.

Per eseguire una query continua:

Console

  1. Nella console Google Cloud , vai alla pagina BigQuery.

    Vai a BigQuery

  2. Nell'editor di query, fai clic su Modifica > Modalità query > Query continua. Per confermare questa scelta, fai clic su Conferma.

  3. (Facoltativo) Per controllare la durata dell'esecuzione della query, fai clic su Modifica > Impostazioni query e imposta il Timeout job in millisecondi.

  4. Nell'editor di query, digita l'istruzione SQL per la query continua. L'istruzione SQL deve contenere solo operazioni supportate.

  5. Fai clic su Esegui.

bq

  1. Nella console Google Cloud , attiva Cloud Shell.

    Attiva Cloud Shell

    Nella parte inferiore della console Google Cloud , viene avviata una sessione di Cloud Shell e viene visualizzato un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

  2. In Cloud Shell, esegui la query continua utilizzando il comando bq query con il flag --continuous:

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

    Sostituisci QUERY con l'istruzione SQL per la query continua. L'istruzione SQL deve contenere solo operazioni supportate. Puoi controllare la durata della query utilizzando il flag --job_timeout_ms.

API

Esegui la query continua chiamando il metodo jobs.insert. Devi impostare il campo continuous su true nel JobConfigurationQuery della risorsa Job che passi. Facoltativamente, puoi controllare la durata della query impostando il campo jobTimeoutMs.

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

Sostituisci quanto segue:

  • PROJECT_ID: il tuo ID progetto.
  • QUERY: l'istruzione SQL per la query continua. L'istruzione SQL deve contenere solo operazioni supportate.

Esegui una query continua utilizzando un account di servizio

Questa sezione descrive come eseguire una query continua utilizzando un account di servizio. Una volta avviata la query continua, puoi chiudere la console Google Cloud , la finestra del terminale o l'applicazione senza interrompere l'esecuzione della query. Una query continua eseguita utilizzando un account di servizio può essere eseguita per un massimo di 150 giorni, dopodiché si interrompe automaticamente. Per continuare a elaborare i nuovi dati in arrivo, avvia una nuova query continua e specifica un punto di partenza. Per automatizzare questo processo, consulta la sezione Riprova le query non riuscite.

Se utilizzi un account di servizio per eseguire una query continua tra progetti, devi aggiornare la policy dell'organizzazione per consentire il collegamento dei service account di un progetto alle risorse di un altro progetto. Per saperne di più, consulta Abilita il collegamento dei service account tra progetti.

Segui questi passaggi per utilizzare un account di servizio per eseguire una query continua:

Console

  1. Crea un account di servizio.
  2. Concedi le autorizzazioni richieste al account di servizio.

  3. Nella console Google Cloud , vai alla pagina BigQuery.

    Vai a BigQuery

  4. Nell'editor di query, fai clic su Modifica > Modalità query > Query continua. Per confermare questa scelta, fai clic su Conferma.

  5. Nell'editor di query, fai clic su Modifica > Impostazioni query.

  6. Nella sezione Query continua, utilizza la casella Service account per selezionare il account di servizio che hai creato.

  7. (Facoltativo) Per controllare la durata dell'esecuzione della query, imposta il Timeout job in millisecondi.

  8. Fai clic su Salva.

  9. Nell'editor di query, digita l'istruzione SQL per la query continua. L'istruzione SQL deve contenere solo operazioni supportate.

  10. Fai clic su Esegui.

bq

  1. Crea un account di servizio.
  2. Concedi le autorizzazioni richieste al account di servizio.

  3. Nella console Google Cloud , attiva Cloud Shell.

    Attiva Cloud Shell

    Nella parte inferiore della console Google Cloud , viene avviata una sessione di Cloud Shell e viene visualizzato un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

  4. Nella riga di comando, esegui la query continua utilizzando il comando bq query con i seguenti flag:

    • Imposta il flag --continuous su true per rendere la query continua.
    • Utilizza il flag --connection_property per specificare un account di servizio da utilizzare.
    • (Facoltativo) Imposta il flag --job_timeout_ms per limitare il runtime della query.
    bq query --project_id=PROJECT_ID --use_legacy_sql=false \
--continuous=true --connection_property=service_account=SERVICE_ACCOUNT_EMAIL \
'QUERY'

    Sostituisci quanto segue:

    • PROJECT_ID: il tuo ID progetto.
    • SERVICE_ACCOUNT_EMAIL: l'indirizzo email del service account. Puoi ottenere l'indirizzo email del account di servizio dalla pagina Service account della console Google Cloud .
    • QUERY: l'istruzione SQL per la query continua. L'istruzione SQL deve contenere solo operazioni supportate.

API

  1. Crea un account di servizio.
  2. Concedi le autorizzazioni richieste al account di servizio.

  3. Esegui la query continua chiamando il metodo jobs.insert. Imposta i seguenti campi nella risorsa JobConfigurationQuery della risorsa Job che passi:

    • Imposta il campo continuous su true per rendere la query continua.
    • Utilizza il campo connectionProperties per specificare un account di servizio da utilizzare.

    Facoltativamente, puoi controllare la durata della query impostando il campo jobTimeoutMs nella risorsa 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

    Sostituisci quanto segue:

    • PROJECT_ID: il tuo ID progetto.
    • QUERY: l'istruzione SQL per la query continua. L'istruzione SQL deve contenere solo operazioni supportate.
    • SERVICE_ACCOUNT_EMAIL: l'indirizzo email del service account. Puoi trovare l'indirizzo email del account di servizio nella pagina Service account della console Google Cloud .

Crea un ID job personalizzato

A ogni job di query viene assegnato un ID job che puoi utilizzare per cercare e gestire il job. Per impostazione predefinita, gli ID job vengono generati in modo casuale. Per semplificare la ricerca dell'ID job di una query continua utilizzando la cronologia dei job o l'explorer dei job, puoi assegnare un prefisso ID job personalizzato:

  1. Nella console Google Cloud , vai alla pagina BigQuery.

    Vai a BigQuery

  2. Nell'editor di query, fai clic su Modifica > Modalità di query > Query continua.

  3. Fai clic su Conferma.

  4. Nell'editor di query, fai clic su Modifica > Impostazioni query.

  5. Nella sezione Prefisso ID job personalizzato, inserisci un prefisso per il nome personalizzato.

  6. Fai clic su Salva.

Elaborazione stateful con JOIN e aggregazioni di windowing

Le operazioni stateful consentono alle query continue di eseguire analisi complesse conservando le informazioni su più righe o intervalli di tempo. Queste operazioni includono JOIN e aggregazioni di windowing.

Per informazioni dettagliate su come utilizzare queste operazioni stateful, consulta i seguenti argomenti:

  • Le query continue JOIN eseguono correlazioni in tempo reale tra più stream di dati orientati al tempo.
  • Le aggregazioni di windowing raggruppano i dati in streaming in intervalli di tempo coerenti per l'analisi utilizzando funzioni di aggregazione.

Esempi

I seguenti esempi SQL mostrano casi d'uso comuni per le query continue.

Esporta i dati in un argomento Pub/Sub

L'esempio seguente mostra una query continua che filtra i dati di una tabella BigQuery che riceve informazioni in streaming sulle corse in taxi e pubblica i dati delle corse annullate in un argomento Pub/Sub in tempo reale con attributi dei messaggi:

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

Esporta i dati in una tabella Bigtable

L'esempio seguente mostra una query continua che filtra i dati di una tabella BigQuery che riceve informazioni in streaming sulle corse in taxi ed esporta i dati in una tabella Bigtable in tempo reale:

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

Esporta i dati in una tabella Spanner

L'esempio seguente mostra una query continua che filtra i dati di una tabella BigQuery che riceve informazioni in streaming sulle corse in taxi, quindi esporta i dati in una tabella Spanner in tempo reale:

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

Scrivi i dati in una tabella BigQuery o in tabelle gestite di Apache Iceberg

L'esempio seguente mostra una query continua che filtra e trasforma i dati di una tabella BigQuery che riceve informazioni in streaming sulle corse in taxi, quindi scrive i dati in un'altra tabella BigQuery o in una tabella gestita da Iceberg in tempo reale. In questo modo, i dati sono disponibili per ulteriori analisi downstream in BigQuery o nel tuo lakehouse in formato aperto.

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

Per scrivere l'output della query continua in una tabella Iceberg gestita, sostituisci myproject.real_time_taxi_streaming.transformed_taxirides con il nome della tabella Iceberg gestita di destinazione. La sintassi è identica perché i dettagli della connessione e le impostazioni di archiviazione del lakehouse sono definiti nella tabella di destinazione stessa.

Elabora i dati utilizzando un modello Gemini Enterprise Agent Platform

L'esempio seguente mostra una query continua che utilizza un modello Agent Platform per generare un annuncio per i passeggeri di taxi in base alla loro latitudine e longitudine attuali, quindi esporta i risultati in un argomento Pub/Sub in tempo reale:

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

Esegui JOIN e aggregazioni di windowing

L'esempio seguente mostra una query continua che esegue un JOIN e aggregazioni di windowing.

Supponiamo che tu voglia unire una tabella delle corse in taxi a una tabella delle richieste di taxi per comprendere lo stato dei taxi in ogni quartiere ogni cinque minuti. Utilizzando le funzioni di aggregazione, puoi acquisire il volume della domanda di taxi per quartiere e la distanza minima, massima, media e la deviazione standard di un passeggero da un taxi quando ha richiesto una corsa.

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;

Modifica l'SQL di una query continua

Non puoi aggiornare l'SQL utilizzato in una query continua mentre il job di query continua è in esecuzione. Devi annullare il job di query continua, modificare l'SQL e poi avviare un nuovo job di query continua dal punto in cui hai interrotto il job di query continua originale.

Per modificare l'SQL utilizzato in una query continua:

  1. Visualizza i dettagli del job per il job di query continua che vuoi aggiornare e prendi nota dell'ID job.
  2. Se possibile, metti in pausa la raccolta dei dati upstream. Se non puoi farlo, potresti riscontrare una duplicazione dei dati quando la query continua viene riavviata.
  3. Annulla la query continua che vuoi modificare.

  4. Ottieni il valore end_time per il job di query continua originale utilizzando la 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';

    Sostituisci quanto segue:

    • PROJECT_ID: il tuo ID progetto.
    • REGION: la regione utilizzata dal progetto.
    • JOB_ID: l'ID job di query continua che hai identificato nel passaggio 1.

  5. Modifica l'istruzione SQL della query continua per avviare la query continua da un momento specifico, utilizzando il valore end_time recuperato nel passaggio 5 come punto di partenza.

  6. Modifica l'istruzione SQL della query continua in modo che rifletta le modifiche necessarie.

  7. Esegui la query continua modificata.

Annulla una query continua

Puoi annullare un job di query continua come qualsiasi altro job. Potrebbe essere necessario fino a un minuto prima che la query smetta di essere eseguita dopo l'annullamento del job.

Se annulli e poi riavvii una query, la query riavviata si comporta come una nuova query indipendente. La query riavviata non inizia a elaborare i dati dal punto in cui il job precedente si è interrotto e non può fare riferimento ai risultati della query precedente. Consulta Avvia una query continua da un momento specifico.

Monitora le query e gestisci gli errori

Una query continua potrebbe essere interrotta a causa di fattori quali incoerenze dei dati, modifiche dello schema, interruzioni temporanee del servizio o manutenzione. Sebbene BigQuery gestisca alcuni errori temporanei, le best practice per migliorare la resilienza dei job includono:

Passaggi successivi