Esporta i dati in Pub/Sub (ETL inverso)

L'estrazione, la trasformazione e il caricamento (RETL) inversi da BigQuery a Pub/Sub ti consentono di fornire dati ad applicazioni e servizi downstream in modo basato sugli eventi. Questo workflow combina le funzionalità di analisi di BigQuery con il servizio di messaggistica globale asincrono e scalabile di Pub/Sub.

Per esportare i dati in Pub/Sub, utilizza l'istruzione EXPORT DATA in una query continua per esportare i dati da BigQuery a un argomento Pub/Sub.

Prerequisiti

Devi creare un service account. Per eseguire una query continua che esporta i risultati in un argomento Pub/Sub è necessario un service account.

Devi creare un argomento Pub/Sub per ricevere i risultati delle query continue come messaggi e una sottoscrizione Pub/Sub che l'applicazione di destinazione può utilizzare per ricevere i messaggi.

Ruoli obbligatori

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:

• BigQuery User (roles/bigquery.user)
• BigQuery Job User (roles/bigquery.jobUser)
• BigQuery Admin (roles/bigquery.admin)

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 disporre del 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:

• BigQuery Data Viewer (roles/bigquery.dataViewer)
• BigQuery Data Editor (roles/bigquery.dataEditor)
• BigQuery Data Owner (roles/bigquery.dataOwner)
• BigQuery Admin (roles/bigquery.admin)

Affinché il account di servizio possa accedere a Pub/Sub, devi concedergli entrambi i seguenti ruoli IAM:

• Pub/Sub Viewer (roles/pubsub.viewer)
• Pub/Sub Publisher (roles/pubsub.publisher)

Potresti anche riuscire a ottenere le autorizzazioni richieste tramite i ruoli personalizzati.

Prima di iniziare

Abilita le API BigQuery e Pub/Sub, se non sono già abilitate.

Ruoli richiesti per abilitare le API

Per abilitare le API, devi disporre dell'autorizzazione serviceusage.services.enable. Se hai creato il progetto, probabilmente hai già 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.

Abilita le API

Esporta in Pub/Sub

Utilizza l'istruzione EXPORT DATA per esportare i dati in un argomento Pub/Sub:

Console

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

   Vai a BigQuery

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

3. Nella sezione Query continua, seleziona la casella di controllo Utilizza la modalità di query continua.

4. Nella casella Service account, seleziona il account di servizio che hai creato.

5. Fai clic su Salva.

6. Nell'editor di query, inserisci la seguente istruzione:

   EXPORT DATA
   OPTIONS (
   format = 'CLOUD_PUBSUB',
   uri = 'https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID'
   ) AS
   (
   QUERY
   );

   Sostituisci quanto segue:

   • PROJECT_ID: il tuo ID progetto.
   • TOPIC_ID: l'ID argomento Pub/Sub. Puoi ottenere l'ID argomento dalla pagina Argomenti della console Google Cloud .
   • QUERY: l'istruzione SQL per selezionare i dati da esportare. L'istruzione SQL deve contenere solo operazioni supportate. Devi utilizzare la funzione APPENDS nella clausola FROM di una query continua per specificare il momento in cui iniziare l'elaborazione dei dati.

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

   bq query --project_id=PROJECT_ID --use_legacy_sql=false \
   --continuous=true --connection_property=service_account=SERVICE_ACCOUNT_EMAIL \
   'EXPORT DATA OPTIONS (format = "CLOUD_PUBSUB", uri = "https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID") AS (QUERY);'

   Sostituisci quanto segue:

   • PROJECT_ID: il tuo ID progetto.
   • 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 .
   • QUERY: l'istruzione SQL per selezionare i dati da esportare. L'istruzione SQL deve contenere solo operazioni supportate. Devi utilizzare la funzione APPENDS nella clausola FROM di una query continua per specificare il momento in cui iniziare l'elaborazione dei dati.

API

1. 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 connection_property per specificare un account di servizio da utilizzare.

   curl --request POST \
     'https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs'
     --header 'Authorization: Bearer $(gcloud auth print-access-token) \
     --header 'Accept: application/json' \
     --header 'Content-Type: application/json' \
     --data '("configuration":("query":"EXPORT DATA OPTIONS (format = 'CLOUD_PUBSUB', uri = 'https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID') AS (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 selezionare i dati da esportare. L'istruzione SQL deve contenere solo operazioni supportate. Devi utilizzare la funzione APPENDS nella clausola FROM di una query continua per specificare il momento in cui iniziare l'elaborazione dei dati.
   • 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 .

Esporta più colonne in Pub/Sub

Se vuoi includere più colonne nell'output, puoi creare una colonna struct per contenere i valori della colonna, quindi convertire il valore struct in una stringa JSON utilizzando la funzione TO_JSON_STRING. L'esempio seguente esporta i dati da quattro colonne, formattate come stringa JSON:

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.taxi_rides`,
      -- 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'
);

Ottimizzazione dell'esportazione

Se le prestazioni del job di query continua sembrano essere limitate dalle risorse di computing disponibili, prova ad aumentare le dimensioni dell'assegnazione della prenotazione di slotCONTINUOUS di BigQuery.

Limitazioni

• I dati esportati devono essere costituiti da una singola colonna STRING o BYTES. Il nome della colonna può essere quello che preferisci.
• Per esportare in Pub/Sub, devi utilizzare una query continua.
• Non puoi passare uno schema a un argomento Pub/Sub nella query continua.
• Non puoi esportare dati in un argomento Pub/Sub che utilizza uno schema.
• Quando esporti in Pub/Sub, puoi esportare record in formato JSON in cui alcuni valori sono NULL, ma non puoi esportare record costituiti solo da valori NULL. Puoi escludere i record NULL dai risultati della query includendo un filtro WHERE message IS NOT NULL nella query continua.
• Quando esporti i dati in un argomento Pub/Sub configurato con un endpoint di località, l'endpoint deve essere configurato all'interno dello stesso confine regionale Google Cloud del set di dati BigQuery che contiene la tabella su cui stai eseguendo la query.
• I dati esportati non devono superare le quote di Pub/Sub.

Prezzi

Quando esporti i dati in una query continua, ti vengono addebitati i costi in base ai prezzi di calcolo della capacità di BigQuery. Per eseguire query continue, devi disporre di una prenotazione che utilizzi la versione Enterprise o Enterprise Plus e di un'assegnazione della prenotazione che utilizzi il tipo di job CONTINUOUS.

Dopo l'esportazione dei dati, ti verrà addebitato l'utilizzo di Pub/Sub. Per saperne di più, consulta i prezzi di Pub/Sub.