Creare una pipeline Dataflow utilizzando Go
Questa pagina mostra come utilizzare l'SDK Apache Beam per Go per creare un programma che definisce una pipeline. Poi esegui la pipeline in locale e dal servizio Dataflow. Per un'introduzione alla pipeline WordCount, consulta Video su come utilizzare WordCount in Apache Beam.
Prima di iniziare
- Accedi al tuo account Google Cloud. Se non conosci Google Cloud, crea un account per valutare le prestazioni dei nostri prodotti in scenari reali. I nuovi clienti ricevono anche 300 $di crediti gratuiti per l'esecuzione, il test e il deployment dei carichi di lavoro.
- Install the Google Cloud CLI.
-
To initialize the gcloud CLI, run the following command:
gcloud init
-
Create or select a Google Cloud project.
-
Create a Google Cloud project:
gcloud projects create PROJECT_ID
Replace
PROJECT_ID
with a name for the Google Cloud project you are creating. -
Select the Google Cloud project that you created:
gcloud config set project PROJECT_ID
Replace
PROJECT_ID
with your Google Cloud project name.
-
-
Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.
-
Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager APIs:
gcloud services enable dataflow
compute_component logging storage_component storage_api cloudresourcemanager.googleapis.com -
Create local authentication credentials for your user account:
gcloud auth application-default login
-
Grant roles to your user account. Run the following command once for each of the following IAM roles:
roles/iam.serviceAccountUser
gcloud projects add-iam-policy-binding PROJECT_ID --member="USER_IDENTIFIER" --role=ROLE
- Replace
PROJECT_ID
with your project ID. -
Replace
USER_IDENTIFIER
with the identifier for your user account. For example,user:myemail@example.com
. - Replace
ROLE
with each individual role.
- Replace
- Install the Google Cloud CLI.
-
To initialize the gcloud CLI, run the following command:
gcloud init
-
Create or select a Google Cloud project.
-
Create a Google Cloud project:
gcloud projects create PROJECT_ID
Replace
PROJECT_ID
with a name for the Google Cloud project you are creating. -
Select the Google Cloud project that you created:
gcloud config set project PROJECT_ID
Replace
PROJECT_ID
with your Google Cloud project name.
-
-
Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.
-
Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager APIs:
gcloud services enable dataflow
compute_component logging storage_component storage_api cloudresourcemanager.googleapis.com -
Create local authentication credentials for your user account:
gcloud auth application-default login
-
Grant roles to your user account. Run the following command once for each of the following IAM roles:
roles/iam.serviceAccountUser
gcloud projects add-iam-policy-binding PROJECT_ID --member="USER_IDENTIFIER" --role=ROLE
- Replace
PROJECT_ID
with your project ID. -
Replace
USER_IDENTIFIER
with the identifier for your user account. For example,user:myemail@example.com
. - Replace
ROLE
with each individual role.
- Replace
Concedi ruoli al tuo account di servizio predefinito Compute Engine. Esegui il comando seguente una volta per ciascuno dei seguenti ruoli IAM:
roles/dataflow.admin
roles/dataflow.worker
roles/storage.objectAdmin
gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" --role=SERVICE_ACCOUNT_ROLE
- Sostituisci
PROJECT_ID
con l'ID progetto. - Sostituisci
PROJECT_NUMBER
con il numero del tuo progetto. Per trovare il numero del progetto, consulta Identificare i progetti o utilizza il comandogcloud projects describe
. - Sostituisci
SERVICE_ACCOUNT_ROLE
con ogni singolo ruolo.
-
Create a Cloud Storage bucket and configure it as follows:
-
Set the storage class to
S
(standard). -
Imposta la posizione di archiviazione come segue:
US
(Stati Uniti). -
Sostituisci
BUCKET_NAME
con un nome bucket univoco. Non includere informazioni sensibili nella perché lo spazio dei nomi dei bucket è globale e visibile pubblicamente. - Copia l'ID progetto Google Cloud e il nome del bucket Cloud Storage. Questi valori ti serviranno più avanti in questa guida rapida.
gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US
-
Set the storage class to
Configurazione dell'ambiente di sviluppo
L'SDK Apache Beam è un modello di programmazione open source per pipeline di dati. Definisci una pipeline con un programma Apache Beam e quindi scegli un runner, come Dataflow, per eseguire la pipeline.
Ti consigliamo di utilizzare l'ultima versione di Go quando lavori con l'SDK Apache Beam per Go. In caso contrario avere l'ultima versione di Go, usa la sua Scarica e installa la guida per scaricare e installare Go per il sistema operativo specifico.
Per verificare la versione di Go installata, esegui questo comando nel tuo terminale locale:
go version
Esegui l'esempio di conteggio parole di Beam
L'SDK Apache Beam per Go include un file
Esempio di pipeline wordcount
.
L'esempio wordcount
fa quanto segue:
- Legge un file di testo come input. Per impostazione predefinita, legge un file di testo che si trova in una
Bucket Cloud Storage con il nome della risorsa
gs://dataflow-samples/shakespeare/kinglear.txt
. - Analizza ogni riga in parole.
- Esegue un conteggio della frequenza sulle parole tokenizzate.
Per eseguire l'ultima versione dell'esempio Beam wordcount
sulla tua macchina locale,
usa questo comando. Il flag input
specifica il file da leggere.
e il flag output
specifica il nome del file per l'output del conteggio della frequenza.
go run github.com/apache/beam/sdks/v2/go/examples/wordcount@latest \
--input gs://dataflow-samples/shakespeare/kinglear.txt \
--output outputs
Al termine della pipeline, visualizza i risultati di output:
more outputs*
Per uscire, premi q.
Modifica il codice della pipeline
La pipeline Beam wordcount
fa distinzione tra lettere maiuscole e minuscole
parole. I passaggi seguenti mostrano come creare il proprio modulo Go, modificare
Pipeline wordcount
in modo che non faccia distinzione tra maiuscole e minuscole ed eseguila su
e Dataflow.
Crea un modulo Go
Per apportare modifiche al codice della pipeline, segui questi passaggi.
Crea una directory per il modulo Go in una posizione a tua scelta:
mkdir wordcount
cd wordcount
Crea un modulo Go. Per questo esempio, usa
example/dataflow
come modulo del tuo percorso di apprendimento.go mod init example/dataflow
Scarica l'ultima copia dell' Codice
wordcount
dal repository GitHub di Apache Beam. Inserisci questo file nella Directorywordcount
creata da te.Se utilizzi un sistema operativo non Linux, devi scaricare il Pacchetto Go
unix
. Questo pacchetto è necessario per eseguire pipeline in Dataflow completamente gestito di Google Cloud.go get -u golang.org/x/sys/unix
Assicurati che il file
go.mod
corrisponda al codice sorgente del modulo:go mod tidy
Esegui la pipeline non modificata
Verifica che la pipeline wordcount
non modificata venga eseguita localmente.
Dal terminale, crea ed esegui la pipeline in locale:
go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \ --output outputs
Visualizza i risultati dell'output:
more outputs*
Per uscire, premi q.
Modifica il codice della pipeline
Per cambiare la pipeline in modo che non sia sensibile alle maiuscole, modifica il codice in
applica il strings.ToLower
a tutte le parole.
Apri il file
wordcount.go
in un editor a tua scelta.Esamina il blocco
init
(i commenti sono stati rimossi per maggiore chiarezza):func init() { register.DoFn3x0[context.Context, string, func(string)](&extractFn{}) register.Function2x1(formatFn) register.Emitter1[string]() }
Aggiungi una nuova riga per registrare la funzione
strings.ToLower
:func init() { register.DoFn3x0[context.Context, string, func(string)](&extractFn{}) register.Function2x1(formatFn) register.Emitter1[string]() register.Function1x1(strings.ToLower) }
Esamina la funzione
CountWords
:func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection { s = s.Scope("CountWords") // Convert lines of text into individual words. col := beam.ParDo(s, &extractFn{SmallWordLength: *smallWordLength}, lines) // Count the number of times each word occurs. return stats.Count(s, col) }
Per scrivere le parole in minuscolo, aggiungi un ParDo che applica
strings.ToLower
a ogni parola:func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection { s = s.Scope("CountWords") // Convert lines of text into individual words. col := beam.ParDo(s, &extractFn{SmallWordLength: *smallWordLength}, lines) // Map all letters to lowercase. lowercaseWords := beam.ParDo(s, strings.ToLower, col) // Count the number of times each word occurs. return stats.Count(s, lowercaseWords) }
Salva il file.
Esegui la pipeline aggiornata in locale
Esegui la pipeline wordcount
aggiornata in locale e verifica che l'output sia cambiato.
Crea ed esegui la pipeline
wordcount
modificata:go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \ --output outputs
Visualizza i risultati di output della pipeline modificata. Tutte le parole devono essere minuscolo.
more outputs*
Per uscire, premi q.
Esegui la pipeline sul servizio Dataflow
Per eseguire l'esempio aggiornato di wordcount
sul servizio Dataflow,
utilizza questo comando:
go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
--output gs://BUCKET_NAME/results/outputs \
--runner dataflow \
--project PROJECT_ID \
--region DATAFLOW_REGION \
--staging_location gs://BUCKET_NAME/binaries/
Sostituisci quanto segue:
BUCKET_NAME
: il nome del bucket Cloud Storage.PROJECT_ID
: l'ID del progetto Google Cloud.DATAFLOW_REGION
: il region in cui vuoi eseguire il deployment del job Dataflow. Ad esempio:europe-west1
. Per un elenco delle località disponibili, vedi Località di Dataflow. Il flag--region
sostituisce la regione predefinita impostata nei metadati del server, del client locale o delle variabili di ambiente.
Visualizza i tuoi risultati
Puoi visualizzare un elenco dei tuoi job Dataflow in nella console Google Cloud. Nella console Google Cloud, vai alla Pagina Job di Dataflow.
La pagina Job mostra i dettagli del job wordcount
, incluso uno stato
su In esecuzione e poi su Riuscito.
Quando esegui una pipeline utilizzando Dataflow, i risultati vengono archiviati in in un bucket Cloud Storage. Visualizza i risultati di output utilizzando nella console Google Cloud o nel terminale locale.
Console
Per visualizzare i risultati nella console Google Cloud, vai a Bucket di Cloud Storage.
Nell'elenco dei bucket del progetto, fai clic sul bucket
creati in precedenza. I file di output creati dal job vengono visualizzati nell'intestazione
Directory results
.
Terminale
Visualizza i risultati dal terminale o utilizzando Cloud Shell.
Per elencare i file di output, utilizza il comando
gcloud storage ls
:gcloud storage ls gs://BUCKET_NAME/results/outputs* --long
Sostituisci
BUCKET_NAME
con il nome del specificato per il bucket Cloud Storage di output.Per visualizzare i risultati nei file di output, utilizza il comando
gcloud storage cat
:gcloud storage cat gs://BUCKET_NAME/results/outputs*
Esegui la pulizia
Per evitare che al tuo account Google Cloud vengano addebitati costi per le risorse utilizzate in questa pagina, elimina il progetto Google Cloud Google Cloud.
- In the Google Cloud console, go to the Cloud Storage Buckets page.
- Click the checkbox for the bucket that you want to delete.
- To delete the bucket, click Delete, and then follow the instructions.
Se mantieni il progetto, revoca i ruoli che hai concesso all'account di servizio predefinito di Compute Engine. Esegui il comando seguente una volta per ciascuno dei seguenti ruoli IAM:
roles/dataflow.admin
roles/dataflow.worker
roles/storage.objectAdmin
gcloud projects remove-iam-policy-binding PROJECT_ID \ --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \ --role=SERVICE_ACCOUNT_ROLE
-
Optional: Revoke the authentication credentials that you created, and delete the local credential file.
gcloud auth application-default revoke
-
Optional: Revoke credentials from the gcloud CLI.
gcloud auth revoke