Crea una pipeline Dataflow utilizzando Python

Crea una pipeline Dataflow utilizzando Python

In questa guida rapida, imparerai a utilizzare l'SDK Apache Beam per Python per creare un programma che definisca una pipeline. Quindi, esegui la pipeline utilizzando un runner locale diretto o un runner basato su cloud come Dataflow. Per un'introduzione alla pipeline WordCount, guarda il video Come utilizzare WordCount in Apache Beam.

---

Per seguire le indicazioni dettagliate per questa attività direttamente nella console Google Cloud, fai clic su Aiuto:

Guidami

---

Prima di iniziare

1. 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.
2. Installa Google Cloud CLI.

3. Per initialize gcloud CLI, esegui questo comando:

   gcloud init

4. Crea o seleziona un progetto Google Cloud.

   Nota: se non prevedi di conservare le risorse create in questa procedura, crea un progetto anziché selezionarne uno esistente. Dopo aver completato questi passaggi, puoi eliminare il progetto, rimuovendo tutte le risorse associate.

   • Crea 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 tuo progetto Google Cloud.

5. Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.

6. Abilita le API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, JSON di Google Cloud Storage, BigQuery, Cloud Pub/Sub, Cloud Datastore e Cloud Resource Manager.

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

7. Crea credenziali di autenticazione locali per il tuo Account Google:

   gcloud auth application-default login

8. Concedi i ruoli al tuo Account Google. Esegui questo comando una volta per ciascuno dei seguenti ruoli IAM: roles/iam.serviceAccountUser

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

   • Sostituisci PROJECT_ID con l'ID progetto.
   • Sostituisci EMAIL_ADDRESS con il tuo indirizzo email.
   • Sostituisci ROLE con ogni singolo ruolo.
9. Installa Google Cloud CLI.

10. Per initialize gcloud CLI, esegui questo comando:

    gcloud init

11. Crea o seleziona un progetto Google Cloud.

    Nota: se non prevedi di conservare le risorse create in questa procedura, crea un progetto anziché selezionarne uno esistente. Dopo aver completato questi passaggi, puoi eliminare il progetto, rimuovendo tutte le risorse associate.

    • Crea 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 tuo progetto Google Cloud.

12. Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.

13. Abilita le API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, JSON di Google Cloud Storage, BigQuery, Cloud Pub/Sub, Cloud Datastore e Cloud Resource Manager.

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

14. Crea credenziali di autenticazione locali per il tuo Account Google:

    gcloud auth application-default login

15. Concedi i ruoli al tuo Account Google. Esegui questo comando una volta per ciascuno dei seguenti ruoli IAM: roles/iam.serviceAccountUser

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

    • Sostituisci PROJECT_ID con l'ID progetto.
    • Sostituisci EMAIL_ADDRESS con il tuo indirizzo email.
    • Sostituisci ROLE con ogni singolo ruolo.

16. Concedi ruoli al tuo account di servizio predefinito di Compute Engine. Esegui il comando seguente una volta per ognuno 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.
17. Crea un bucket Cloud Storage e configuralo nel seguente modo:
    • Imposta la classe di archiviazione su S (standard).
    • Imposta la località di archiviazione come segue: 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

18. Copia l'ID progetto Google Cloud e il nome del bucket Cloud Storage. Questi valori saranno necessari più avanti in questo documento.

configura l'ambiente

Dataflow non supporta più le pipeline che utilizzano Python 2. Per ulteriori informazioni, consulta la pagina Assistenza di Python 2 su Google Cloud. Per maggiori dettagli sulle versioni supportate, consulta Supporto del runtime Apache Beam.

In questa sezione utilizza il prompt dei comandi per configurare un ambiente virtuale Python isolato per eseguire il progetto della pipeline utilizzando venv. Questo processo consente di isolare le dipendenze di un progetto da quelle di altri progetti.

Se non hai un prompt dei comandi immediatamente disponibile, puoi utilizzare Cloud Shell. Cloud Shell ha già installato il gestore di pacchetti per Python 3, quindi puoi passare direttamente alla creazione di un ambiente virtuale.

Per installare Python e quindi creare un ambiente virtuale, segui questi passaggi:

1. Verifica che nel sistema siano in esecuzione Python 3 e pip:

   python --version
   python -m pip --version
   

2. Se necessario, installa Python 3, quindi configura un ambiente virtuale Python: segui le istruzioni fornite nelle sezioni Installazione di Python e Configurazione di venv della pagina Configurazione di un ambiente di sviluppo Python. Se utilizzi Python 3.10 o versioni successive, devi anche abilitare Dataflow Runner v2. Per usare Runner v1, usa Python 3.9 o versioni precedenti.

Nota: Cython non è necessario, ma se è installato, la versione deve essere 0.28.1 o successiva. Per controllare la tua versione di Cython, esegui pip show cython.

Una volta completata la guida rapida, puoi disattivare l'ambiente virtuale eseguendo deactivate.

Scarica l'SDK Apache Beam

L'SDK Apache Beam è un modello di programmazione open source per le pipeline di dati. Devi definire una pipeline con un programma Apache Beam, quindi scegliere un runner, ad esempio Dataflow, per eseguire la pipeline.

Per scaricare e installare l'SDK Apache Beam:

1. Verifica di essere nell'ambiente virtuale Python che hai creato nella sezione precedente. Assicurati che il prompt inizi con <env_name>, dove env_name è il nome dell'ambiente virtuale.
2. Installa lo standard di pacchetto Python wheel:

   pip install wheel
   

3. Installa la versione più recente dell'SDK Apache Beam per Python:

pip install 'apache-beam[gcp]'

Su Microsoft Windows, utilizza il seguente comando:

pip install apache-beam[gcp]

A seconda della connessione, l'installazione potrebbe richiedere del tempo.

esegui la pipeline in locale

Per vedere come viene eseguita una pipeline localmente, utilizza un modulo Python pronto per l'esempio wordcount incluso nel pacchetto apache_beam.

L'esempio di pipeline wordcount esegue quanto segue:

1. Prende un file di testo come input.

   Questo file di testo 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 di frequenza sulle parole tokenizzate.

Per posizionare in un'area intermedia la pipeline wordcount in locale, segui questi passaggi:

1. Dal terminale locale, esegui l'esempio wordcount:

   python -m apache_beam.examples.wordcount \
     --output outputs

2. Visualizza l'output della pipeline:

   more outputs*

3. Per uscire, premi Q.

L'esecuzione della pipeline in locale consente di testare ed eseguire il debug del programma Apache Beam. Puoi visualizzare il codice sorgente wordcount.py su GitHub di Apache Beam.

Esegui la pipeline sul servizio Dataflow

In questa sezione, esegui la pipeline di esempio wordcount dal pacchetto apache_beam sul servizio Dataflow. Questo esempio specifica DataflowRunner come parametro per --runner.

• Esegui la pipeline:

  python -m apache_beam.examples.wordcount \
      --region DATAFLOW_REGION \
      --input gs://dataflow-samples/shakespeare/kinglear.txt \
      --output gs://BUCKET_NAME/results/outputs \
      --runner DataflowRunner \
      --project PROJECT_ID \
      --temp_location gs://BUCKET_NAME/tmp/

  Sostituisci quanto segue:

  • DATAFLOW_REGION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio europe-west1

    Il flag --region sostituisce la regione predefinita impostata nel server di metadati, nel client locale o nelle variabili di ambiente.

  • BUCKET_NAME: il nome del bucket Cloud Storage che hai copiato in precedenza
  • PROJECT_ID: l'ID progetto Google Cloud che hai copiato in precedenza

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

Visualizza i tuoi risultati

Quando esegui una pipeline utilizzando Dataflow, i risultati vengono archiviati in un bucket Cloud Storage. In questa sezione, verifica che la pipeline sia in esecuzione utilizzando la console Google Cloud o il terminale locale.

Console Google Cloud

Per visualizzare i risultati nella console Google Cloud, segui questi passaggi:

1. Nella console Google Cloud, vai alla pagina Job di Dataflow.

   Vai a Job

   La pagina Job mostra i dettagli del job wordcount, tra cui lo stato In esecuzione prima del job Riuscito.

2. Vai alla pagina Bucket di Cloud Storage.

   Vai a Bucket

3. Dall'elenco dei bucket del progetto, fai clic sul bucket di archiviazione creato in precedenza.

   Nella directory wordcount, vengono visualizzati i file di output creati dal job.

Terminale locale

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 utilizzato nel programma della pipeline.

2. Per visualizzare i risultati nei file di output, utilizza il comando gcloud storage cat:

   gcloud storage cat gs://BUCKET_NAME/results/outputs*

Modifica il codice della pipeline

La pipeline wordcount negli esempi precedenti fa una distinzione tra parole maiuscole e minuscole. I passaggi seguenti mostrano come modificare la pipeline in modo che la pipeline wordcount non sia sensibile alle maiuscole.

1. Sulla macchina locale, scarica la copia più recente del codice wordcount dal repository GitHub di Apache Beam.
2. Esegui la pipeline dal terminale locale:

   python wordcount.py --output outputs

3. Visualizza i risultati:

   more outputs*

4. Per uscire, premi Q.
5. In un editor a tua scelta, apri il file wordcount.py.
6. All'interno della funzione run, esamina i passaggi della pipeline:

   counts = (
           lines
           | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
           | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
           | 'GroupAndSum' >> beam.CombinePerKey(sum))

   Dopo split, le righe vengono divise in parole come stringhe.

7. Per modificare le stringhe in minuscolo, modifica la riga dopo split:

   counts = (
           lines
           | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
           | 'lowercase' >> beam.Map(str.lower)
           | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
           | 'GroupAndSum' >> beam.CombinePerKey(sum)) 

   Questa modifica mappa la funzione str.lower su ogni parola. Questa riga equivale a beam.Map(lambda word: str.lower(word)).
8. Salva il file ed esegui il job wordcount modificato:

   python wordcount.py --output outputs

9. Visualizza i risultati della pipeline modificata:

   more outputs*

10. Per uscire, premi Q.
11. Esegui la pipeline modificata sul servizio Dataflow:

    python wordcount.py \
        --region DATAFLOW_REGION \
        --input gs://dataflow-samples/shakespeare/kinglear.txt \
        --output gs://BUCKET_NAME/results/outputs \
        --runner DataflowRunner \
        --project PROJECT_ID \
        --temp_location gs://BUCKET_NAME/tmp/

    Sostituisci quanto segue:

    • DATAFLOW_REGION: la regione in cui vuoi eseguire il deployment del job Dataflow
    • BUCKET_NAME: nome del tuo bucket Cloud Storage
    • PROJECT_ID: il tuo ID progetto Google Cloud

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

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

   Vai a Bucket

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

4. 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 ognuno 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 dallgcloud CLI.

   gcloud auth revoke

Passaggi successivi

• Scopri di più sul modello di programmazione Apache Beam.
• Sviluppa in modo interattivo una pipeline utilizzando un blocco note Apache Beam.
• Scopri come progettare e creare la tua pipeline.
• Esamina gli esempi di Conteggio parole e Giochi mobile.