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
ValueProviderpour toutes les options de pipeline des modèles classiques 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
DoFnacceptant 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 |
L'exécuteur Dataflow n'est pas compatible avec les options ValueProvider pour les sujets Pub/Sub et les paramètres d'abonnement. Si vous avez besoin des options Pub/Sub dans vos paramètres d'exécution, utilisez plutôt les modèles Flex.
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
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.
Dans l'exemple suivant, l'utilisateur fournit le nom de fichier file.txt. La transformation ajoute le chemin gs://directory_name/ devant le nom du fichier. L'appel .get() renvoie gs://directory_name/file.txt.
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();
}
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>_metadataen utilisant les paramètres décrits dans le tableau ci-dessous. - Stockez le fichier au format JSON dans Cloud Storage, dans le même dossier que le modèle.
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. Si vous n'incluez pas cette clé de paramètre pour vos métadonnées, les métadonnées deviennent un paramètre obligatoire. |
|
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
Java
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)"
}
]
}
Python
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": "input",
"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 des fichiers de métadonnées pour les modèles fournis par Google à 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)PubSubIOSpannerIO
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.
Consultez les exemples suivants pour savoir comment préproduire un fichier de modèle :
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.
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"
Vérifiez que le chemin d'accès à templateLocation est correct. Remplacez l'élément suivant :
com.example.myclass: votre classe JavaPROJECT_ID: ID de votre projet.BUCKET_NAME: nom du bucket Cloud StorageTEMPLATE_NAME: nom du modèle.REGION: point de terminaison régional où déployer votre tâche Dataflow.
Python
Cette commande Python crée et prépare en préproduction un modèle à l'emplacement Cloud Storage spécifié avec --template_location.
python -m examples.mymodule \
--runner DataflowRunner \
--project PROJECT_ID \
--staging_location gs://BUCKET_NAME/staging \
--temp_location gs://BUCKET_NAME/temp \
--template_location gs://BUCKET_NAME/templates/TEMPLATE_NAME \
--region REGION
Vérifiez que le chemin d'accès à template_location est correct. Remplacez l'élément suivant :
examples.mymodule: votre module Python.PROJECT_ID: ID de votre projet.BUCKET_NAME: nom du bucket Cloud StorageTEMPLATE_NAME: nom du modèle.REGION: point de terminaison régional où déployer votre tâche Dataflow.
Java : SDK 1.x
Une fois votre modèle créé et préproduit, la prochaine étape consiste à l'exécuter.