Crea una pipeline Dataflow utilizzando Go

Crea una pipeline Dataflow utilizzando Go

Questa pagina mostra come utilizzare l'Apache Beam SDK 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

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.
Install the Google Cloud CLI.
Se utilizzi un provider di identità (IdP) esterno, devi prima accedere alla gcloud CLI con la tua identità federata.
Per inizializzare gcloud CLI, esegui questo comando:
gcloud init
Create or select a Google Cloud project.
Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the 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.
Verify that billing is enabled for your Google Cloud project.
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
If an authentication error is returned, and you are using an external identity provider (IdP), confirm that you have signed in to the gcloud CLI with your federated identity.
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:USER_IDENTIFIER" --role=ROLE
Replace the following:
• PROJECT_ID: your project ID.
• USER_IDENTIFIER: the identifier for your user account—for example, [e-mail].
• ROLE: the IAM role that you grant to your user account.

Install the Google Cloud CLI.

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

Per inizializzare gcloud CLI, esegui questo comando:

gcloud init

Create or select a Google Cloud project.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the 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.

Verify that billing is enabled for your Google Cloud project.

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

If an authentication error is returned, and you are using an external identity provider (IdP), confirm that you have signed in to the gcloud CLI with your federated identity.

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:USER_IDENTIFIER" --role=ROLE

Replace the following:

• PROJECT_ID: your project ID.
• USER_IDENTIFIER: the identifier for your user account—for example, [e-mail].
• ROLE: the IAM role that you grant to your user account.

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 tuo progetto. Per trovare il numero del progetto, consulta Identificare i progetti o utilizza il comando gcloud projects describe.
   • Sostituisci SERVICE_ACCOUNT_ROLE con ogni singolo ruolo.
2. Create a Cloud Storage bucket and configure it as follows:
   • Set the storage class to 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

   • Copia l'ID progetto Google Cloud 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 di Beam wordcount sulla tua macchina locale, segui questi 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 il tuo modulo Go, 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. Apri il file wordcount.go in un editor a tua scelta.

   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' Google Cloud ID progetto.

   • 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 dei metadati, nel client locale o nelle variabili di ambiente.

   Nota: per specificare un account di servizio worker gestito dall'utente, includi l'--service_account_email opzione della 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 nel 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. Elimina il bucket:

      gcloud storage buckets delete BUCKET_NAME

      Importante:il bucket deve essere vuoto prima di poterlo eliminare.

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

   3. Optional: Revoke the authentication credentials that you created, and delete the local credential file.

      gcloud auth application-default revoke

   4. Optional: Revoke credentials from the 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