Impostare le opzioni della pipeline Dataflow

Java

Crea un oggetto PipelineOptions utilizzando il metodo PipelineOptionsFactory.fromArgs.

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

Python

Crea un oggetto PipelineOptions.

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

Vai

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

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

Java

La seguente sintassi di esempio proviene dalla pipeline WordCount nel tutorial 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 seguente sintassi di esempio proviene dalla pipeline WordCount nel tutorial su 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 seguente sintassi di esempio proviene dalla pipeline WordCount nel tutorial su 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.

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

options.setExperiments("streaming_boot_disk_size_gb=80")

Per un esempio che mostra come creare l'oggetto PipelineOptions, consulta la sezione Esempio di avvio 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 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 avvio su Dataflow di questa pagina.

Vai

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

Java

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

--experiments=streaming_boot_disk_size_gb=80

Python

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

--experiments=streaming_boot_disk_size_gb=80

Vai

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

--experiments=streaming_boot_disk_size_gb=80

Modello classico

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

modello flessibile

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

Java

Accedi a PipelineOptions all'interno di qualsiasi istanza di ParDo trasformazione DoFn 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.

Java

• project: l'ID del tuo progetto Google Cloud .
• runner: il runner della pipeline che esegue la pipeline. Per l'esecuzione di Google Cloud , questo valore deve essere DataflowRunner.

• gcpTempLocation: un percorso di Cloud Storage in cui Dataflow esegue il provisioning della maggior parte dei file temporanei. Il bucket specificato deve esistere già.

  Se non specifichi gcpTempLocation, Dataflow utilizza il valore dell'opzione tempLocation. Se non specifichi nessuna di queste opzioni, Dataflow crea un nuovo bucket Cloud Storage.

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 di Google Cloud , questo valore deve essere DataflowRunner.
• temp_location: un percorso di Cloud Storage in cui Dataflow esegue il provisioning dei file di job temporanei creati durante l'esecuzione della pipeline.

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 di Google Cloud , questo valore deve essere dataflow.
• staging_location: un percorso di Cloud Storage in cui Dataflow esegue il provisioning dei file di job temporanei creati durante l'esecuzione della pipeline.

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()

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 flag Go 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()

Java

Quando un programma Java Apache Beam esegue una pipeline su un servizio come Dataflow, in genere viene eseguito in modo asincrono. Per eseguire una pipeline e attendere il completamento del job, imposta DataflowRunner come runner 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 durante l'attesa.

Python

Quando un programma Python Apache Beam esegue una pipeline su un servizio come Dataflow, in genere viene eseguito in modo asincrono. Per bloccare 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 della riga di comando --async, che si trova nel pacchetto jobopts.

Java

Se la pipeline legge da un'origine dati senza limiti, ad esempio Pub/Sub, viene eseguita automaticamente in modalità di streaming.

Python

Se la pipeline utilizza un'origine dati senza limiti, ad esempio Pub/Sub, devi impostare l'opzione streaming su true.

Vai

Se la pipeline legge da un'origine dati senza limiti, ad esempio Pub/Sub, viene eseguita automaticamente in modalità di streaming.

Java

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

// 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')
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))

Vai

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

beam.Init()

p := beam.NewPipeline()

Java

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

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 nel seguente esempio:

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", "", "")
)

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 passarla durante la creazione dell'oggetto PipelineOptions. Quando registri la tua interfaccia con PipelineOptionsFactory, --help può trovare la tua interfaccia di opzioni personalizzate e aggiungerla all'output del comando --help. PipelineOptionsFactory verifica 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")
)