Questa pagina è stata tradotta dall'API Cloud Translation.

Sviluppa blocchi note Apache Beam con il runner interattivo

Utilizza il runner interattivo Apache Beam con i blocchi note JupyterLab per completare le seguenti attività:

Sviluppare iterativamente le pipeline.
Esamina il grafico della pipeline.
Analizza il singolo PCollections in un flusso di lavoro Read-Eval-print-loop (REPL).

Questi blocchi note Apache Beam sono resi disponibili Blocchi note gestiti dall'utente di Vertex AI Workbench, è un servizio contiene macchine virtuali notebook preinstallate con le versioni più recenti di data science e framework di machine learning.

Questa guida si concentra sulle funzionalità introdotte dai blocchi note Apache Beam, ma non mostra come creare un blocco note. Per ulteriori informazioni su Apache Beam, vedi consulta la guida alla programmazione di Apache Beam.

Supporto e limitazioni

I blocchi note Apache Beam supportano solo Python.
I segmenti della pipeline Apache Beam in esecuzione in questi blocchi note vengono eseguiti in un ambiente di test e non con un runner Apache Beam di produzione. Per avviare i blocchi note sulla Servizio Dataflow, esporta le pipeline create in Apache Beam un blocco note. Per ulteriori dettagli, vedi Avvia i job Dataflow da una pipeline creata nel blocco note.

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.

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Go to project selector

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

Abilita le API Compute Engine and Notebooks.

Abilita le API

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Go to project selector

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

Abilita le API Compute Engine and Notebooks.

Abilita le API

Prima di creare l'istanza di blocco note Apache Beam, abilita API per pipeline che utilizzano altri servizi, come Pub/Sub.

Se non specificata, l'istanza del blocco note viene eseguita account di servizio Compute Engine predefinito con il ruolo di editor del progetto IAM. Se il progetto in modo esplicito limita i ruoli dell'account di servizio, assicurati che abbia comunque un numero sufficiente per l'esecuzione dei blocchi note. Ad esempio, la lettura da un L'argomento Pub/Sub crea implicitamente una sottoscrizione e il tuo servizio deve avere un ruolo IAM Editor Pub/Sub. Di la lettura da una sottoscrizione Pub/Sub richiede solo Ruolo di sottoscrittore Pub/Sub IAM.

Al termine di questa guida, per evitare interruzioni nella fatturazione, elimina il le risorse che hai creato. Per ulteriori dettagli, vedi Pulizia.

avvia un'istanza di blocco note Apache Beam

Nella console Google Cloud, vai alla pagina Workbench di Dataflow.

Vai al workbench
Assicurati che sia attiva la scheda Blocchi note gestiti dall'utente.
Nella barra degli strumenti, fai clic su Crea nuova.
Nella sezione Ambiente, per Ambiente, seleziona Apache Beam.
(Facoltativo) Se vuoi eseguire blocchi note su una GPU, nella sezione Tipo di macchina seleziona un tipo di macchina che supporti le GPU, quindi Seleziona Installa il driver GPU NVIDIA automaticamente per me. Per ulteriori informazioni, vedi piattaforme GPU.
Nella sezione Networking, seleziona una subnet per la VM del blocco note.
(Facoltativo) Se vuoi configurare un'istanza di blocco note personalizzata, consulta Crea un'istanza di blocchi note gestiti dall'utente con proprietà specifiche.
Fai clic su Crea. Dataflow Workbench crea una nuova istanza di blocco note Apache Beam.
Dopo aver creato l'istanza del blocco note, il link Apri JupyterLab diventa attivo. Fai clic su Apri JupyterLab.

(Facoltativo) Installa le dipendenze

I blocchi note Apache Beam includono già Apache Beam Dipendenze del connettore Google Cloud installate. Se la pipeline contiene connettori personalizzati o PTransforms personalizzati che dipendono da librerie di terze parti, dopo aver creato un'istanza del blocco note. Per ulteriori informazioni, vedi Installa le dipendenze consulta la documentazione relativa ai blocchi note gestiti dall'utente.

Blocchi note Apache Beam di esempio

Dopo aver creato un'istanza di blocchi note gestiti dall'utente, aprila in JupyterLab. Nella scheda File della barra laterale JupyterLab, la cartella Esempi contiene blocchi note di esempio. Per ulteriori informazioni sull'utilizzo dei file JupyterLab, consulta l'articolo Utilizzo dei file. nella guida dell'utente di JupyterLab.

Sono disponibili i seguenti blocchi note:

Conteggio parole
Conteggio parole in modalità flusso
Streaming dei dati delle corse in taxi a New York
Apache Beam SQL nei blocchi note con confronti con le pipeline
SQL Apache Beam nei blocchi note con Dataflow Runner
Apache Beam SQL nei blocchi note
Conteggio parole Dataflow
Flink interattivo su larga scala
RunInference
Utilizza GPU con Apache Beam
Visualizzare i dati

La cartella Tutorial contiene tutorial aggiuntivi che spiegano le concetti fondamentali di Apache Beam. Sono disponibili i seguenti tutorial:

Operazioni di base
Operazioni Element Wise
Aggregazioni
Windows
Operazioni I/O
Streaming
Esercizi finali

Questi blocchi note includono testo esplicativo e blocchi di codice commentato per aiutarti comprendere i concetti di Apache Beam e l'utilizzo delle API. I tutorial offrono inoltre ti forniscono degli esercizi per mettere in pratica i concetti.

Le sezioni seguenti utilizzano un codice di esempio del blocco note Conteggio parole in streaming. Gli snippet di codice in questa guida e cosa si trova nel conteggio delle parole dei flussi di dati il blocco note potrebbe avere lievi discrepanze.

Crea un'istanza di blocco note

Vai a File > Nuovo > blocco note e seleziona un kernel Apache Beam 2.22 o versioni successive.

I blocchi note Apache Beam vengono creati in base al ramo master del SDK Apache Beam. Ciò significa che l'ultima versione del kernel mostrata potrebbe essere un passo avanti rispetto alla versione più recente dell'SDK.

Apache Beam è installato sull'istanza del blocco note, quindi includi i moduli interactive_runner e interactive_beam nel blocco note.

import apache_beam as beam
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
import apache_beam.runners.interactive.interactive_beam as ib

Se il blocco note utilizza altre API di Google, aggiungi le seguenti istruzioni di importazione:

from apache_beam.options import pipeline_options
from apache_beam.options.pipeline_options import GoogleCloudOptions
import google.auth

Imposta opzioni di interattività

La riga seguente consente di impostare per quanto tempo InteractiveRunner registra i dati provenienti da un'origine illimitata. In questo esempio, la durata è impostata su 10 minuti.

ib.options.recording_duration = '10m'

Puoi anche modificare il limite delle dimensioni di registrazione (in byte) per un'origine illimitata utilizzando la proprietà recording_size_limit.

# Set the recording size limit to 1 GB.
ib.options.recording_size_limit = 1e9

Per ulteriori opzioni interattive, consulta la classe interactive_beam.options.

Crea la tua pipeline

Inizializza la pipeline utilizzando un oggetto InteractiveRunner.

options = pipeline_options.PipelineOptions(flags={})

# Set the pipeline mode to stream the data from Pub/Sub.
options.view_as(pipeline_options.StandardOptions).streaming = True

# Set the project to the default project in your current Google Cloud environment.
# The project is used to create a subscription to the Pub/Sub topic.
_, options.view_as(GoogleCloudOptions).project = google.auth.default()

p = beam.Pipeline(InteractiveRunner(), options=options)

Leggere e visualizzare i dati

L'esempio seguente mostra una pipeline Apache Beam che crea un'istanza all'argomento Pub/Sub specificato e legge la sottoscrizione.

words = p | "read" >> beam.io.ReadFromPubSub(topic="projects/pubsub-public-data/topics/shakespeare-kinglear")

La pipeline conteggia le parole per finestre dall'origine. Crea con una durata di 10 secondi.

windowed_words = (words
   | "window" >> beam.WindowInto(beam.window.FixedWindows(10)))

Quando i dati vengono inseriti in una finestra, le parole vengono conteggiate in base alla finestra.

windowed_word_counts = (windowed_words
   | "count" >> beam.combiners.Count.PerElement())

Il metodo show() visualizza la PCollection risultante nel blocco note.

ib.show(windowed_word_counts, include_window_info=True)

Il metodo show che visualizza una PCollection in formato tabulare.

Puoi definire l'ambito del set di risultati arretrato da show() impostando due facoltativi parametri: n e duration.

Imposta n per limitare il set di risultati in modo che mostri al massimo n numero di elementi, ad esempio 20. Se il criterio n non viene configurato, il comportamento predefinito prevede l'elenco delle app più recenti elementi acquisiti fino al termine della registrazione di origine.
Imposta duration per limitare il risultato impostato a un numero specifico di secondi di dati a partire dall'inizio della registrazione di origine. Se duration non è impostato, il comportamento predefinito prevede l'elenco di tutti gli elementi fino alla è finita.

Se sono impostati entrambi i parametri facoltativi, show() si interrompe ogni volta che viene raggiunta una delle due soglie. Nell'esempio seguente, show() restituisce al massimo 20 elementi calcolati in base ai primi 30 secondi di dati provenienti dalle sorgenti registrate.

ib.show(windowed_word_counts, include_window_info=True, n=20, duration=30)

Per mostrare le visualizzazioni dei dati, trasmetti visualize_data=True al Metodo show(). Puoi applicare più filtri alle visualizzazioni. La che ti consente di filtrare per etichetta e asse:

Il metodo show che visualizza una PCollection come un ricco insieme di elementi dell'interfaccia utente filtrabili.

Per garantire la rigiocabilità durante la prototipazione delle pipeline di flusso, Le chiamate al metodo show() riutilizzano i dati acquisiti per impostazione predefinita. Per modificare questa impostazione e fare in modo che il metodo show() recuperi sempre nuovi dati, imposta interactive_beam.options.enable_capture_replay = False. Inoltre, se aggiungi seconda origine illimitata al blocco note, i dati della precedente viene ignorato.

Un'altra visualizzazione utile nei blocchi note Apache Beam è DataFrame Panda. L'esempio seguente converte prima le parole in lettere minuscole, quindi calcola la frequenza di ogni parola.

windowed_lower_word_counts = (windowed_words
   | beam.Map(lambda word: word.lower())
   | "count" >> beam.combiners.Count.PerElement())

Il metodo collect() fornisce l'output in un DataFrame Pandas.

ib.collect(windowed_lower_word_counts, include_window_info=True)

Il metodo collect che rappresenta una PCollection in un DataFrame Pandas.

La modifica e la nuova esecuzione di una cella è una pratica comune nel blocco note sviluppo del prodotto. Quando modifichi e riesegui una cella in un blocco note Apache Beam, la cella non annulla l'azione prevista del codice nella cella originale. Per Ad esempio, se una cella aggiunge PTransform a una pipeline, eseguendo di nuovo la cella aggiunge un ulteriore PTransform alla pipeline. Se vuoi cancellare lo stato, riavviare il kernel e quindi eseguire nuovamente le celle.

Visualizza i dati tramite l'ispettore Interactive Beam

Potrebbe distrarre introspezionare i dati di un PCollection chiama costantemente show() e collect(), soprattutto quando l'output occupa molto spazio sullo schermo e la navigazione tra i un blocco note. Puoi anche confrontare più PCollections affiancati con verificare se una trasformazione funziona come previsto. Ad esempio, quando PCollection passa attraverso una trasformazione e produce l'altra. Per questi casi d'uso, L'ispettore interattivo Beam è una soluzione pratica.

Lo strumento di controllo interattivo Beam viene fornito come estensione JupyterLab apache-beam-jupyterlab-sidepanel preinstallate nel blocco note Apache Beam. Con l'estensione, puoi esaminare in modo interattivo lo stato delle pipeline e dei dati associati ogni PCollection senza richiamare esplicitamente show() o collect().

Puoi aprire la finestra di ispezione in tre modi:

Fai clic su Interactive Beam sulla barra dei menu in alto di JupyterLab. Nel menu a discesa, Individua Open Inspector e fai clic per aprire la finestra di ispezione.
Utilizza la pagina Avvio app. Se non è aperta alcuna pagina Avvio app, fai clic su File -> New Launcher per aprirlo. Nella pagina Avvio app, individua Interactive Beam e fai clic su Open Inspector per aprire la finestra di ispezione.
Utilizza la tavolozza dei comandi. Sulla barra dei menu JupyterLab, fai clic su View > Activate Command Palette. Nella finestra di dialogo, cerca Interactive Beam per elencare tutti le opzioni dell'estensione. Fai clic su Open Inspector per aprire la finestra di ispezione.

Quando la finestra di controllo sta per aprire:

Se c'è esattamente un blocco note aperto, la finestra di ispezione si connette automaticamente che le sono assegnati.
Se non è aperto alcun blocco note, viene visualizzata una finestra di dialogo che ti consente di selezionare un kernel.
Se sono aperti più blocchi note, viene visualizzata una finestra di dialogo che ti consente di selezionare la sessione corrispondente.

Ti consigliamo di aprire almeno un blocco note e selezionare un kernel prima di aprire la finestra di ispezione. Se apri una finestra di ispezione con un kernel prima l'apertura di qualsiasi blocco note, in un secondo momento, quando aprirai un blocco note per connetterti inspector, devi selezionare Interactive Beam Inspector Session da Use Kernel from Preferred Session. Un inspector e un blocco note sono connessi quando condividono la stessa sessione, non sessioni diverse create dalla stessa in un kernel. Se selezioni lo stesso kernel da Start Preferred Kernel, viene creata una una nuova sessione indipendente dalle sessioni esistenti di blocchi note aperti ispettori.

Puoi aprire più controlli per un blocco note aperto e organizzare in ispettori trascinando liberamente le schede nell'area di lavoro.

Apri due ispettori e disponili uno accanto all'altro

La pagina di controllo si aggiorna automaticamente quando esegui celle nella un blocco note. La pagina elenca le pipeline e i criteri PCollections definiti nel un blocco note connesso. I PCollections sono organizzati in base alle pipeline a cui appartengono e puoi comprimerle facendo clic sulla pipeline di intestazione.

Per gli elementi nelle pipeline e nell'elenco PCollections, al clic viene mostrata la finestra di ispezione mostra le visualizzazioni corrispondenti sul lato destro:

Se si tratta di un PCollection, la finestra di controllo esegue il rendering dei propri dati (in modo dinamico se i dati è ancora disponibile per PCollections illimitato) con widget aggiuntivi da ottimizzare la visualizzazione dopo aver fatto clic sul pulsante APPLY.

Poiché l'inspector e il blocco note aperto condividono lo stesso kernel si bloccano a vicenda. Ad esempio, se il blocco note è impegnato nell'esecuzione del codice, la finestra di ispezione non si aggiorna fino al completamento dell'esecuzione. Al contrario, se vuoi il codice immediatamente nel blocco note mentre la finestra di ispezione visualizzare un PCollection in modo dinamico, devi fare clic sul pulsante STOP per interrompere la visualizzazione e rilasciare preventivamente il kernel nel blocco note.
Se si tratta di una pipeline, lo strumento di controllo mostra il grafico della pipeline.

Potresti notare pipeline anonime. Queste pipeline hanno PCollections a cui puoi accedere, ma a cui non fa più riferimento durante la sessione. Ad esempio:

p = beam.Pipeline()
pcoll = p | beam.Create([1, 2, 3])

p = beam.Pipeline()

L'esempio precedente crea una pipeline vuota p e una pipeline anonima che contiene un solo PCollection pcoll. Puoi accedere alla pipeline anonima mediante pcoll.pipeline.

Puoi attivare/disattivare la pipeline e l'elenco PCollection per risparmiare spazio visualizzazioni di grandi dimensioni. Elenco a sinistra Toggel

Comprendi lo stato di registrazione di una pipeline

Oltre alle visualizzazioni, puoi anche controllare lo stato della registrazione per una o tutte le pipeline nell'istanza del blocco note chiamando describe.

# Return the recording status of a specific pipeline. Leave the parameter list empty to return
# the recording status of all pipelines.
ib.recordings.describe(p)

Il metodo describe() fornisce i seguenti dettagli:

Dimensioni totali (in byte) di tutte le registrazioni per la pipeline su disco
Ora di inizio del processo di registrazione in background (in secondi dall'epoca Unix)
Stato attuale della pipeline del job di registrazione in background
Variabile Python per la pipeline

Avvia i job Dataflow da una pipeline creata nel tuo blocco note

(Facoltativo) Prima di utilizzare il blocco note per eseguire job Dataflow, riavvia il kernel, eseguire nuovamente tutte le celle e verificare l'output. Se salti questo passaggio, gli stati nascosti nel blocco note potrebbero influire sul grafico del job nella pipeline .
Abilitare l'API Dataflow.

Aggiungi la seguente istruzione di importazione:

from apache_beam.runners import DataflowRunner

Passa la pipeline opzioni.

# Set up Apache Beam pipeline options.
options = pipeline_options.PipelineOptions()

# Set the project to the default project in your current Google Cloud
# environment.
_, options.view_as(GoogleCloudOptions).project = google.auth.default()

# Set the Google Cloud region to run Dataflow.
options.view_as(GoogleCloudOptions).region = 'us-central1'

# Choose a Cloud Storage location.
dataflow_gcs_location = 'gs://<change me>/dataflow'

# Set the staging location. This location is used to stage the
# Dataflow pipeline and SDK binary.
options.view_as(GoogleCloudOptions).staging_location = '%s/staging' % dataflow_gcs_location

# Set the temporary location. This location is used to store temporary files
# or intermediate results before outputting to the sink.
options.view_as(GoogleCloudOptions).temp_location = '%s/temp' % dataflow_gcs_location

# If and only if you are using Apache Beam SDK built from source code, set
# the SDK location. This is used by Dataflow to locate the SDK
# needed to run the pipeline.
options.view_as(pipeline_options.SetupOptions).sdk_location = (
    '/root/apache-beam-custom/packages/beam/sdks/python/dist/apache-beam-%s0.tar.gz' %
    beam.version.__version__)

Puoi modificare i valori dei parametri. Ad esempio, puoi modificare region valore da us-central1.

Esegui la pipeline con DataflowRunner. Questo passaggio esegue il job dal servizio Dataflow.
```
runner = DataflowRunner()
runner.run_pipeline(p, options=options)
```
p è un oggetto pipeline di creazione una pipeline di Cloud Shell.

Per un esempio su come eseguire questa conversione in un blocco note interattivo, consulta il blocco note Conteggio parole di Dataflow nell'istanza del blocco note.

In alternativa, puoi esportare il blocco note come script eseguibile, modificare generato il file .py seguendo i passaggi precedenti e poi eseguire il deployment pipeline di dati a Dataflow completamente gestito di Google Cloud.

Salva il blocco note

Notebooks che crei vengono salvati localmente nell'istanza del blocco note in esecuzione. Se reset o l'istanza del blocco note durante lo sviluppo, i nuovi blocchi note rimangono valide purché siano create nella directory /home/jupyter. Tuttavia, se viene eliminata un'istanza di blocco note, vengono eliminati anche i relativi blocchi note.

Per conservare i blocchi note per l'uso futuro, scaricali in locale sul tuo workstation, salvale su GitHub, o esportarle in un formato file diverso.

Salva il blocco note su dischi permanenti aggiuntivi

Se vuoi mantenere il tuo lavoro, ad esempio blocchi note e script, in vari le istanze di blocco note, archiviarle in Persistent Disk.

Crea o allega un Persistent Disk (Disco permanente). Segui le istruzioni per utilizzare ssh per connettersi alla VM dell'istanza del blocco note nell'istanza di Cloud Shell aperta.

Attenzione: non inviare comandi dai terminali all'interno dell'istanza del blocco note.
Annota la directory in cui è montato il Persistent Disk, ad esempio: /mnt/myDisk.
Modifica i dettagli della VM dell'istanza del blocco note per aggiungere una voce alla chiave Custom metadata: - container-custom-params; valore: -v /mnt/myDisk:/mnt/myDisk.
Fai clic su Salva.
Per aggiornare queste modifiche, reimposta l'istanza del blocco note.
Dopo il ripristino, fai clic su Apri JupyterLab. it potrebbe volerci del tempo prima che la UI JupyterLab diventi disponibile. Dopo l'interfaccia utente apri un terminale ed esegui questo comando: ls -al /mnt Dovrebbe essere presente la directory /mnt/myDisk.

Ora puoi salvare il tuo lavoro nella directory /mnt/myDisk. Anche se il blocco note viene eliminata, il Persistent Disk è presente nel progetto. Tu quindi può collegare questo Persistent Disk ad altre istanze del blocco note.

Esegui la pulizia

Dopo aver finito di utilizzare l'istanza di blocco note Apache Beam, esegui la pulizia create su Google Cloud l'arresto dell'istanza del blocco note.

Passaggi successivi

Scopri le funzionalità avanzate che che puoi utilizzare con i tuoi blocchi note Apache Beam. Le funzionalità avanzate includono i seguenti flussi di lavoro: