Creazione di modelli Dataflow classici

In questo documento imparerai a creare un modello classico personalizzato dal codice della pipeline Dataflow. I modelli classici impacchettano le pipeline Dataflow esistenti per creare modelli riutilizzabili che puoi personalizzare per ogni job modificando parametri specifici della pipeline. Anziché scrivere il modello, utilizzerai un comando per generare il modello da una pipeline esistente.

Di seguito è riportata una breve panoramica della procedura. I dettagli di questa procedura sono forniti nelle sezioni successive.

1. Nel codice della pipeline, utilizza l'interfaccia ValueProvider per tutte le opzioni della pipeline che vuoi impostare o utilizzare in fase di esecuzione. Utilizza oggetti DoFn che accettano parametri di runtime.
2. Estendi il modello con metadati aggiuntivi in modo che i parametri personalizzati vengano convalidati quando viene eseguito il modello classico. Alcuni esempi di questi metadati includono il nome del modello classico personalizzato e i parametri facoltativi.
3. Controlla se i connettori I/O della pipeline supportano gli oggetti ValueProvider e apporta le modifiche necessarie.
4. Crea e archivia in un'area intermedia il modello classico personalizzato.
5. Esegui il modello classico personalizzato.

Per informazioni sui diversi tipi di modelli Dataflow, sui loro vantaggi e su quando scegliere un modello classico, consulta Modelli Dataflow.

Autorizzazioni necessarie per l'esecuzione di un modello classico

Le autorizzazioni necessarie per eseguire il modello classico di Dataflow dipendono da dove viene eseguito il modello e dal fatto che l'origine e il sink per la pipeline si trovino in un altro progetto.

Per ulteriori informazioni sull'esecuzione di pipeline Dataflow in locale o utilizzando Google Cloud, consulta Sicurezza e autorizzazioni di Dataflow.

Per un elenco di ruoli e autorizzazioni di Dataflow, consulta Controllo dell'accesso a Dataflow.

Limitazioni

• La seguente opzione di pipeline non è supportata con i modelli classici. Se devi controllare il numero i thread della tecnologia worker, usa Modelli Flex.

  Java

  numberOfWorkerHarnessThreads
    

  Python

  number_of_worker_harness_threads
    

• L'esecutore Dataflow non supporta le opzioni ValueProvider per argomenti e parametri della sottoscrizione Pub/Sub. Se hai bisogno di opzioni Pub/Sub nei parametri di runtime, utilizza i modelli flessibili.

Informazioni sui parametri di runtime e sull'interfaccia ValueProvider

L'interfaccia ValueProvider consente alle pipeline di accettare parametri di runtime. Apache Beam fornisce tre tipi di oggetti ValueProvider.

Nome	Descrizione
RuntimeValueProvider	RuntimeValueProvider è il tipo predefinito di ValueProvider. RuntimeValueProvider consente alla pipeline di accettare un valore disponibile solo durante l'esecuzione della pipeline. Il valore non è disponibile durante la creazione della pipeline, quindi non puoi utilizzarlo per modificare del flusso di lavoro di Google Cloud. Puoi utilizzare isAccessible() per verificare se il valore di un ValueProvider è disponibile. Se chiami get() prima dell'esecuzione della pipeline, Apache Beam restituisce un errore: Value only available at runtime, but accessed from a non-runtime context. Utilizza RuntimeValueProvider se non conosci il valore in anticipo. Per modificare i valori dei parametri in fase di runtime, non impostare i valori dei parametri nel modello. Imposta i valori dei parametri quando crei job dal modello.
StaticValueProvider	StaticValueProvider ti consente di fornire un valore statico alla pipeline. Il valore è disponibile durante la creazione della pipeline, quindi puoi utilizzare questo valore per modificare il grafico del flusso di lavoro della pipeline. Usa StaticValueProvider quando conosci il valore che ti aspetta del tempo. Vedi il valore del campo StaticValueProvider per vedere alcuni esempi.
NestedValueProvider	NestedValueProvider ti consente di calcolare un valore da un altro oggetto ValueProvider. NestedValueProvider wrap ValueProvider e il tipo di ValueProvider aggregato determina se il valore è accessibile durante la creazione della pipeline. Usa NestedValueProvider quando vuoi usare il valore per calcolare con un altro valore in fase di runtime. Consulta la sezione NestedValueProvider per alcuni esempi.

Usa i parametri di runtime nel codice della pipeline

Questa sezione illustra come utilizzare ValueProvider, StaticValueProvider e NestedValueProvider.

Nota: se non utilizzi l'interfaccia ValueProvider in il modello classico, le opzioni della pipeline che vuoi impostare o utilizzare in fase di runtime potrebbero essere ignorato.

Utilizza ValueProvider nelle opzioni della tua pipeline

Utilizza ValueProvider per tutte le opzioni della pipeline che vuoi impostare o utilizzare in fase di esecuzione.

Ad esempio, lo snippet di codice WordCount seguente non supporta i parametri di runtime. Il codice aggiunge un'opzione di file di input, crea una pipeline e legge le righe dal file di input:

Java

  public interface WordCountOptions extends PipelineOptions {
    @Description("Path of the file to read from")
    @Default.String("gs://dataflow-samples/shakespeare/kinglear.txt")
    String getInputFile();
    void setInputFile(String value);
  }

  public static void main(String[] args) {
    WordCountOptions options =
          PipelineOptionsFactory.fromArgs(args).withValidation()
            .as(WordCountOptions.class);
    Pipeline p = Pipeline.create(options);

    p.apply("ReadLines", TextIO.read().from(options.getInputFile()));
    ...

Python

  class WordcountOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
      parser.add_argument(
          '--input',
          default='gs://dataflow-samples/shakespeare/kinglear.txt',
          help='Path of the file to read from')
      parser.add_argument(
          '--output',
          required=True,
          help='Output file to write results to.')
  pipeline_options = PipelineOptions(['--output', 'some/output_path'])
  p = beam.Pipeline(options=pipeline_options)

  wordcount_options = pipeline_options.view_as(WordcountOptions)
  lines = p | 'read' >> ReadFromText(wordcount_options.input)

Per aggiungere il supporto dei parametri di runtime, modifica l'opzione del file di input da utilizzare ValueProvider.

Java

Utilizza ValueProvider<String> anziché String per il tipo di opzione del file di input.

  public interface WordCountOptions extends PipelineOptions {
    @Description("Path of the file to read from")
    @Default.String("gs://dataflow-samples/shakespeare/kinglear.txt")
    ValueProvider<String> getInputFile();
    void setInputFile(ValueProvider<String> value);
  }

  public static void main(String[] args) {
    WordCountOptions options =
          PipelineOptionsFactory.fromArgs(args).withValidation()
            .as(WordCountOptions.class);
    Pipeline p = Pipeline.create(options);

    p.apply("ReadLines", TextIO.read().from(options.getInputFile()));
    ...

Python

Sostituisci add_argument con add_value_provider_argument.

 class WordcountOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
      # Use add_value_provider_argument for arguments to be templatable
      # Use add_argument as usual for non-templatable arguments
      parser.add_value_provider_argument(
          '--input',
          default='gs://dataflow-samples/shakespeare/kinglear.txt',
          help='Path of the file to read from')
      parser.add_argument(
          '--output',
          required=True,
          help='Output file to write results to.')
  pipeline_options = PipelineOptions(['--output', 'some/output_path'])
  p = beam.Pipeline(options=pipeline_options)

  wordcount_options = pipeline_options.view_as(WordcountOptions)
  lines = p | 'read' >> ReadFromText(wordcount_options.input)

Usa ValueProvider nelle tue funzioni

Per utilizzare i valori dei parametri di runtime nelle tue funzioni, aggiorna le funzioni in modo che utilizzino i parametri ValueProvider.

L'esempio seguente contiene un'opzione ValueProvider di tipo intero e una semplice funzione che aggiunge un intero. La funzione dipende dall'intero ValueProvider. Durante l'esecuzione, la pipeline applica MySumFn a ogni numero intero in un PCollection che contiene [1, 2, 3]. Se il valore del runtime è 10, il conseguente PCollection contiene [11, 12, 13].

Java

  public interface SumIntOptions extends PipelineOptions {
      // New runtime parameter, specified by the --int
      // option at runtime.
      ValueProvider<Integer> getInt();
      void setInt(ValueProvider<Integer> value);
  }

  class MySumFn extends DoFn<Integer, Integer> {
      ValueProvider<Integer> mySumInteger;

      MySumFn(ValueProvider<Integer> sumInt) {
          // Store the value provider
          this.mySumInteger = sumInt;
      }

      @ProcessElement
      public void processElement(ProcessContext c) {
         // Get the value of the value provider and add it to
         // the element's value.
         c.output(c.element() + mySumInteger.get());
      }
  }

  public static void main(String[] args) {
    SumIntOptions options =
          PipelineOptionsFactory.fromArgs(args).withValidation()
            .as(SumIntOptions.class);

    Pipeline p = Pipeline.create(options);

    p.apply(Create.of(1, 2, 3))
      // Get the value provider and pass it to MySumFn
     .apply(ParDo.of(new MySumFn(options.getInt())))
     .apply("ToString", MapElements.into(TypeDescriptors.strings()).via(x -> x.toString()))
     .apply("OutputNums", TextIO.write().to("numvalues"));

    p.run();
  }

Python

  import apache_beam as beam
  from apache_beam.options.pipeline_options import PipelineOptions
  from apache_beam.options.value_provider import StaticValueProvider
  from apache_beam.io import WriteToText

  class UserOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
      parser.add_value_provider_argument('--templated_int', type=int)

  class MySumFn(beam.DoFn):
    def __init__(self, templated_int):
      self.templated_int = templated_int

    def process(self, an_int):
      yield self.templated_int.get() + an_int

  pipeline_options = PipelineOptions()
  p = beam.Pipeline(options=pipeline_options)

  user_options = pipeline_options.view_as(UserOptions)
  sum = (p
         | 'ReadCollection' >> beam.io.ReadFromText(
             'gs://some/integer_collection')
         | 'StringToInt' >> beam.Map(lambda w: int(w))
         | 'AddGivenInt' >> beam.ParDo(MySumFn(user_options.templated_int))
         | 'WriteResultingCollection' >> WriteToText('some/output_path'))

Utilizza StaticValueProvider

Per fornire un valore statico alla pipeline, utilizza StaticValueProvider.

Questo esempio utilizza MySumFn, che è un DoFn che richiede ValueProvider<Integer>. Se conosci del parametro in anticipo, puoi utilizzare StaticValueProvider per specificare il valore statico ValueProvider.

Java

Questo codice recupera il valore al momento dell'esecuzione della pipeline:

  .apply(ParDo.of(new MySumFn(options.getInt())))

In alternativa, puoi utilizzare StaticValueProvider con un valore statico:

  .apply(ParDo.of(new MySumFn(StaticValueProvider.of(10))))

Python

Questo codice recupera il valore al momento dell'esecuzione della pipeline:

  beam.ParDo(MySumFn(user_options.templated_int))

In alternativa, puoi utilizzare StaticValueProvider con un valore statico:

  beam.ParDo(MySumFn(StaticValueProvider(int,10)))

Puoi utilizzare StaticValueProvider anche quando implementi un modulo I/O che supporta sia i parametri regolari sia i parametri di runtime. StaticValueProvider riduce la duplicazione del codice dovuta all'implementazione di due metodi simili.

Java

Il codice sorgente di questo esempio proviene da TextIO.java su GitHub di Apache Beam.

  // Create a StaticValueProvider<String> from a regular String parameter
  // value, and then call .from() with this new StaticValueProvider.
  public Read from(String filepattern) {
    checkNotNull(filepattern, "Filepattern cannot be empty.");
    return from(StaticValueProvider.of(filepattern));
  }

  // This method takes a ValueProvider parameter.
  public Read from(ValueProvider<String> filepattern) {
    checkNotNull(filepattern, "Filepattern cannot be empty.");
    return toBuilder().setFilepattern(filepattern).build();
  }

Python

In questo esempio è presente un singolo costruttore che accetta sia un argomento string sia un argomento ValueProvider. Se è un string, viene convertito in un StaticValueProvider.

class Read():

  def __init__(self, filepattern):
    if isinstance(filepattern, str):
      # Create a StaticValueProvider from a regular string parameter
      filepattern = StaticValueProvider(str, filepattern)

    self.filepattern = filepattern

Utilizza NestedStaticValueProvider

Per calcolare un valore da un altro oggetto ValueProvider, utilizza NestedValueProvider.

NestedValueProvider prende ValueProvider e traduttore SerializableFunction come input. Quando chiami .get() su un NestedValueProvider, il traduttore crea un nuovo valore in base al valore ValueProvider. Questa traduzione ti consente di utilizzare un valore ValueProvider per creare il valore finale che preferisci.

Nota: NestedValueProvider accetta un solo valore di input. Non puoi usare un NestedValueProvider per combinare due valori diversi.

Nell'esempio seguente, l'utente fornisce il nome file file.txt. La la trasformazione antepone il percorso gs://directory_name/ a il nome del file. Ritorni a chiamare .get() gs://directory_name/file.txt.

Java

  public interface WriteIntsOptions extends PipelineOptions {
      // New runtime parameter, specified by the --fileName
      // option at runtime.
      ValueProvider<String> getFileName();
      void setFileName(ValueProvider<String> value);
  }

  public static void main(String[] args) {
     WriteIntsOptions options =
          PipelineOptionsFactory.fromArgs(args).withValidation()
            .as(WriteIntsOptions.class);
    Pipeline p = Pipeline.create(options);

    p.apply(Create.of(1, 2, 3))
     // Write to the computed complete file path.
     .apply("OutputNums", TextIO.write().to(NestedValueProvider.of(
        options.getFileName(),
        new SerializableFunction<String, String>() {
          @Override
          public String apply(String file) {
            return "gs://directoryname/" + file;
          }
        })));

    p.run();
  }

Utilizza i metadati nel codice della pipeline

Puoi estendere il modello con metadati aggiuntivi in modo che i parametri personalizzati vengano convalidati quando il modello viene eseguito. Se vuoi creare metadati per il modello, segui questi passaggi:

1. Crea un file in formato JSON denominato TEMPLATE_NAME_metadata utilizzando i parametri in Parametri dei metadati e il formato in File di metadati di esempio. Sostituisci TEMPLATE_NAME con il nome del tuo modello.

   Assicurati che il file dei metadati non abbia un'estensione del nome file. Ad esempio, se il nome del modello è myTemplate, il file dei metadati deve essere myTemplate_metadata.

2. Archivia il file di metadati in Cloud Storage nella stessa cartella del modello.

Parametri dei metadati

Chiave parametro		Obbligatorio	Descrizione del valore
name		Sì	Il nome del modello.
description		No	Un breve paragrafo di testo che descrive il modello.
streaming		No	Se true, questo modello supporta i flussi di dati. Il valore predefinito è false.
supportsAtLeastOnce		No	Se true, questo modello supporta l'elaborazione almeno una volta. Il valore predefinito è false. Imposta questo parametro su true se il modello è progettato per funzionare con la modalità flusso di dati Almeno una volta.
supportsExactlyOnce		No	Se true, questo modello supporta l'elaborazione exactly-once. Il valore predefinito è true.
defaultStreamingMode		No	La modalità di streaming predefinita per i modelli che supportano sia la modalità almeno una volta sia la modalità esattamente una volta. Utilizza uno dei seguenti valori: "AT_LEAST_ONCE", "EXACTLY_ONCE". Se non specificato, la modalità di streaming predefinita è esattamente una volta.
parameters		No	Un array di parametri aggiuntivi utilizzati dal modello. Viene utilizzato un array vuoto predefinito.
	name	Sì	Il nome del parametro utilizzato nel modello.
	label	Sì	Una stringa leggibile che viene utilizzata nella console Google Cloud per etichettare la .
	helpText	Sì	Un breve paragrafo di testo che descrive il parametro.
	isOptional	No	false se il parametro è obbligatorio e true se il parametro è facoltativo. Se non viene configurato un valore, il valore predefinito di isOptional è false. Se non includi questa chiave di parametro per i metadati, questi diventano un parametro obbligatorio.
	regexes	No	Un array di espressioni regolari POSIX-egrep in forma stringa che viene utilizzata per convalidare del parametro. Ad esempio, ["^[a-zA-Z][a-zA-Z0-9]+"] è un'unica expressione regolare che convalida che il valore inizi con una lettera e contenga uno o più caratteri. Per impostazione predefinita viene utilizzato un array vuoto.

File di metadati di esempio

Java

Il servizio Dataflow utilizza i seguenti metadati per convalidare Parametri personalizzati del modello WordCount:

{
  "description": "An example pipeline that counts words in the input file.",
  "name": "Word Count",
  "streaming": false,
  "parameters": [
    {
      "regexes": [
        "^gs:\\/\\/[^\\n\\r]+$"
      ],
      "name": "inputFile",
      "helpText": "Path of the file pattern glob to read from - for example, gs://dataflow-samples/shakespeare/kinglear.txt",
      "label": "Input Cloud Storage file(s)"
    },
    {
      "regexes": [
        "^gs:\\/\\/[^\\n\\r]+$"
      ],
      "name": "output",
      "helpText": "Path and filename prefix for writing output files - for example, gs://MyBucket/counts",
      "label": "Output Cloud Storage file(s)"
    }
  ]
}

Python

Il servizio Dataflow utilizza i seguenti metadati per convalidare i parametri personalizzati del modello WordCount:

{
  "description": "An example pipeline that counts words in the input file.",
  "name": "Word Count",
  "streaming": false,
  "parameters": [
    {
      "regexes": [
        "^gs:\\/\\/[^\\n\\r]+$"
      ],
      "name": "input",
      "helpText": "Path of the file pattern glob to read from - for example, gs://dataflow-samples/shakespeare/kinglear.txt",
      "label": "Input Cloud Storage file(s)"
    },
    {
      "regexes": [
        "^gs:\\/\\/[^\\n\\r]+$"
      ],
      "name": "output",
      "helpText": "Path and filename prefix for writing output files - for example, gs://MyBucket/counts",
      "label": "Output Cloud Storage file(s)"
    }
  ]
}

Puoi scaricare i file di metadati per i modelli forniti da Google dalla directory dei modelli di Dataflow.

Connettori I/O della pipeline e ValueProvider supportati

Java

Alcuni connettori I/O contengono metodi che accettano oggetti ValueProvider. Per determinare il supporto di una specifica connettore e metodo, consulta la documentazione di riferimento API per il connettore di I/O. I metodi supportati hanno un sovraccarico con un ValueProvider. Se un metodo non ha un sovraccarico, e supportare parametri di runtime. I seguenti connettori I/O hanno un supporto ValueProvider almeno parziale:

• IO basati su file: TextIO, AvroIO, FileIO, TFRecordIO e XmlIO
• BigQueryIO*
• BigtableIO (richiede SDK 2.3.0 o versioni successive)
• PubSubIO
• SpannerIO

* Nota: se vuoi eseguire una pipeline batch che legge da BigQuery, devi utilizzare .withTemplateCompatibility() su tutte le letture di BigQuery.

Python

Alcuni connettori I/O contengono metodi che accettano oggetti ValueProvider. Per determinare il supporto dei connettori I/O e dei relativi metodi, consulta la documentazione di riferimento dell'API per il connettore. I seguenti connettori I/O accettano i parametri di runtime:

• IO basati su file: textio, avroio, tfrecordio

Crea e organizza un modello classico

Dopo aver scritto la pipeline, devi creare e organizzare il file modello. Quando crei e gestisci temporaneamente un modello, la posizione di gestione temporanea contiene file aggiuntivi necessari per l'esecuzione del modello. Se elimini la posizione temporanea, il modello non verrà eseguito. Il job Dataflow non viene eseguito immediatamente dopo l'archiviazione del modello. Per eseguire un un job Dataflow basato su modelli personalizzati, puoi utilizzare Console Google Cloud, l'API REST Dataflow, o gcloud CLI.

Nota: la creazione e l'implementazione di un modello richiedono l'autenticazione. Per istruzioni su come creare un account di servizio e una chiave dell'account di servizio, consulta le Guida rapida per la lingua che stai utilizzando: Guida rapida di Java, guida rapida di Python, oppure Guida rapida di Go.

L'esempio seguente mostra come eseguire il commit di un file modello:

Java

Questo comando Maven crea e gestisce un modello nella posizione Cloud Storage specificata con --templateLocation.

Nota: se utilizzi l'SDK Apache Beam per Java 2.15.0 o versioni successive, devi specificare --region.

    mvn compile exec:java \
     -Dexec.mainClass=com.example.myclass \
     -Dexec.args="--runner=DataflowRunner \
                  --project=PROJECT_ID \
                  --stagingLocation=gs://BUCKET_NAME/staging \
                  --templateLocation=gs://BUCKET_NAME/templates/TEMPLATE_NAME \
                  --region=REGION" \
     -P dataflow-runner
    

Verifica che il percorso templateLocation sia corretto. Sostituisci quanto segue:

• com.example.myclass: la tua classe Java
• PROJECT_ID: il tuo ID progetto
• BUCKET_NAME: il nome del bucket Cloud Storage
• TEMPLATE_NAME: il nome del modello
• REGION: la regione in cui eseguire il deployment del job Dataflow

Python

Questo comando Python crea e crea un modello in un ambiente Cloud Storage località specificata con --template_location.

Nota: se utilizzi l'SDK Apache Beam per Python 2.15.0 o versioni successive, devi specificare --region.

  python -m examples.mymodule \
    --runner DataflowRunner \
    --project PROJECT_ID \
    --staging_location gs://BUCKET_NAME/staging \
    --template_location gs://BUCKET_NAME/templates/TEMPLATE_NAME \
    --region REGION

Verifica che il percorso template_location sia corretto. Sostituisci quanto segue:

• examples.mymodule: il tuo modulo Python
• PROJECT_ID: il tuo ID progetto
• BUCKET_NAME: il nome del bucket Cloud Storage
• TEMPLATE_NAME: il nome del modello
• REGION: il region per eseguire il deployment Job Dataflow in

Dopo aver creato e archiviato il modello, il passaggio successivo consiste nell'eseguire il modello.