Crea una pipeline Dataflow utilizzando Go

Crea 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 sul servizio Dataflow. Per un'introduzione alla pipeline WordCount, guarda il video 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 senza costi per l'esecuzione, il test e il deployment dei workload.
Installa Google Cloud CLI.
Se utilizzi un provider di identità (IdP) esterno, devi prima accedere a gcloud CLI con la tua identità federata.
Per inizializzare gcloud CLI, esegui questo comando:
gcloud init
Crea o seleziona un Google Cloud progetto.
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 il quale 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.
Nota: se non prevedi di conservare le risorse che crei in questa procedura, crea un progetto invece di selezionarne uno già esistente. Una volta completata questa procedura, puoi eliminare il progetto e tutte le relative risorse.
• Creare un progetto Google Cloud :

  gcloud projects create PROJECT_ID

  Sostituisci PROJECT_ID con un nome per il progetto Google Cloud che stai creando.

• Seleziona il progetto Google Cloud che hai creato:

  gcloud config set project PROJECT_ID

  Sostituisci PROJECT_ID con il nome del progetto Google Cloud .
Verifica che la fatturazione sia abilitata per il tuo progetto Google Cloud .
Abilita le API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON e Cloud Resource Manager:
Ruoli richiesti per abilitare le API
Per abilitare le API, devi disporre del ruolo IAM Amministratore utilizzo dei servizi (roles/serviceusage.serviceUsageAdmin), che include l'autorizzazione serviceusage.services.enable. Scopri come concedere i ruoli.
gcloud services enable dataflow compute_component logging storage_component storage_api cloudresourcemanager.googleapis.com
Crea credenziali di autenticazione locali per il tuo account utente:
gcloud auth application-default login
Se viene restituito un errore di autenticazione e utilizzi un provider di identità (IdP) esterno, verifica di aver acceduto a gcloud CLI con la tua identità federata.
Concedi ruoli al tuo account utente. Esegui il seguente comando una volta per ciascuno dei seguenti ruoli IAM: roles/iam.serviceAccountUser
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE
Sostituisci quanto segue:
• PROJECT_ID: il tuo ID progetto.
• USER_IDENTIFIER: l'identificatore del tuo account utente . Ad esempio: [e-mail].
• ROLE: il ruolo IAM che concedi al tuo account utente.

Installa Google Cloud CLI.

Se utilizzi un provider di identità (IdP) esterno, devi prima accedere a gcloud CLI con la tua identità federata.

Per inizializzare gcloud CLI, esegui questo comando:

gcloud init

Crea o seleziona un Google Cloud progetto.

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

Nota: se non prevedi di conservare le risorse che crei in questa procedura, crea un progetto invece di selezionarne uno già esistente. Una volta completata questa procedura, puoi eliminare il progetto e tutte le relative risorse.

• Creare un progetto Google Cloud :

  gcloud projects create PROJECT_ID

  Sostituisci PROJECT_ID con un nome per il progetto Google Cloud che stai creando.

• Seleziona il progetto Google Cloud che hai creato:

  gcloud config set project PROJECT_ID

  Sostituisci PROJECT_ID con il nome del progetto Google Cloud .

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

Abilita le API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON e Cloud Resource Manager:

Ruoli richiesti per abilitare le API

Per abilitare le API, devi disporre del ruolo IAM Amministratore utilizzo dei servizi (roles/serviceusage.serviceUsageAdmin), che include l'autorizzazione serviceusage.services.enable. Scopri come concedere i ruoli.

gcloud services enable dataflow compute_component logging storage_component storage_api cloudresourcemanager.googleapis.com

Crea credenziali di autenticazione locali per il tuo account utente:

gcloud auth application-default login

Se viene restituito un errore di autenticazione e utilizzi un provider di identità (IdP) esterno, verifica di aver acceduto a gcloud CLI con la tua identità federata.

Concedi ruoli al tuo account utente. Esegui il seguente comando una volta per ciascuno dei seguenti ruoli IAM: roles/iam.serviceAccountUser

gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

Sostituisci quanto segue:

• PROJECT_ID: il tuo ID progetto.
• USER_IDENTIFIER: l'identificatore del tuo account utente . Ad esempio: [e-mail].
• ROLE: il ruolo IAM che concedi al tuo account utente.

1. Concedi ruoli al account di servizio Compute Engine predefinito. Esegui il seguente comando 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:[e-mail]" --role=SERVICE_ACCOUNT_ROLE

   • Sostituisci PROJECT_ID con l'ID progetto.
   • Sostituisci PROJECT_NUMBER con il numero del progetto. Per trovare il numero del progetto, consulta Identifica i progetti o utilizza il comando gcloud projects describe.
   • Sostituisci SERVICE_ACCOUNT_ROLE con ogni singolo ruolo.
2. Crea un bucket Cloud Storage e configuralo nel seguente modo:
   • Imposta la classe di archiviazione su S (Standard).
   • Imposta la posizione di archiviazione su: US (Stati Uniti).
   • Sostituisci BUCKET_NAME con un nome di bucket univoco. Non includere informazioni sensibili nel nome del bucket perché lo spazio dei nomi dei bucket è globale e visibile pubblicamente.

   gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US

3. Copia l' Google Cloud ID progetto e il nome del bucket Cloud Storage. Questi valori ti serviranno più avanti in questa guida rapida.

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 poi scegli un runner, ad esempio Dataflow, per eseguire la pipeline.

Ti consigliamo di utilizzare l'ultima versione di Go quando lavori con l'SDK Apache Beam per Go. Se non hai installato l'ultima versione di Go, utilizza la guida al download e all'installazione di Go per scaricare e installare Go per il tuo sistema operativo specifico.

Per verificare la versione di Go installata, esegui questo comando nel terminale locale:

go version

Esegui l'esempio di conteggio parole di Beam

L'SDK Apache Beam per Go include un wordcount esempio di pipeline. L'esempio wordcount esegue le seguenti operazioni:

1. Legge un file di testo come input. Per impostazione predefinita, legge un file di testo che si trova in un bucket Cloud Storage con il nome della risorsa gs://dataflow-samples/shakespeare/kinglear.txt.
2. Analizza ogni riga in parole.
3. Esegue un conteggio della frequenza delle parole tokenizzate.

Per eseguire l'ultima versione dell'esempio wordcount di Beam sulla tua macchina locale, svolgi i seguenti passaggi:

1. Utilizza il comando git clone per clonare il repository GitHub apache/beam:

   git clone https://github.com/apache/beam.git

2. Passa alla directory beam/sdks/go:

   cd beam/sdks/go

3. Utilizza il seguente comando per eseguire la pipeline:

   go run examples/wordcount/wordcount.go \
     --input gs://dataflow-samples/shakespeare/kinglear.txt \
     --output outputs

   Il flag input specifica il file da leggere, mentre il flag output specifica il nome del file per l'output del conteggio della frequenza.

Al termine della pipeline, visualizza i risultati dell'output:

more outputs*

Per uscire, premi q.

Modifica il codice della pipeline

La pipeline Beam wordcount distingue tra parole maiuscole e minuscole. I seguenti passaggi mostrano come creare un modulo Go personalizzato, modificare la pipeline wordcount in modo che non faccia distinzione tra maiuscole e minuscole ed eseguirla su Dataflow.

Crea un modulo Go

Per apportare modifiche al codice della pipeline:

1. Crea una directory per il modulo Go nella posizione che preferisci:

   mkdir wordcount
   cd wordcount

2. Crea un modulo Go. Per questo esempio, utilizza example/dataflow come percorso del modulo.

   go mod init example/dataflow

3. Scarica l'ultima copia del codice wordcount dal repository GitHub di Apache Beam. Inserisci questo file nella directory wordcount che hai creato.

4. Se utilizzi un sistema operativo non Linux, devi scaricare il pacchetto Go unix. Questo pacchetto è necessario per eseguire le pipeline sul servizio Dataflow.

   go get -u golang.org/x/sys/unix

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

1. Dal terminale, crea ed esegui la pipeline in locale:

    go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
        --output outputs

2. Visualizza i risultati dell'output:

    more outputs*

3. Per uscire, premi q.

Modificare il codice della pipeline

Per modificare la pipeline in modo che non faccia distinzione tra maiuscole e minuscole, modifica il codice in modo da applicare la funzione strings.ToLower a tutte le parole.

1. In un editor a tua scelta, apri il file wordcount.go.

2. 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]()
    }
   

3. 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)
    }
   

4. 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)
    }
   

5. Per convertire le parole in minuscolo, aggiungi un ParDo che applichi 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)
    }
   

6. Salva il file.

Esegui la pipeline aggiornata in locale

Esegui la pipeline wordcount aggiornata localmente e verifica che l'output sia cambiato.

1. Crea ed esegui la pipeline wordcount modificata:

    go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
        --output outputs

2. Visualizza i risultati dell'output della pipeline modificata. Tutte le parole devono essere in minuscolo.

    more outputs*

3. Per uscire, premi q.

Esegui la pipeline sul servizio Dataflow

Per eseguire l'esempio wordcount aggiornato sul servizio Dataflow, utilizza il seguente 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 progetto Google Cloud .

• DATAFLOW_REGION: la regione in cui vuoi eseguire il deployment del job Dataflow. Ad esempio: europe-west1. Per un elenco delle località disponibili, consulta Località di Dataflow. Il flag --region sostituisce la regione predefinita impostata nel server di metadati, nel client locale o nelle variabili di ambiente.

Nota: per specificare un service account worker gestito dall'utente, includi l'--service_account_email opzione pipeline. I service account worker gestiti dall'utente sono consigliati per i carichi di lavoro di produzione. Se non specifichi un service account worker quando crei un job, Dataflow utilizza l'account di servizio predefinito di Compute Engine.

Visualizza i tuoi risultati

Puoi visualizzare un elenco dei tuoi job Dataflow nella consoleGoogle Cloud . Nella console Google Cloud , vai alla pagina Job Dataflow.

Vai a Job

La pagina Job mostra i dettagli del job wordcount, incluso uno stato inizialmente In esecuzione e poi Riuscito.

Quando esegui una pipeline utilizzando Dataflow, i risultati vengono archiviati in un bucket Cloud Storage. Visualizza i risultati dell'output utilizzando la consoleGoogle Cloud o il terminale locale.

Console

Per visualizzare i risultati nella console Google Cloud , vai alla pagina Bucket in Cloud Storage.

Vai a Bucket

Nell'elenco dei bucket del progetto, fai clic sul bucket di archiviazione che hai creato in precedenza. I file di output creati dal job vengono visualizzati nella directory results.

Terminale

Visualizza i risultati dal terminale o utilizzando Cloud Shell.

1. 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 bucket Cloud Storage di output specificato.

2. 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 relativi alle risorse utilizzate in questa pagina, elimina il progetto Google Cloud con le risorse.

Nota:se hai seguito questa guida rapida in un nuovo progetto, puoi eliminare il progetto.

1. Nella console Google Cloud , vai alla pagina Bucket in Cloud Storage.

   Vai a Bucket

2. Fai clic sulla casella di controllo del bucket da eliminare.
3. Per eliminare il bucket, fai clic su Elimina, quindi segui le istruzioni.

4. Se mantieni il progetto, revoca i ruoli che hai concesso al account di servizio Compute Engine predefinito. Esegui il seguente comando 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:[e-mail] \
       --role=SERVICE_ACCOUNT_ROLE

5. (Facoltativo) Revoca le credenziali di autenticazione che hai creato ed elimina il file delle credenziali locale.

   gcloud auth application-default revoke

6. (Facoltativo) Revoca le credenziali da gcloud CLI.

   gcloud auth revoke

Passaggi successivi

• Modello di programmazione per Apache Beam
• Impostazione delle opzioni pipeline
• Guida di riferimento alle opzioni della pipeline
• Deployment di una pipeline