Les modèles Dataflow exploitent des paramètres d'exécution pour accepter des valeurs qui ne sont disponibles qu'au moment de l'exécution du pipeline. Pour personnaliser l'exécution d'un pipeline modélisé, vous pouvez transmettre ces paramètres aux fonctions qui s'exécutent au sein du pipeline (telles que DoFn
).
Si vous souhaitez créer un modèle à partir de votre pipeline Apache Beam, vous devez modifier le code de celui-ci pour qu'il soit compatible avec les paramètres d'exécution :
- Spécifiez
ValueProvider
pour toutes les options de pipeline que vous souhaitez définir ou utiliser au moment de l'exécution. - Appelez des méthodes d'E/S acceptant les paramètres d'exécution là où vous souhaitez paramétrer votre pipeline.
- Utilisez des objets
DoFn
acceptant les paramètres d'exécution.
Ensuite, créez et préproduisez votre modèle.
Paramètres d'exécution et interface ValueProvider
L'interface ValueProvider
permet aux pipelines d'accepter les paramètres d'exécution. Apache Beam fournit trois types d'objets ValueProvider
.
Nom | Description |
---|---|
RuntimeValueProvider |
Vous pouvez vérifier si la valeur d'un objet Utilisez |
StaticValueProvider |
Utilisez |
NestedValueProvider |
Utilisez Remarque : Le SDK Apache Beam pour Python n'est pas compatible avec |
Modifier le code pour utiliser des paramètres d'exécution
Cette section vous explique comment utiliser ValueProvider
, StaticValueProvider
et NestedValueProvider
.
Utiliser ValueProvider dans les options de pipeline
Spécifiez ValueProvider
pour toutes les options de pipeline que vous souhaitez définir ou utiliser au moment de l'exécution.
Par exemple, l'extrait de code WordCount
suivant n'est pas compatible avec les paramètres d'exécution. Le code ajoute une option de fichier d'entrée, crée un pipeline et lit les lignes du fichier d'entrée :
Java : SDK 2.x
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)
Java : SDK 1.x
Pour ajouter la compatibilité avec les paramètres d'exécution, modifiez l'option du fichier d'entrée afin d'utiliser ValueProvider
.
Java : SDK 2.x
Utilisez ValueProvider<String>
au lieu de String
pour le type de l'option de fichier d'entrée.
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
Remplacez add_argument
par 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)
Java : SDK 1.x
Utiliser ValueProvider dans les fonctions
Pour utiliser des valeurs de paramètre d'exécution dans vos propres fonctions, mettez à jour les fonctions pour qu'elles puissent exploiter les paramètres ValueProvider
.
L'exemple suivant contient une option ValueProvider
(nombre entier) et une fonction simple effectuant une addition. La fonction dépend de l'entier ValueProvider
. Lors de l'exécution, le pipeline applique MySumFn
à chaque nombre entier d'une PCollection
contenant [1, 2, 3]
. Si la valeur d'exécution est égale à 10, la PCollection
obtenue contient [11, 12, 13]
.
Java : SDK 2.x
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'))
Java : SDK 1.x
Utiliser StaticValueProvider
Pour fournir une valeur statique à votre pipeline, utilisez StaticValueProvider
.
Cet exemple exploite MySumFn
, qui est une fonction DoFn
nécessitant un objet ValueProvider<Integer>
. Si vous connaissez la valeur du paramètre à l'avance, vous pouvez spécifier votre valeur statique en tant que StaticValueProvider
à l'aide de ValueProvider
.
Java : SDK 2.x
Le code ci-dessous récupère la valeur au moment de l'exécution du pipeline :
.apply(ParDo.of(new MySumFn(options.getInt())))
Une autre option consiste à utiliser StaticValueProvider
avec une valeur statique :
.apply(ParDo.of(new MySumFn(StaticValueProvider.of(10))))
Python
Le code ci-dessous récupère la valeur au moment de l'exécution du pipeline :
beam.ParDo(MySumFn(user_options.templated_int))
Une autre option consiste à utiliser StaticValueProvider
avec une valeur statique :
beam.ParDo(MySumFn(StaticValueProvider(int,10)))
Java : SDK 1.x
Vous pouvez également utiliser StaticValueProvider
lorsque vous mettez en œuvre un module d'E/S compatible avec les paramètres standards et les paramètres d'exécution.
StaticValueProvider
réduit la duplication de code liée à la mise en œuvre de deux méthodes semblables.
Java : SDK 2.x
Le code source utilisé dans cet exemple provient du fichier d'Apache Beam TextIO.java sur GitHub.
// Create a StaticValueProviderfrom 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
Dans cet exemple, un constructeur unique accepte un argument string
ou ValueProvider
. Si l'argument est une chaîne string
, il est converti en StaticValueProvider
.
class Read(): def __init__(self, filepattern): if isinstance(filepattern, basestring): # Create a StaticValueProvider from a regular string parameter filepattern = StaticValueProvider(str, filepattern) self.filepattern = filepattern
Java : SDK 1.x
Utiliser NestedValueProvider
Remarque : Le SDK Apache Beam pour Python n'est pas compatible avec NestedValueProvider
.
Pour calculer une valeur à partir d'un autre objet ValueProvider
, utilisez NestedValueProvider
.
NestedValueProvider
exploite un objet ValueProvider
et un traducteur SerializableFunction
en tant qu'entrée. Lorsque vous appelez .get()
sur un objet NestedValueProvider
, le traducteur crée une valeur basée sur celle de ValueProvider
. Cette traduction vous permet de créer la valeur finale souhaitée à l'aide d'une valeur ValueProvider
:
- Exemple 1 : l'utilisateur fournit un nom de fichier
file.txt
. La transformation ajoute le chemings://directory_name/
devant le nom du fichier. L'appel.get()
renvoiegs://directory_name/file.txt
. - Exemple 2 : l'utilisateur fournit une sous-chaîne (telle qu'une date spécifique) pour une requête BigQuery. La transformation crée la requête complète à l'aide de la sous-chaîne. L'appel
.get()
renvoie la requête complète.
Remarque : NestedValueProvider
n'accepte qu'une seule valeur d'entrée. Vous ne pouvez pas combiner deux valeurs différentes à l'aide de NestedValueProvider
.
Le code ci-dessous met en œuvre le premier exemple à l'aide de NestedValueProvider
. L'utilisateur fournit un nom de fichier, puis la transformation ajoute le chemin devant le nom de fichier.
Java : SDK 2.x
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(); }
Python
Le SDK Apache Beam pour Python n'est pas compatible avec NestedValueProvider
.
Java : SDK 1.x
Metadata
Vous pouvez étendre vos modèles en ajoutant des métadonnées afin de valider des paramètres personnalisés lors de l'exécution du modèle. Si vous souhaitez créer des métadonnées pour votre modèle, procédez comme suit :
- Créez un fichier au format JSON appelé
<template-name>_metadata
en utilisant les paramètres décrits dans le tableau ci-dessous.Remarque : Ne nommez pas le fichier que vous créez
<template-name>_metadata.json
. Même si le fichier contient du JSON, il ne doit pas se terminer par l'extension.json
. - Stockez le fichier JSON dans le même dossier Cloud Storage que le modèle.
Remarque : Le modèle doit être stocké dans
<template-name>
et les métadonnées dans<template-name>_metadata
.
Paramètres de métadonnées
Clé du paramètre | Obligatoire | Description de la valeur | |
---|---|---|---|
name |
Oui | Nom du modèle. | |
description |
Non | Court paragraphe de texte décrivant les modèles. | |
parameters |
Non. Par défaut, un tableau vide. | Tableau de paramètres supplémentaires qui seront utilisés par le modèle. | |
name |
Oui | Nom du paramètre utilisé dans le modèle. | |
label |
Oui | Libellé lisible qui permettra de classer le paramètre dans l'interface utilisateur. | |
helpText |
Oui | Court paragraphe de texte décrivant le paramètre. | |
isOptional |
Non. La valeur par défaut est "false". | false si le paramètre est requis et true s'il est facultatif. |
|
regexes |
Non. Par défaut, un tableau vide. | Tableau d'expressions régulières POSIX-egrep sous forme de chaînes qui permettront de valider la valeur du paramètre. Par exemple : ["^[a-zA-Z][a-zA-Z0-9]+"] est une expression régulière unique qui vérifie que la valeur commence par une lettre, puis comporte un ou plusieurs caractères. |
Exemple de fichier de métadonnées
Le service Dataflow utilise les métadonnées suivantes pour valider les paramètres personnalisés du modèle WordCount :
{ "description": "An example pipeline that counts words in the input file.", "name": "Word Count", "parameters": [ { "regexes": [ "^gs:\\/\\/[^\\n\\r]+$" ], "name": "inputFile", "helpText": "Path of the file pattern glob to read from. ex: 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. ex: gs://MyBucket/counts", "label": "Output Cloud Storage file(s)" } ] }
Vous pouvez télécharger ce fichier de métadonnées à partir du répertoire de modèles de Dataflow.
E/S de pipeline et paramètres d'exécution
Java : SDK 2.x
Certains connecteurs d'E/S contiennent des méthodes qui acceptent les objets ValueProvider
. Pour déterminer la compatibilité d'un connecteur et d'une méthode spécifiques, consultez la documentation de référence de l'API pour le connecteur d'E/S. Les méthodes compatibles possèdent une surcharge avec ValueProvider
. Si une méthode ne possède pas de surcharge, elle n'est pas compatible avec les paramètres d'exécution. Les connecteurs d'E/S suivants sont au moins partiellement compatibles avec ValueProvider
:
- E/S orientées fichier :
TextIO
,AvroIO
,FileIO
,TFRecordIO
,XmlIO
BigQueryIO
*BigtableIO
(nécessite la version 2.3.0 ou ultérieure du SDK)PubSubIO
SpannerIO
* Remarque : Si vous souhaitez exécuter un pipeline de traitement par lot qui lit des données depuis BigQuery, vous devez utiliser .withTemplateCompatibility()
sur toutes les lectures BigQuery.
Python
Certains connecteurs d'E/S contiennent des méthodes qui acceptent les objets ValueProvider
. Pour déterminer la compatibilité de connecteurs d'E/S et de leurs méthodes, consultez la documentation de référence de l'API pour le connecteur. Les connecteurs d'E/S suivants acceptent les paramètres d'exécution :
- E/S orientées fichier :
textio
,avroio
,tfrecordio
Java : SDK 1.x
Créer et préproduire des modèles
Une fois votre pipeline conçu, vous devez créer et préproduire votre fichier de modèle. Utilisez la commande associée à votre version du SDK.
Remarque : Une fois le modèle créé et préproduit, l'emplacement de préproduction contient des fichiers supplémentaires nécessaires à l'exécution du modèle. Si vous supprimez cet emplacement, l'exécution du modèle échoue.
Java : SDK 2.x
Cette commande Maven crée et prépare en préproduction un modèle à l'emplacement Cloud Storage spécifié avec --templateLocation
.
Remplacez
YOUR_PROJECT_ID
par l'ID du projet.Remplacez
YOUR_BUCKET_NAME
par le nom de votre bucket Cloud Storage.Remplacez
YOUR_TEMPLATE_NAME
par le nom de votre modèle.Remplacez
com.example.myclass
par votre classe Java.Vérifiez que le chemin d'accès à
templateLocation
est correct.
mvn compile exec:java \ -Dexec.mainClass=com.example.myclass \ -Dexec.args="--runner=DataflowRunner \ --project=YOUR_PROJECT_ID \ --stagingLocation=gs://YOUR_BUCKET_NAME/staging \ --templateLocation=gs://YOUR_BUCKET_NAME/templates/YOUR_TEMPLATE_NAME"
Python
Cette commande Python crée et prépare en préproduction un modèle à l'emplacement Cloud Storage spécifié avec --template_location
. Apportez les modifications suivantes à la commande :
Remplacez
YOUR_PROJECT_ID
par l'ID du projet.Remplacez
YOUR_BUCKET_NAME
par le nom de votre bucket Cloud Storage.Remplacez
YOUR_TEMPLATE_NAME
par le nom de votre modèle.Remplacez
examples.mymodule
par votre module Python.Vérifiez que le chemin d'accès à
template_location
est correct.
python -m examples.mymodule \ --runner DataflowRunner \ --project YOUR_PROJECT_ID \ --staging_location gs://YOUR_BUCKET_NAME/staging \ --temp_location gs://YOUR_BUCKET_NAME/temp \ --template_location gs://YOUR_BUCKET_NAME/templates/YOUR_TEMPLATE_NAME
Java : SDK 1.x
Une fois votre modèle créé et préproduit, la prochaine étape consiste à l'exécuter.