Questo tutorial illustra come utilizzare Dataflow SQL per unire uno stream di dati di Pub/Sub con i dati di una tabella BigQuery.
Obiettivi
In questo tutorial imparerai a:
- Scrivi una query SQL Dataflow che unisce i flussi di dati Pub/Sub con dati della tabella BigQuery.
- Esegui il deployment di un job Dataflow dall'interfaccia utente di Dataflow SQL.
Costi
In questo documento utilizzi i seguenti componenti fatturabili di Google Cloud:
- Dataflow
- Cloud Storage
- Pub/Sub
- Data Catalog
Per generare una stima dei costi basata sull'utilizzo previsto,
utilizza il Calcolatore prezzi.
Prima di iniziare
- Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
Make sure that billing is enabled for your Google Cloud project.
-
Enable the Cloud Dataflow, Compute Engine, Logging, Cloud Storage, Cloud Storage JSON, BigQuery, Cloud Pub/Sub, Cloud Resource Manager and Data Catalog. APIs.
-
Create a service account:
-
In the Google Cloud console, go to the Create service account page.
Go to Create service account - Select your project.
-
In the Service account name field, enter a name. The Google Cloud console fills in the Service account ID field based on this name.
In the Service account description field, enter a description. For example,
Service account for quickstart
. - Click Create and continue.
-
Grant the Project > Owner role to the service account.
To grant the role, find the Select a role list, then select Project > Owner.
- Click Continue.
-
Click Done to finish creating the service account.
Do not close your browser window. You will use it in the next step.
-
-
Create a service account key:
- In the Google Cloud console, click the email address for the service account that you created.
- Click Keys.
- Click Add key, and then click Create new key.
- Click Create. A JSON key file is downloaded to your computer.
- Click Close.
-
Set the environment variable
GOOGLE_APPLICATION_CREDENTIALS
to the path of the JSON file that contains your credentials. This variable applies only to your current shell session, so if you open a new session, set the variable again. -
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
Make sure that billing is enabled for your Google Cloud project.
-
Enable the Cloud Dataflow, Compute Engine, Logging, Cloud Storage, Cloud Storage JSON, BigQuery, Cloud Pub/Sub, Cloud Resource Manager and Data Catalog. APIs.
-
Create a service account:
-
In the Google Cloud console, go to the Create service account page.
Go to Create service account - Select your project.
-
In the Service account name field, enter a name. The Google Cloud console fills in the Service account ID field based on this name.
In the Service account description field, enter a description. For example,
Service account for quickstart
. - Click Create and continue.
-
Grant the Project > Owner role to the service account.
To grant the role, find the Select a role list, then select Project > Owner.
- Click Continue.
-
Click Done to finish creating the service account.
Do not close your browser window. You will use it in the next step.
-
-
Create a service account key:
- In the Google Cloud console, click the email address for the service account that you created.
- Click Keys.
- Click Add key, and then click Create new key.
- Click Create. A JSON key file is downloaded to your computer.
- Click Close.
-
Set the environment variable
GOOGLE_APPLICATION_CREDENTIALS
to the path of the JSON file that contains your credentials. This variable applies only to your current shell session, so if you open a new session, set the variable again. - Installa e inizializza l'interfaccia a riga di comando gcloud. Scegli una delle
opzioni di installazione.
Potresti dover impostare la proprietà
project
nel progetto che utilizzi per questa procedura dettagliata. - Vai all'interfaccia utente web di Dataflow SQL nella console Google Cloud. Questo
apre il progetto a cui hai eseguito l'accesso più di recente. Per passare a un'altra
progetto, fai clic sul nome del progetto nella parte superiore della
nell'interfaccia utente web di Dataflow SQL e cerca il progetto che vuoi utilizzare.
Vai all'interfaccia utente web di Dataflow SQL
Crea origini di esempio
Se vuoi seguire l'esempio fornito in questo tutorial, crea il le seguenti fonti e usale nei passaggi del tutorial.
- Un argomento Pub/Sub denominato
transactions
: uno stream di dati sulle transazioni che arriva tramite una sottoscrizione all'argomento Pub/Sub. I dati di ogni transazione includono informazioni come il prodotto acquistato, il prezzo scontato, nonché la città e lo stato in cui si è verificato un errore. Dopo aver creato l'argomento Pub/Sub, crei uno script che pubblica messaggi nel tuo argomento. Eseguirai questo script in una sezione successiva di questo tutorial. - Una tabella BigQuery denominata
us_state_salesregions
: una tabella che fornisce una mappatura degli stati alle regioni di vendita. Prima di crearlo devi creare un set di dati BigQuery.
Assegna uno schema all'argomento Pub/Sub
L'assegnazione di uno schema consente di eseguire query SQL sull'argomento Pub/Sub e i dati di Google Cloud. Al momento, Dataflow SQL si aspetta che i messaggi negli argomenti Pub/Sub vengano serializzati in formato JSON.
Per assegnare uno schema
argomento Pub/Sub di esempio transactions
:
Crea un file di testo e assegnagli il nome
transactions_schema.yaml
. Copia e incolla il seguente testo dello schema intransactions_schema.yaml
.- column: event_timestamp description: Pub/Sub event timestamp mode: REQUIRED type: TIMESTAMP - column: tr_time_str description: Transaction time string mode: NULLABLE type: STRING - column: first_name description: First name mode: NULLABLE type: STRING - column: last_name description: Last name mode: NULLABLE type: STRING - column: city description: City mode: NULLABLE type: STRING - column: state description: State mode: NULLABLE type: STRING - column: product description: Product mode: NULLABLE type: STRING - column: amount description: Amount of transaction mode: NULLABLE type: FLOAT
Assegna lo schema utilizzando Google Cloud CLI.
a. Aggiorna gcloud CLI con seguente comando. Assicurati che la versione di gcloud CLI sia 242.0.0 o successiva.
gcloud components update
b. Esegui il seguente comando in una finestra della riga di comando. Sostituisci project-id con il tuo ID progetto e path-to-file con il percorso di
transactions_schema.yaml
.gcloud data-catalog entries update \ --lookup-entry='pubsub.topic.`project-id`.transactions' \ --schema-from-file=path-to-file/transactions_schema.yaml
Per ulteriori informazioni sui parametri del comando e sui formati dei file dello schema consentiti, consulta la pagina della documentazione di gcloud data-catalog entries update.
c. Verifica che lo schema sia stato assegnato correttamente all'argomento
transactions
Pub/Sub. Sostituisci project-id con il tuo progetto ID.gcloud data-catalog entries lookup 'pubsub.topic.`project-id`.transactions'
Trova origini Pub/Sub
L'interfaccia utente di Dataflow SQL offre un modo per trovare gli oggetti di origine dati Pub/Sub per qualsiasi progetto a cui hai accesso, quindi non devi ricordare i relativi nomi completi.
Per l'esempio in questo tutorial, vai all'editor Dataflow SQL e cerca l'argomento Pub/Sub transactions
che hai creato:
Vai ad Area di lavoro SQL.
Nel riquadro Editor SQL Dataflow, cerca
projectid=project-id transactions
nella barra di ricerca. Sostituisci project-id con l'ID del tuo progetto.
Visualizza lo schema
- Nel riquadro Editor Dataflow SQL dell'interfaccia utente di Dataflow SQL, fai clic su transactions o cerca un argomento Pub/Sub digitando
projectid=project-id system=cloud_pubsub
e selezionalo. In Schema, puoi visualizzare lo schema assegnato all'argomento Pub/Sub.
Crea una query SQL
L'interfaccia utente SQL di Dataflow consente di creare query SQL per eseguire di job Dataflow.
La seguente query SQL è una query di arricchimento dei dati. Aggiunge un campo aggiuntivo,sales_region
, allo stream di eventi Pub/Sub (transactions
), utilizzando una tabella BigQuery (us_state_salesregions
) che mappa gli stati alle regioni di vendita.
Copia e incolla la seguente query SQL nell'Editor query. Sostituisci project-id con l'ID del tuo progetto.
SELECT tr.*, sr.sales_region FROM pubsub.topic.`project-id`.transactions as tr INNER JOIN bigquery.table.`project-id`.dataflow_sql_tutorial.us_state_salesregions AS sr ON tr.state = sr.state_code
Quando inserisci una query nell'interfaccia utente di Dataflow SQL, il convalidatore della query verifica la sintassi della query. Se la query è valida, viene visualizzata un'icona con un segno di spunta verde. Se la query non è valida, viene visualizzata un'icona rossa con punto esclamativo. Se la sintassi della query è non valido, facendo clic sull'icona dello validator vengono fornite informazioni su cosa che devi correggere.
Il seguente screenshot mostra la query valida nell'Editor di query. Il convalidatore mostra un segno di spunta verde.
Crea un job Dataflow per eseguire la query SQL
Per eseguire la query SQL, crea un job Dataflow dal UI di Dataflow SQL.
Nell'Editor di query, fai clic su Crea job.
Nel riquadro Crea job Dataflow che si apre:
- Per Destinazione, seleziona BigQuery.
- In ID set di dati, seleziona
dataflow_sql_tutorial
. - In Nome tabella, inserisci
sales
.
(Facoltativo) Dataflow sceglie automaticamente le impostazioni che sono ottimale per il tuo job SQL di Dataflow, ma puoi espandere Menu Parametri facoltativi per specificare manualmente quanto segue. opzioni pipeline:
- Numero massimo di worker
- Zona
- Email del service account
- Tipo di macchina
- Esperimenti aggiuntivi
- Configurazione dell'indirizzo IP del worker
- Rete
- Subnet
Fai clic su Crea. L'avvio del job Dataflow richiede alcuni minuti.
Visualizza il job Dataflow
Dataflow trasforma la query SQL in una pipeline Apache Beam. Fai clic su Visualizza job per aprirlo la UI web di Dataflow, dove puoi vedere una rappresentazione grafica della pipeline.
Per visualizzare un'analisi dettagliata delle trasformazioni che avvengono della pipeline, fai clic sulle caselle. Ad esempio, se fai clic sulla prima casella nella rappresentazione grafica, etichettata come Esegui query SQL, viene visualizzata un'immagine che mostra le operazioni in esecuzione dietro le quinte.
Le prime due caselle rappresentano i due input che hai unito:
argomento Pub/Sub, transactions
e BigQuery
tabella us_state_salesregions
.
Per visualizzare la tabella di output che contiene i risultati del job, vai alla
UI di BigQuery.
Nel riquadro Explorer, nel tuo progetto, fai clic su
il set di dati dataflow_sql_tutorial
che hai creato. Quindi, fai clic sulla tabella di output
sales
. La scheda Anteprima mostra i contenuti della tabella di output.
Visualizza le offerte di lavoro precedenti e modifica le query
L'interfaccia utente di Dataflow archivia le query e i job precedenti nella pagina Job di Dataflow.
Puoi utilizzare l'elenco della cronologia dei job per visualizzare le query SQL precedenti. Ad esempio, vuoi modificare la query per aggregare le vendite per regione di vendita ogni 15 secondi. Utilizza la pagina Job per accedere al job in esecuzione che hai avviato in precedenza nel tutorial, copia la query SQL ed esegui un altro job con una query modificata.
Nella pagina Job di Dataflow, fai clic sul job da modificare.
Nella pagina Dettagli job, nel riquadro Informazioni job, individua la query SQL in Opzioni pipeline. Trova la riga relativa a queryString.
Copia e incolla la seguente query SQL nell'editor Dataflow SQL in Spazio di lavoro SQL per aggiungere finestre con scorrimento. Sostituisci project-id con l'ID del tuo progetto.
SELECT sr.sales_region, TUMBLE_START("INTERVAL 15 SECOND") AS period_start, SUM(tr.amount) as amount FROM pubsub.topic.`project-id`.transactions AS tr INNER JOIN bigquery.table.`project-id`.dataflow_sql_tutorial.us_state_salesregions AS sr ON tr.state = sr.state_code GROUP BY sr.sales_region, TUMBLE(tr.event_timestamp, "INTERVAL 15 SECOND")
Fai clic su Crea job per creare un nuovo job con la query modificata.
Esegui la pulizia
Per evitare che al tuo account di fatturazione Cloud vengano addebitati costi relativi alle risorse utilizzate in questo tutorial:
Interrompi lo script di pubblicazione di
transactions_injector.py
, se è ancora in esecuzione.Interrompi i job Dataflow in esecuzione. Vai all'UI web di Dataflow nella console Google Cloud.
Vai all'interfaccia utente web di Dataflow
Per ogni job creato seguendo questa procedura dettagliata, svolgi i seguenti passaggi:
Fai clic sul nome del job.
Nella pagina Dettagli job, fai clic su Arresta. Viene visualizzata la finestra di dialogo Interrompi job con le opzioni per interrompere il job.
Seleziona Annulla.
Fai clic su Interrompi job. Il servizio interrompe l'importazione e l'elaborazione dei dati il prima possibile. Poiché Annulla interrompe immediatamente l'elaborazione, potresti perdere i dati "in transito". L'arresto di un job potrebbe richiedere alcuni minuti.
Elimina il set di dati BigQuery. Vai all'UI web di BigQuery nella console Google Cloud.
Nel riquadro Explorer (Esplorazione), nella sezione Risorse, fai clic sul set di dati dataflow_sql_tutorial che hai creato.
Nel riquadro dei dettagli, fai clic su Elimina. Viene visualizzata una finestra di dialogo di conferma.
Nella finestra di dialogo Elimina set di dati, digita per confermare il comando di eliminazione
delete
, quindi fai clic su Elimina.
Elimina l'argomento Pub/Sub. Vai alla pagina Argomenti Pub/Sub nella console Google Cloud.
Vai alla pagina degli argomenti Pub/Sub
Seleziona l'argomento
transactions
.Fai clic su Elimina per eliminare definitivamente l'argomento. Viene visualizzata una finestra di dialogo di conferma.
Nella finestra di dialogo Elimina argomento, digita per confermare il comando di eliminazione
delete
, quindi fai clic su Elimina.Vai alla pagina delle sottoscrizioni Pub/Sub.
Seleziona eventuali abbonamenti rimanenti per
transactions
. Se i job non sono più in esecuzione, è possibile che non siano presenti abbonati.Fai clic su Elimina per eliminare definitivamente le iscrizioni. Nella finestra di dialogo di conferma, fai clic su Elimina.
Elimina il bucket di staging di Dataflow in Cloud Storage. Vai alla pagina Bucket di Cloud Storage nella console Google Cloud.
Seleziona il bucket gestione temporanea Dataflow.
Fai clic su Elimina per eliminare definitivamente il bucket. Si apre una finestra di dialogo di conferma.
Nella finestra di dialogo Elimina bucket, conferma il comando di eliminazione digitando
DELETE
, quindi fai clic su Elimina.
Passaggi successivi
- Consulta un introduzione a Dataflow SQL.
- Scopri di più sulle nozioni di base sulla pipeline di streaming.
- Esplora il riferimento SQL di Dataflow.
- Guarda la demo di analisi dei flussi di dati presentata a Cloud Next 2019.