Impostare le opzioni della pipeline Dataflow

Questa pagina spiega come impostare le opzioni di pipeline per i job Dataflow. Queste opzioni di pipeline configurano come e dove viene eseguita la pipeline e quali risorse utilizza.

L'esecuzione della pipeline è separata dall'esecuzione del programma Apache Beam. Il programma Apache Beam che hai scritto costruisce una pipeline per l'esecuzione differita. Ciò significa che il programma genera una serie di passaggi che qualsiasi runner Apache Beam supportato può eseguire. I runner compatibili includono il runner Dataflow su Google Cloud e il runner diretto che esegue la pipeline direttamente in un ambiente locale.

Puoi passare i parametri a un job Dataflow in fase di esecuzione. Per ulteriori informazioni sull'impostazione delle opzioni della pipeline in fase di esecuzione, consulta Configurare le opzioni della pipeline.

Nota: se stai creando un modello classico che accetta opzioni di pipeline, devi utilizzare l'interfaccia ValueProvider nel codice della pipeline. Per maggiori informazioni, consulta Utilizzare i parametri di runtime nel codice della pipeline.

Utilizzare le opzioni di pipeline con gli SDK Apache Beam

Puoi utilizzare i seguenti SDK per impostare le opzioni di pipeline per i job Dataflow:

• SDK Apache Beam per Python
• SDK Apache Beam per Java
• SDK Apache Beam per Go

Per utilizzare gli SDK, imposta il programma di esecuzione della pipeline e altri parametri di esecuzione utilizzando la classe SDK Apache Beam PipelineOptions.

Esistono due metodi per specificare le opzioni della pipeline:

• Imposta le opzioni della pipeline in modo programmatico fornendo un elenco di opzioni.
• Imposta le opzioni della pipeline direttamente sulla riga di comando quando esegui il codice della pipeline.

Impostare le opzioni della pipeline in modo programmatico

Puoi impostare le opzioni della pipeline in modo programmatico creando e modificando un oggetto PipelineOptions.

Java

Costruisci un oggetto PipelineOptions utilizzando il metodo PipelineOptionsFactory.fromArgs.

Per un esempio, consulta la sezione Esempio di lancio su Dataflow in questa pagina.

Python

Crea un oggetto PipelineOptions.

Per un esempio, consulta la sezione Esempio di lancio su Dataflow in questa pagina.

Vai

L'impostazione delle opzioni di pipeline in modo programmatico utilizzando PipelineOptions non è supportata nell'SDK Apache Beam per Go. Utilizza gli argomenti della riga di comando Go.

Per un esempio, consulta la sezione Esempio di lancio su Dataflow in questa pagina.

Impostare le opzioni della pipeline sulla riga di comando

Puoi impostare le opzioni della pipeline utilizzando gli argomenti della riga di comando.

Java

La sintassi di esempio seguente proviene dalla pipeline WordCount nel quickstart di Java.

mvn -Pdataflow-runner compile exec:java \
  -Dexec.mainClass=org.apache.beam.examples.WordCount \
  -Dexec.args="--project=PROJECT_ID \
  --gcpTempLocation=gs://BUCKET_NAME/temp/ \
  --output=gs://BUCKET_NAME/output \
  --runner=DataflowRunner \
  --region=REGION"

Sostituisci quanto segue:

• PROJECT_ID: il tuo ID progetto Google Cloud
• BUCKET_NAME: il nome del bucket Cloud Storage
• REGION: una regione Dataflow, us-central1

Python

La sintassi dell'esempio seguente è tratta dalla pipeline WordCount nel quickstart di Python.

python -m apache_beam.examples.wordcount \
  --region DATAFLOW_REGION \
  --input gs://dataflow-samples/shakespeare/kinglear.txt \
  --output gs://STORAGE_BUCKET/results/outputs \
  --runner DataflowRunner \
  --project PROJECT_ID \
  --temp_location gs://STORAGE_BUCKET/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 dei metadati, nel client locale o nelle variabili di ambiente.

• STORAGE_BUCKET: il nome del bucket Cloud Storage

• PROJECT_ID: l'ID progetto Google Cloud

Vai

La sintassi di esempio riportata di seguito proviene dalla pipeline WordCount nella guida rapida di Go.

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. Il flag --region sostituisce la regione predefinita impostata nel server dei metadati, nel client locale o nelle variabili di ambiente.

Impostare le opzioni della pipeline sperimentale

Negli SDK Java, Python e Go, l'experiments opzione pipeline attiva le funzionalità sperimentali o pre-GA di Dataflow.

Impostato in modo programmatico

Per impostare l'opzione experiments in modo programmatico, utilizza la seguente sintassi.

Java

Nell'oggetto PipelineOptions, includi l'opzione experiments utilizzando la seguente sintassi. Questo esempio imposta la dimensione del disco di avvio su 80 GB con il flag dell'esperimento.

options.setExperiments("streaming_boot_disk_size_gb=80")

Per un esempio che mostra come creare l'oggetto PipelineOptions, consulta la sezione Esempio di lancio su Dataflow di questa pagina.

Python

Nell'oggetto PipelineOptions, includi l'opzione experiments utilizzando la seguente sintassi. Questo esempio imposta la dimensione del disco di avvio su 80 GB con il flag dell'esperimento.

beam_options = PipelineOptions(
  beam_args,
  experiments=['streaming_boot_disk_size_gb=80'])

Per un esempio che mostra come creare l'oggetto PipelineOptions, consulta la sezione Esempio di lancio su Dataflow di questa pagina.

Vai

L'impostazione delle opzioni di pipeline in modo programmatico utilizzando PipelineOptions non è supportata nell'SDK Apache Beam per Go. Utilizza gli argomenti della riga di comando Go.

Impostato sulla riga di comando

Per impostare l'opzione experiments sulla riga di comando, utilizza la seguente sintassi.

Java

Questo esempio imposta la dimensione del disco di avvio su 80 GB con il flag dell'esperimento.

--experiments=streaming_boot_disk_size_gb=80

Python

Questo esempio imposta la dimensione del disco di avvio su 80 GB con il flag dell'esperimento.

--experiments=streaming_boot_disk_size_gb=80

Vai

Questo esempio imposta la dimensione del disco di avvio su 80 GB con il flag dell'esperimento.

--experiments=streaming_boot_disk_size_gb=80

Impostato in un modello

Per attivare una funzionalità sperimentale durante l'esecuzione di un modello Dataflow, utilizza il flag --additional-experiments.

Modello classico

gcloud dataflow jobs run JOB_NAME --additional-experiments=EXPERIMENT[,...]

modello flessibile

gcloud dataflow flex-template run JOB_NAME --additional-experiments=EXPERIMENT[,...]

Accedere all'oggetto delle opzioni della pipeline

Quando crei l'oggetto Pipeline nel programma Apache Beam, passa PipelineOptions. Quando il servizio Dataflow esegue la pipeline, invia una copia del PipelineOptions a ciascun worker.

Java

Accedi a PipelineOptions all'interno di qualsiasi istanza DoFn della trasformazione ParDo utilizzando il metodo ProcessContext.getPipelineOptions.

Python

Questa funzionalità non è supportata nell'SDK Apache Beam per Python.

Vai

Accedi alle opzioni della pipeline utilizzando beam.PipelineOptions.

Lancia su Dataflow

Esegui il job sulle risorse Google Cloud gestite utilizzando il servizio di esecuzione Dataflow. L'esecuzione della pipeline con Dataflow crea un job Dataflow che utilizza le risorse Compute Engine e Cloud Storage nel progetto Google Cloud. Per informazioni sulle autorizzazioni di Dataflow, consulta Sicurezza e autorizzazioni di Dataflow.

I job di Dataflow utilizzano Cloud Storage per archiviare i file temporanei durante l'esecuzione della pipeline. Per evitare l'addebito per costi di archiviazione non necessari, disattiva la funzionalità di eliminazione temporanea sui bucket utilizzati dai job di Dataflow per l'archiviazione temporanea. Per ulteriori informazioni, consulta Rimuovere una regola di eliminazione temporanea da un bucket.

Nota: digitare Ctrl+C dalla riga di comando non annulla il job. Il servizio Dataflow sta ancora eseguendo il job su Google Cloud. Per annullare il job, utilizza l'interfaccia di monitoraggio di Dataflow o l'interfaccia a riga di comando di Dataflow.

Imposta le opzioni richieste

Per eseguire la pipeline utilizzando Dataflow, imposta le seguenti opzioni della pipeline:

Java

• project: l'ID del tuo progetto Google Cloud.
• runner: il runner della pipeline che esegue la pipeline. Per l'esecuzione su Google Cloud, deve essere DataflowRunner.
• gcpTempLocation: un percorso di Cloud Storage per la gestione in anteprima della maggior parte dei file temporanei da parte di Dataflow. Se vuoi specificare un bucket, devi crearlo in anticipo. Se non impostate gcpTempLocation, potete impostare l'opzione pipeline tempLocation e gcpTempLocation viene impostato sul valore di tempLocation. Se non viene specificato nessuno dei due, viene creato un gcpTempLocation predefinito.

• stagingLocation: un percorso Cloud Storage per la gestione in anteprima dei file binari da parte di Dataflow. Se utilizzi l'SDK Apache Beam 2.28 o versioni successive, non impostare questa opzione. Per l'SDK Apache Beam 2.28 o versioni precedenti, se non imposti questa opzione, per la posizione di staging viene utilizzato quanto specificato per tempLocation.

  Viene creato un gcpTempLocation predefinito se né questo né tempLocation sono specificati. Se tempLocation è specificato e gcpTempLocation non lo è, tempLocation deve essere un percorso Cloud Storage e gcpTempLocation assume questo valore per impostazione predefinita. Se tempLocation non è specificato e gcpTempLocation lo è, tempLocation non viene compilato.

Python

• project: il tuo ID progetto Google Cloud.
• region: la regione per il job Dataflow.
• runner: il runner della pipeline che esegue la pipeline. Per l'esecuzione su Google Cloud, deve essere DataflowRunner.
• temp_location: un percorso Cloud Storage per la gestione in anteprima dei file temporanei dei job creati durante l'esecuzione della pipeline da parte di Dataflow.

Vai

• project: il tuo ID progetto Google Cloud.
• region: la regione per il job Dataflow.
• runner: il runner della pipeline che esegue la pipeline. Per l'esecuzione su Google Cloud, deve essere dataflow.
• staging_location: un percorso Cloud Storage per la gestione in anteprima dei file temporanei dei job creati durante l'esecuzione della pipeline da parte di Dataflow.

Impostare le opzioni della pipeline in modo programmatico

Il seguente codice di esempio mostra come creare una pipeline impostando programmaticamente il runner e altre opzioni necessarie per eseguirla utilizzando Dataflow.

Java

// Create and set your PipelineOptions.
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);

// For cloud execution, set the Google Cloud project, staging location,
// and set DataflowRunner.
options.setProject("my-project-id");
options.setStagingLocation("gs://my-bucket/binaries");
options.setRunner(DataflowRunner.class);

// Create the Pipeline with the specified options.
Pipeline p = Pipeline.create(options);

Python

import argparse

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

parser = argparse.ArgumentParser()
# parser.add_argument('--my-arg', help='description')
args, beam_args = parser.parse_known_args()

# Create and set your PipelineOptions.
# For Cloud execution, specify DataflowRunner and set the Cloud Platform
# project, job name, temporary files location, and region.
# For more information about regions, check:
# https://cloud.google.com/dataflow/docs/concepts/regional-endpoints
beam_options = PipelineOptions(
    beam_args,
    runner='DataflowRunner',
    project='my-project-id',
    job_name='unique-job-name',
    temp_location='gs://my-bucket/temp',
    region='us-central1')
# Note: Repeatable options like dataflow_service_options or experiments must
# be specified as a list of string(s).
# e.g. dataflow_service_options=['enable_prime']

# Create the Pipeline with the specified options.
with beam.Pipeline(options=beam_options) as pipeline:
  pass  # build your pipeline here.

Vai

L'SDK Apache Beam per Go utilizza gli argomenti della riga di comando Go. Utilizza flag.Set() per impostare i valori dei flag.

// Use the Go flag package to parse custom options.
flag.Parse()

// Set the required options programmatically.
// For Cloud execution, specify the Dataflow runner, Google Cloud
// project ID, region, and staging location.
// For more information about regions, see
// https://cloud.google.com/dataflow/docs/concepts/regional-endpoints
flag.Set("runner", "dataflow")
flag.Set("project", "my-project-id")
flag.Set("region", "us-central1")
flag.Set("staging_location", "gs://my-bucket/binaries")

beam.Init()

// Create the Pipeline.
p := beam.NewPipeline()
s := p.Root()

Dopo aver costruito la pipeline, specifica tutte le letture, le trasformazioni e le scritture della pipeline ed eseguila.

Utilizzare le opzioni della pipeline dalla riga di comando

L'esempio seguente mostra come utilizzare le opzioni della pipeline specificate sulla riga di comando. Questo esempio non imposta le opzioni della pipeline in modo programmatico.

Java

// Set your PipelineOptions to the specified command-line options
MyOptions options = PipelineOptionsFactory.fromArgs(args).withValidation();

// Create the Pipeline with the specified options.
Pipeline p = Pipeline.create(options);

Python

Utilizza il modulo argparse di Python per analizzare le opzioni della riga di comando.

# Use Python argparse module to parse custom arguments
import argparse

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# For more details on how to use argparse, take a look at:
#   https://docs.python.org/3/library/argparse.html
parser = argparse.ArgumentParser()
parser.add_argument(
    '--input-file',
    default='gs://dataflow-samples/shakespeare/kinglear.txt',
    help='The file path for the input text to process.')
parser.add_argument(
    '--output-path', required=True, help='The path prefix for output files.')
args, beam_args = parser.parse_known_args()

# Create the Pipeline with remaining arguments.
beam_options = PipelineOptions(beam_args)
with beam.Pipeline(options=beam_options) as pipeline:
  lines = (
      pipeline
      | 'Read files' >> beam.io.ReadFromText(args.input_file)
      | 'Write files' >> beam.io.WriteToText(args.output_path))

Vai

Utilizza il pacchetto Go flag per analizzare le opzioni della riga di comando. Devi analizzare le opzioni prima di chiamare beam.Init(). In questo esempio, output è un'opzione della riga di comando.

// Define configuration options
var (
  output = flag.String("output", "", "Output file (required).")
)

// Parse options before beam.Init()
flag.Parse()

beam.Init()

// Input validation must be done after beam.Init()
if *output == "" {
  log.Fatal("No output provided!")
}

p := beam.NewPipeline()

Dopo aver costruito la pipeline, specifica tutte le letture, le trasformazioni e le scritture della pipeline, quindi eseguila.

Controllare le modalità di esecuzione

Quando un programma Apache Beam esegue una pipeline su un servizio come Dataflow, può eseguirla in modo asincrono o bloccarsi fino al completamento della pipeline. Puoi modificare questo comportamento seguendo le indicazioni riportate di seguito.

Java

Quando un programma Java Apache Beam esegue una pipeline su un servizio come Dataflow, in genere viene eseguita in modo asincrono. Per eseguire una pipeline e attendere il completamento del job, imposta DataflowRunner come eseguitrice della pipeline e chiama esplicitamente pipeline.run().waitUntilFinish().

Quando utilizzi DataflowRunner e chiami waitUntilFinish() sull'oggetto PipelineResult restituito da pipeline.run(), la pipeline viene eseguita su Google Cloud, ma il codice locale attende il completamento del job cloud e restituisce l'oggetto DataflowPipelineJob finale. Durante l'esecuzione del job, il servizio Dataflow stampa gli aggiornamenti dello stato del job e i messaggi della console mentre è in attesa.

Python

Quando un programma Python Apache Beam esegue una pipeline su un servizio come Dataflow, in genere viene eseguita in modo asincrono. Per bloccare il processo fino al completamento della pipeline, utilizza il metodo wait_until_finish() dell'oggetto PipelineResult, restituito dal metodo run() del runner.

Vai

Quando un programma Apache Beam Go esegue una pipeline su Dataflow, è sincrono per impostazione predefinita e si blocca fino al completamento della pipeline. Se non vuoi bloccare, hai due opzioni:

1. Avvia il job in una routine Go.

   go func() {
     pr, err := beamx.Run(ctx, p)
     if err != nil {
       // Handle the error.
     }
     // Send beam.PipelineResult into a channel.
     results <- pr
   }()
   // Do other operations while the pipeline runs.
   

2. Utilizza il flag --async della riga di comando, che si trova nel pacchetto jobopts.

Per visualizzare i dettagli di esecuzione, monitorare l'avanzamento e verificare lo stato di completamento del job, utilizza l'interfaccia di monitoraggio di Dataflow o l'interfaccia a riga di comando di Dataflow.

Utilizzare le origini di streaming

Java

Se la pipeline legge da un'origine dati illimitata, come Pub/Sub, viene eseguita automaticamente in modalità di streaming.

Python

Se la pipeline utilizza un'origine dati illimitata, come Pub/Sub, devi impostare l'opzione streaming su true.

Vai

Se la pipeline legge da un'origine dati illimitata, come Pub/Sub, viene eseguita automaticamente in modalità di streaming.

Per impostazione predefinita, i job di streaming utilizzano un tipo di macchina di Compute Engine di n1-standard-2 o superiore.

Nota: se la pipeline utilizza origini dati e destinazioni illimitate, devi scegliere una strategia di definizione di finestre per le tue PCollection illimitate prima di utilizzare qualsiasi aggregazione, ad esempio un GroupByKey.

Avvia localmente

Anziché eseguire la pipeline su risorse cloud gestite, puoi scegliere di eseguire la pipeline localmente. L'esecuzione locale presenta alcuni vantaggi per il test, il debug o l'esecuzione della pipeline su piccoli set di dati. Ad esempio, l'esecuzione locale rimuove la dipendenza dal servizio Dataflow remoto e dal progetto Google Cloud associato.

Quando utilizzi l'esecuzione locale, devi eseguire la pipeline con set di dati sufficientemente piccoli da poter essere inseriti nella memoria locale. Puoi creare un piccolo set di dati in memoria utilizzando una trasformazione Create oppure puoi utilizzare una trasformazione Read per lavorare con piccoli file locali o remoti. L'esecuzione locale in genere offre un modo più rapido e semplice per eseguire test e debug con meno dipendenze esterne, ma è limitata dalla memoria disponibile nell'ambiente locale.

Il seguente codice di esempio mostra come creare una pipeline che viene eseguita nel tuo ambiente locale.

Java

// Create and set our Pipeline Options.
PipelineOptions options = PipelineOptionsFactory.create();

// Create the Pipeline with the specified options.
Pipeline p = Pipeline.create(options);

Nota: per la modalità locale, non è necessario impostare il runner perché DirectRunner è già l'impostazione predefinita. Tuttavia, devi includere esplicitamenteDirectRunner come dipendenza o aggiungerlo al classpath.

Python

import argparse

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

parser = argparse.ArgumentParser()
# parser.add_argument('--my-arg')
args, beam_args = parser.parse_known_args()

# Create and set your Pipeline Options.
beam_options = PipelineOptions(beam_args)
args = beam_options.view_as(MyOptions)

with beam.Pipeline(options=beam_options) as pipeline:
  lines = (
      pipeline
      | beam.io.ReadFromText(args.input)
      | beam.io.WriteToText(args.output))

Nota: per la modalità locale, non è necessario impostare il runner perché DirectRunner è già l'impostazione predefinita.

Vai

// Parse options before beam.Init()
flag.Parse()

beam.Init()

p := beam.NewPipeline()

Nota: per la modalità locale, non è necessario impostare il runner perché direct è già il valore predefinito.

Dopo aver creato la pipeline, eseguila.

Creare opzioni pipeline personalizzate

Puoi aggiungere le tue opzioni personalizzate oltre a quelle standard.PipelineOptions La riga di comando di Apache Beam può anche analizzare le opzioni personalizzate utilizzando gli argomenti della riga di comando specificati nello stesso formato.

Java

Per aggiungere le tue opzioni, definisci un'interfaccia con metodi getter e setter per ogni opzione, come nell'esempio seguente:

public interface MyOptions extends PipelineOptions {
  String getMyCustomOption();
  void setMyCustomOption(String myCustomOption);
}

Python

Per aggiungere le tue opzioni, utilizza il metodo add_argument() (che si comporta esattamente come il modulo argparse standard di Python), come nell'esempio seguente:

from apache_beam.options.pipeline_options import PipelineOptions

class MyOptions(PipelineOptions):
  @classmethod
  def _add_argparse_args(cls, parser):
    parser.add_argument('--input')
    parser.add_argument('--output')

Vai

Per aggiungere le tue opzioni, utilizza il pacchetto di flag Go come mostrato nell' esempio seguente:

var (
  input  = flag.String("input", "", "")
  output = flag.String("output", "", "")
)

Puoi anche specificare una descrizione, che viene visualizzata quando un utente passa --help come argomento della riga di comando, e un valore predefinito.

Java

Imposta la descrizione e il valore predefinito utilizzando le annotazioni, come segue:

public interface MyOptions extends PipelineOptions {
  @Description("My custom command line argument.")
  @Default.String("DEFAULT")
  String getMyCustomOption();
  void setMyCustomOption(String myCustomOption);
}

Ti consigliamo di registrare l'interfaccia con PipelineOptionsFactory e poi di passarla quando crei l'oggetto PipelineOptions. Quando registri l'interfaccia con PipelineOptionsFactory, --help può trovare l'interfaccia delle opzioni personalizzate e aggiungerla all'output del comando --help. PipelineOptionsFactory convalida che le opzioni personalizzate siano compatibili con tutte le altre opzioni registrate.

Il seguente codice di esempio mostra come registrare l'interfaccia delle opzioni personalizzate con PipelineOptionsFactory:

PipelineOptionsFactory.register(MyOptions.class);
MyOptions options = PipelineOptionsFactory.fromArgs(args)
                                          .withValidation()
                                          .as(MyOptions.class);

Ora la pipeline può accettare --myCustomOption=value come argomento della riga di comando.

Python

Imposta la descrizione e il valore predefinito come segue:

from apache_beam.options.pipeline_options import PipelineOptions

class MyOptions(PipelineOptions):
  @classmethod
  def _add_argparse_args(cls, parser):
    parser.add_argument(
        '--input',
        default='gs://dataflow-samples/shakespeare/kinglear.txt',
        help='The file path for the input text to process.')
    parser.add_argument(
        '--output', required=True, help='The path prefix for output files.')

Vai

Imposta la descrizione e il valore predefinito come segue:

var (
  input  = flag.String("input", "gs://MY_STORAGE_BUCKET/input", "Input for the pipeline")
  output = flag.String("output", "gs://MY_STORAGE_BUCKET/output", "Output for the pipeline")
)