In questo documento imparerai a 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, 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.
- Nel codice della pipeline, utilizza l'interfaccia
ValueProvider
per tutte le opzioni della pipeline che vuoi impostare o utilizzare in fase di runtime. Utilizza gli oggettiDoFn
che accettano i parametri di runtime. - 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.
- Controlla se i connettori I/O della pipeline supportano gli oggetti
ValueProvider
e apporta le modifiche necessarie. - Crea e archivia in un'area intermedia il modello classico personalizzato.
- 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 richieste 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
- Le seguenti
opzione pipeline
non è supportato 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 delle opzioni Pub/Sub nel tuo i 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 |
Puoi utilizzare Utilizza |
StaticValueProvider |
Usa |
NestedValueProvider |
Usa |
Usa i parametri di runtime nel codice della pipeline
Questa sezione illustra come utilizzare ValueProvider
,
StaticValueProvider
e NestedValueProvider
.
Utilizza ValueProvider
nelle opzioni della tua pipeline
Utilizza ValueProvider
per tutte le opzioni della pipeline da impostare
o da utilizzare in fase di runtime.
Ad esempio, il seguente snippet di codice WordCount
non supporta
parametri di runtime. Il codice aggiunge un'opzione per il file di input, crea una pipeline e
legge le righe del 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 da utilizzare
ValueProvider
.
Java
Usa 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
per utilizzare i parametri ValueProvider
.
L'esempio seguente contiene un numero intero ValueProvider
,
e una funzione semplice che aggiunge un numero intero. La funzione dipende
Numero 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 ottiene il valore al momento del 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 al momento del 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 utilizzare StaticValueProvider
anche quando implementi una
Modulo I/O che supporta sia i parametri normali sia i parametri di runtime.
StaticValueProvider
riduce la duplicazione del codice implementando due metodi simili.
Java
Il codice sorgente per questo esempio proviene dal linguaggio di programmazione TextIO.java su GitHub.
// 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, c'è un singolo costruttore che accetta sia un
string
o 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 NestedValueProvider
, il traduttore
crea un nuovo valore basato sul valore ValueProvider
. Questo
consente di utilizzare un valore ValueProvider
per creare
il valore finale desiderato.
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 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 in Parametri metadati e il formato in File di metadati di esempio: SostituisciTEMPLATE_NAME
con il nome del tuo modello.Verifica che il file di metadati non abbia un'estensione di nome file. Ad esempio, se il tuo 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 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 lavorare con lo streaming "at-least-once"
predefinita.
|
|
supportsExactlyOnce |
No | Se true , questo modello supporta
elaborazione "exactly-once". Il valore predefinito
è true . |
|
defaultStreamingMode |
No | La modalità flusso di dati predefinita, per i modelli che supportano sia la modalità "Almeno una volta" sia
modalità "exactly-once". Utilizza uno dei seguenti valori: "AT_LEAST_ONCE" ,
"EXACTLY_ONCE" . Se non specificata, la modalità flusso di dati predefinita viene impostata "exactly-once".
|
|
parameters |
No | Un array di parametri aggiuntivi utilizzati dal modello. Un array vuoto viene utilizzato predefinito. | |
name |
Sì | Il nome del parametro utilizzato nel modello. | |
label |
Sì | Una stringa leggibile 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 impostato un valore, il valore predefinito di isOptional è false .
Se non includi questa chiave parametro per i metadati, questi diventano obbligatori
. |
|
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 singolo
espressione regolare che convalida che il valore inizi con una lettera e poi abbia una 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 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 dal Dataflow directory dei modelli.
Connettori I/O della pipeline supportati e ValueProvider
Java
Alcuni connettori I/O contengono metodi che accettano
ValueProvider
oggetti. 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
assistenza ValueProvider
almeno parziale:
- IO basati su file:
TextIO
,AvroIO
,FileIO
,TFRecordIO
eXmlIO
BigQueryIO
*BigtableIO
(richiede SDK 2.3.0 o versioni successive)PubSubIO
SpannerIO
Python
Alcuni connettori I/O contengono metodi che accettano
ValueProvider
oggetti. Per determinare il supporto dei connettori I/O
e i relativi metodi, consulta la documentazione di riferimento delle 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 staging di un modello, la posizione temporanea contiene file aggiuntivi che sono necessari per eseguire modello. Se elimini la posizione temporanea, il modello non verrà eseguito. Il job Dataflow non viene eseguito immediatamente dopo l'archiviazione in un'area intermedia del modello. Per eseguire un un job Dataflow basato su modelli personalizzati, puoi utilizzare Console Google Cloud, l'API REST Dataflow, o gcloud CLI.
L'esempio seguente mostra come posizionare in un'area intermedia un file modello:
Java
Questo comando Maven crea e memorizza un modello nella posizione di Cloud Storage
specificato 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 templateLocation
sia corretto. Sostituisci quanto segue:
com.example.myclass
: la tua classe JavaPROJECT_ID
: il tuo ID progettoBUCKET_NAME
: il nome di Cloud Storage del bucket rimanenteTEMPLATE_NAME
: il nome del modelloREGION
: il valore region per eseguire il deployment Job Dataflow in
Python
Questo comando Python crea e memorizza un modello in un ambiente Cloud Storage
località 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 template_location
sia corretto. Sostituisci quanto segue:
examples.mymodule
: il tuo modulo PythonPROJECT_ID
: il tuo ID progettoBUCKET_NAME
: il nome di Cloud Storage del bucket rimanenteTEMPLATE_NAME
: il nome del modelloREGION
: il valore region per eseguire il deployment Job Dataflow in
Dopo aver creato e messo in scena il modello, il passaggio successivo consiste nel esegui il modello.