Cloud Composer 1 | Cloud Composer 2
Questa pagina descrive come utilizzare DataflowTemplateOperator
per avviare pipeline Dataflow da Cloud Composer.
La pipeline Da testo a BigQuery di Cloud Storage è una pipeline batch che ti consente di caricare file di testo archiviati in Cloud Storage, trasformarli utilizzando una funzione JavaScript definita dall'utente (UDF) da te fornita e inviare i risultati a BigQuery.
Panoramica
Prima di avviare il flusso di lavoro, creerai le seguenti entità:
Una tabella BigQuery vuota da un set di dati vuoto che conterrà le seguenti colonne di informazioni:
location
,average_temperature
,month
e, facoltativamente,inches_of_rain
,is_current
elatest_measurement
.Un file JSON che normalizzerà i dati del file
.txt
nel formato corretto per lo schema della tabella BigQuery. L'oggetto JSON avrà un array diBigQuery Schema
, in cui ogni oggetto conterrà un nome colonna, un tipo di input e se si tratta o meno di un campo obbligatorio.Un file
.txt
di input che conterrà i dati che verranno caricati collettivamente nella tabella BigQuery.Una funzione definita dall'utente scritta in JavaScript che trasformerà ogni riga del file
.txt
nelle variabili pertinenti per la nostra tabella.Un file DAG di Airflow che rimanda alla posizione di questi file.
Successivamente, caricherai il file
.txt
, il file UDF.js
e il file dello schema.json
in un bucket Cloud Storage. Inoltre, caricherai il DAG nel tuo ambiente Cloud Composer.Dopo aver caricato il DAG, Airflow esegue un'attività al suo interno. Questa attività avvierà una pipeline Dataflow che applicherà la funzione definita dall'utente al file
.txt
e la formatterà in base allo schema JSON.Infine, i dati verranno caricati nella tabella BigQuery creata in precedenza.
Prima di iniziare
- Questa guida richiede familiarità con JavaScript per scrivere la funzione definita dall'utente.
- Questa guida presuppone che tu abbia già un ambiente Cloud Composer. Vedi Creare ambiente per crearne uno. Con questa guida puoi utilizzare qualsiasi versione di Cloud Composer.
-
Enable the Cloud Composer, Dataflow, Cloud Storage, BigQuery APIs.
Crea una tabella BigQuery vuota con una definizione di schema
Crea una tabella BigQuery con una definizione di schema. Utilizzerai questa definizione di schema più avanti in questa guida. Questa tabella BigQuery conterrà i risultati del caricamento batch.
Per creare una tabella vuota con una definizione di schema:
Console
Nella console Google Cloud, vai alla pagina di BigQuery:
Nella sezione Risorse del pannello di navigazione, espandi il progetto.
Nel riquadro dei dettagli, fai clic su Crea set di dati.
Nella pagina Crea set di dati, nella sezione ID set di dati, assegna il nome
average_weather
al set di dati. Lasciare tutti gli altri campi nello stato predefinito.Fai clic su Crea set di dati.
Torna al pannello di navigazione e nella sezione Risorse espandi il progetto. Quindi, fai clic sul set di dati
average_weather
.Nel riquadro dei dettagli, fai clic su Crea tabella.
Nella pagina Crea tabella, nella sezione Origine, seleziona Tabella vuota.
Nella sezione Destinazione della pagina Crea tabella:
In Nome set di dati, scegli il set di dati
average_weather
.Nel campo Nome tabella, inserisci il nome
average_weather
.Verifica che l'opzione Tipo di tabella sia impostata su Tabella nativa.
Nella sezione Schema, inserisci la definizione di schema. Puoi utilizzare uno dei seguenti approcci:
Inserisci manualmente le informazioni dello schema abilitando Modifica come testo e inserendo lo schema della tabella come array JSON. Digita i seguenti campi:
[ { "name": "location", "type": "GEOGRAPHY", "mode": "REQUIRED" }, { "name": "average_temperature", "type": "INTEGER", "mode": "REQUIRED" }, { "name": "month", "type": "STRING", "mode": "REQUIRED" }, { "name": "inches_of_rain", "type": "NUMERIC" }, { "name": "is_current", "type": "BOOLEAN" }, { "name": "latest_measurement", "type": "DATE" } ]
Utilizza Aggiungi campo per inserire manualmente lo schema:
In Impostazioni di partizione e cluster lascia il valore predefinito,
No partitioning
.Nella sezione Opzioni avanzate, per Crittografia lascia il valore predefinito,
Google-managed key
.Fai clic su Crea tabella.
bq
Utilizza il comando bq mk
per creare un set di dati e una tabella vuoti in questo set di dati.
Esegui questo comando per creare un set di dati del meteo globale medio:
bq --location=LOCATION mk \
--dataset PROJECT_ID:average_weather
Sostituisci quanto segue:
LOCATION
: la regione in cui si trova l'ambiente.PROJECT_ID
: l'ID progetto.
Esegui questo comando per creare una tabella vuota in questo set di dati con la definizione dello schema:
bq mk --table \
PROJECT_ID:average_weather.average_weather \
location:GEOGRAPHY,average_temperature:INTEGER,month:STRING,inches_of_rain:NUMERIC,is_current:BOOLEAN,latest_measurement:DATE
Dopo aver creato la tabella, puoi aggiornare la scadenza, la descrizione e le etichette della tabella. Puoi anche modificare la definizione dello schema.
Python
Salva questo codice come dataflowtemplateoperator_create_dataset_and_table_helper.py
e aggiorna le variabili al suo interno in modo che riflettano il tuo progetto e la tua località, poi eseguilo con il seguente comando:
python dataflowtemplateoperator_create_dataset_and_table_helper.py
Python
Per eseguire l'autenticazione su Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.
crea un bucket Cloud Storage
Crea un bucket per contenere tutti i file necessari per il flusso di lavoro. Il DAG creato più avanti in questa guida farà riferimento ai file caricati in questo bucket di archiviazione. Per creare un nuovo bucket di archiviazione:
Console
Apri Cloud Storage nella console Google Cloud.
Fai clic su Crea bucket per aprire il modulo di creazione del bucket.
Inserisci le informazioni sul bucket e fai clic su Continua per completare ogni passaggio:
Specifica un nome globalmente univoco per il bucket. Questa guida utilizza
bucketName
come esempio.Seleziona Regione come tipo di località. Seleziona una località in cui verranno archiviati i dati del bucket.
Seleziona Standard come classe di archiviazione predefinita per i tuoi dati.
Seleziona controllo dell'accesso Uniforme per accedere ai tuoi oggetti.
Fai clic su Fine.
gsutil
Usa il comando gsutil mb
:
gsutil mb gs://bucketName/
Sostituisci quanto segue:
bucketName
: il nome del bucket che hai creato in precedenza in questa guida.
Esempi di codice
C#
Per eseguire l'autenticazione su Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.
Go
Per eseguire l'autenticazione su Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.
Java
Per eseguire l'autenticazione su Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.
Python
Per eseguire l'autenticazione su Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.
Ruby
Per eseguire l'autenticazione su Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.
Crea uno schema BigQuery in formato JSON per la tabella di output
Crea un file di schema BigQuery in formato JSON che corrisponda alla tabella di output creata in precedenza. Tieni presente che nomi, tipi e modalità dei campi devono corrispondere a quelli definiti in precedenza nello schema della tabella BigQuery. Questo file normalizzerà i dati del file .txt
in un formato compatibile con lo schema BigQuery. Assegna a questo file il nome jsonSchema.json
.
{
"BigQuery Schema": [
{
"name": "location",
"type": "GEOGRAPHY",
"mode": "REQUIRED"
},
{
"name": "average_temperature",
"type": "INTEGER",
"mode": "REQUIRED"
},
{
"name": "month",
"type": "STRING",
"mode": "REQUIRED"
},
{
"name": "inches_of_rain",
"type": "NUMERIC"
},
{
"name": "is_current",
"type": "BOOLEAN"
},
{
"name": "latest_measurement",
"type": "DATE"
}]
}
Crea un file JavaScript per formattare i dati
In questo file, definirai la funzione definita dall'utente che fornisce la logica per trasformare le righe di testo nel file di input. Tieni presente che questa funzione prende ogni riga di testo nel file di input come argomento personale, quindi verrà eseguita una volta per ogni riga del file di input. Assegna a questo file il nome transformCSVtoJSON.js
.
Crea il file di input
Questo file conterrà le informazioni che vuoi caricare nella tabella BigQuery. Copia questo file in locale e assegnagli il nome inputFile.txt
.
POINT(40.7128 74.006),45,'July',null,true,2020-02-16
POINT(41.8781 87.6298),23,'October',13,false,2015-02-13
POINT(48.8566 2.3522),80,'December',null,true,null
POINT(6.5244 3.3792),15,'March',14,true,null
Carica i file nel bucket
Carica i seguenti file nel bucket Cloud Storage che hai creato in precedenza:
- Schema BigQuery in formato JSON (
.json
) - Funzione JavaScript definita dall'utente (
transformCSVtoJSON.js
) Il file di input del testo da elaborare (
.txt
)
Console
- Nella console Google Cloud, vai alla pagina Bucket di Cloud Storage.
Nell'elenco dei bucket, fai clic sul tuo bucket.
Nella scheda Oggetti del bucket, esegui una delle seguenti operazioni:
Trascina i file desiderati dal desktop o da gestore di file nel riquadro principale della console Google Cloud.
Fai clic sul pulsante Carica file, seleziona i file da caricare nella finestra di dialogo visualizzata e fai clic su Apri.
gsutil
Esegui il comando gsutil cp
:
gsutil cp OBJECT_LOCATION gs://bucketName
Sostituisci quanto segue:
bucketName
: il nome del bucket che hai creato in precedenza in questa guida.OBJECT_LOCATION
: il percorso locale dell'oggetto. Ad esempio,Desktop/transformCSVtoJSON.js
.
Esempi di codice
Python
Per eseguire l'autenticazione su Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.
Ruby
Per eseguire l'autenticazione su Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per maggiori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale.
Configura DataflowTemplateOperator
Prima di eseguire il DAG, imposta le seguenti variabili Airflow.
Variabile Airflow | Valore |
---|---|
project_id
|
L'ID progetto |
gce_zone
|
Zona di Compute Engine in cui deve essere creato il cluster Dataflow |
bucket_path
|
La località del bucket Cloud Storage creato in precedenza |
Ora farai riferimento ai file creati in precedenza per creare un DAG che avvii il flusso di lavoro Dataflow. Copia questo DAG e salvalo localmente come composer-dataflow-dag.py
.
Airflow 2
Airflow 1
carica il DAG in Cloud Storage
Carica il DAG nella cartella /dags
nel bucket del tuo ambiente. Una volta completato il caricamento, puoi visualizzarlo facendo clic sul link Cartella DAG nella pagina Ambienti di Cloud Composer.
Visualizzare lo stato dell'attività
- Vai all'interfaccia web di Airflow.
- Nella pagina dei DAG, fai clic sul nome del DAG (ad esempio
composerDataflowDAG
). - Nella pagina dei dettagli dei DAG, fai clic su Visualizzazione grafico.
Controlla lo stato:
Failed
: intorno all'attività è presente un riquadro rosso. Puoi anche tenere il puntatore sull'attività e cercare lo stato Stato: non riuscito.Success
: intorno all'attività è presente un riquadro verde. Puoi anche tenere il puntatore del mouse sull'attività e controllare se è presente Stato: riuscito.
Dopo alcuni minuti, puoi verificare i risultati in Dataflow e BigQuery.
Visualizza il job in Dataflow
Nella console Google Cloud, vai alla pagina Dataflow.
Il job è denominato
dataflow_operator_transform_csv_to_bq
con un ID univoco accodato alla fine del nome con un trattino, in questo modo:Fai clic sul nome per vedere i dettagli del job.
Visualizza i risultati in BigQuery
Nella console Google Cloud, vai alla pagina BigQuery.
Puoi inviare query utilizzando SQL standard. Utilizza la seguente query per visualizzare le righe aggiunte alla tabella:
SELECT * FROM projectId.average_weather.average_weather