In questo documento scoprirai come creare un modello classico personalizzato dal codice della pipeline Dataflow. I modelli classici raggruppano le pipeline Dataflow esistenti per creare modelli riutilizzabili che puoi personalizzare per ogni job modificando parametri specifici della pipeline. Anziché scrivere il modello, utilizzi un comando per generarlo da una pipeline esistente.
Di seguito è riportata una breve panoramica della procedura. I dettagli di questa procedura sono forniti nelle sezioni successive.
- Nel codice della pipeline, utilizza l'interfaccia
ValueProvider
per tutte le opzioni della pipeline che vuoi impostare o utilizzare in fase di esecuzione. Utilizza gli oggettiDoFn
che accettano parametri runtime. - Estendi il modello con metadati aggiuntivi in modo che i parametri personalizzati vengano convalidati quando viene eseguito il modello classico. Esempi di questi metadati includono il nome del modello classico personalizzato e i parametri facoltativi.
- Controlla se i connettori I/O della pipeline supportano gli oggetti
ValueProvider
e apporta le modifiche necessarie. - Crea e prepara il modello classico personalizzato.
- Esegui il modello classico personalizzato.
Per scoprire di più sui diversi tipi di modelli Dataflow, sui loro vantaggi e su quando scegliere un modello classico, consulta Modelli Dataflow.
Autorizzazioni richieste per l'esecuzione di un modello classico
Le autorizzazioni necessarie per eseguire il modello Dataflow classico dipendono da dove esegui il modello e se l'origine e il sink della pipeline si trovano in un altro progetto.
Per ulteriori informazioni sull'esecuzione delle pipeline Dataflow in locale o utilizzando Google Cloud, consulta Sicurezza e autorizzazioni di Dataflow.
Per un elenco di ruoli e autorizzazioni Dataflow, consulta Controllo dell'accesso a Dataflow.
Limitazioni
- La seguente
opzione della pipeline
non è supportata con i modelli classici. Se devi controllare il numero di
thread del worker harness, utilizza
modelli flessibili.
Java
numberOfWorkerHarnessThreads
Python
number_of_worker_harness_threads
- Il runner Dataflow non supporta le opzioni
ValueProvider
per i parametri di argomenti e sottoscrizioni Pub/Sub. Se hai bisogno di opzioni Pub/Sub nei parametri runtime, utilizza i modelli flessibili.
Informazioni sui parametri 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 |
Puoi utilizzare Utilizza |
StaticValueProvider |
Utilizza |
NestedValueProvider |
Utilizza |
Utilizzare i parametri runtime nel codice della pipeline
Questa sezione illustra come utilizzare ValueProvider
,
StaticValueProvider
e NestedValueProvider
.
Utilizzare ValueProvider
nelle opzioni della pipeline
Utilizza ValueProvider
per tutte le opzioni della pipeline che vuoi impostare
o utilizzare in fase di esecuzione.
Ad esempio, il seguente snippet di codice WordCount
non supporta
i parametri di runtime. Il codice aggiunge un'opzione per il 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 in modo che utilizzi
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)
Utilizzare ValueProvider
nelle funzioni
Per utilizzare i valori dei parametri runtime nelle tue funzioni, aggiorna le funzioni
in modo che utilizzino i parametri ValueProvider
.
L'esempio seguente contiene un'opzione ValueProvider
di tipo integer
e una semplice funzione che aggiunge un numero 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 di runtime è 10, il valore
PCollection
risultante 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 accetta un ValueProvider<Integer>
. Se conosci il valore del parametro in anticipo, puoi utilizzare StaticValueProvider
per specificare il valore statico come ValueProvider
.
Java
Questo codice ottiene il valore in fase di runtime 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 ottiene il valore in fase di runtime 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 anche utilizzare StaticValueProvider
quando implementi un modulo I/O che supporta sia i parametri regolari sia i parametri di runtime.
StaticValueProvider
riduce la duplicazione del codice dall'implementazione di due metodi simili.
Java
Il codice sorgente per 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 l'argomento è 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
accetta come input un ValueProvider
e un
SerializableFunction
. 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 desideri.
Nell'esempio seguente, l'utente fornisce il nome file file.txt
. La
trasformazione antepone il percorso gs://directory_name/
al
nome del file. Chiamata a .get()
restituisce
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(); }
Utilizzare i metadati nel codice della pipeline
Puoi estendere il modello con metadati aggiuntivi in modo che i parametri personalizzati vengano convalidati quando viene eseguito il modello. Se vuoi creare metadati per il tuo modello, segui questi passaggi:
- 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. SostituisciTEMPLATE_NAME
con il nome del modello.Assicurati che il file dei metadati non abbia un'estensione del nome file. Ad esempio, se il nome del modello è
myTemplate
, il file di metadati deve esseremyTemplate_metadata
. - 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 lo streaming. 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à flusso di dati predefinita per i modelli che supportano sia la modalità "at least once" che la modalità "exactly once". Utilizza uno dei seguenti valori: "AT_LEAST_ONCE" ,
"EXACTLY_ONCE" . Se non specificata, la modalità di streaming predefinita è esattamente una volta.
|
|
parameters |
No | Un array di parametri aggiuntivi utilizzati dal modello. Per impostazione predefinita viene utilizzato un array vuoto. | |
name |
Sì | Il nome del parametro utilizzato nel modello. | |
label |
Sì | Una stringa leggibile utilizzata nella console Google Cloud per etichettare il parametro. | |
helpText |
Sì | Un breve paragrafo di testo che descrive il parametro. | |
isOptional |
No | false se il parametro è obbligatorio e true se è
facoltativo. Se non viene impostato un valore, isOptional ha come valore predefinito 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 formato stringa utilizzato per convalidare il
valore del parametro. Ad esempio, ["^[a-zA-Z][a-zA-Z0-9]+"] è una singola
espressione regolare che verifica 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 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": "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 Dataflow.
Connettori I/O della pipeline supportati e ValueProvider
Java
Alcuni connettori I/O contengono metodi che accettano
oggetti ValueProvider
. Per determinare il supporto per un connettore e un metodo specifici, consulta la documentazione di riferimento dell'API per il connettore I/O. I metodi supportati hanno un sovraccarico con un
ValueProvider
. Se un metodo non ha un overload, non supporta i parametri di runtime. I seguenti connettori I/O supportano
almeno parzialmente ValueProvider
:
- IO basati su file:
TextIO
,AvroIO
,FileIO
,TFRecordIO
,XmlIO
BigQueryIO
*BigtableIO
(richiede l'SDK 2.3.0 o versioni successive)PubSubIO
SpannerIO
Python
Alcuni connettori I/O contengono metodi che accettano
oggetti ValueProvider
. Per determinare il supporto per i connettori I/O
e i relativi metodi, consulta la documentazione di riferimento dell'API
per il connettore. I seguenti connettori I/O accettano parametri di runtime:
- IO basati su file:
textio
,avroio
,tfrecordio
Creare e preparare un modello classico
Dopo aver scritto la pipeline, devi creare e preparare il file modello. Quando crei e gestisci temporaneamente un modello, la posizione di gestione temporanea contiene file aggiuntivi necessari per eseguire il modello. Se elimini la posizione di staging, il modello non viene eseguito. Il job Dataflow non viene eseguito immediatamente dopo l'archiviazione temporanea del modello. Per eseguire un job Dataflow personalizzato basato su modelli, puoi utilizzare la consoleGoogle Cloud , l'API REST di Dataflow o gcloud CLI.
Il seguente esempio mostra come preparare un file modello:
Java
Questo comando Maven crea e prepara un modello nella posizione Cloud Storage
specificata con --templateLocation
.
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 di templateLocation
sia corretto. Sostituisci quanto segue:
com.example.myclass
: la tua classe JavaPROJECT_ID
: il tuo ID progettoBUCKET_NAME
: il nome del bucket Cloud StorageTEMPLATE_NAME
: il nome del modelloREGION
: la regione in cui eseguire il deployment del job Dataflow
Python
Questo comando Python crea e prepara un modello nella posizione Cloud Storage specificata con --template_location
.
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 di template_location
sia corretto. Sostituisci quanto segue:
examples.mymodule
: il tuo modulo PythonPROJECT_ID
: il tuo ID progettoBUCKET_NAME
: il nome del bucket Cloud StorageTEMPLATE_NAME
: il nome del modelloREGION
: la regione in cui eseguire il deployment del job Dataflow
Dopo aver creato e archiviato il modello, il passaggio successivo consiste nell'eseguirlo.