Las transformaciones en el SDK de Dataflow pueden tener una estructura anidada, en la que puedes redactar una transformación compleja a partir de varias transformaciones más simples. Esta transformación puede estar compuesta por otras operaciones de transformación (es decir, puede que realice más de una ParDo, Combine o GroupByKey). Estas transformaciones se llaman transformaciones compuestas. Las transformaciones compuestas son útiles si quieres crear una transformación que se pueda volver a utilizar y que se componga de varios pasos.
Anidar varias transformaciones dentro de una única transformación compuesta le otorga varios beneficios a tu canalización de Dataflow:
- Las transformaciones compuestas pueden hacer que tu código tenga más módulos y sea más fácil de comprender, lo cual fomenta su reutilización.
- La interfaz de Dataflow Monitoring puede hacer referencia a las transformaciones compuestas por su nombre, lo que te facilita el seguimiento y la comprensión del progreso de tu canalización en el entorno de ejecución.
Ejemplo de una transformación compuesta
Muchas de las transformaciones predefinidas en el SDK de Dataflow son transformaciones compuestas.
La transformación CountWords en el programa de ejemplo WordCount del SDK de Dataflow es un ejemplo de una transformación compuesta. CountWords es una subclase de PTransform compuesta por varias transformaciones anidadas.
En el método apply, la transformación CountWords aplica las siguientes operaciones de transformación:
- Aplica una
ParDoen la entradaPCollectionde líneas de texto, lo que produce una salidaPCollectionde palabras individuales. - Aplica la transformación
Count* en la biblioteca del SDK de Dataflow en laPCollectionde palabras, lo que produce unaPCollectionde pares clave-valor. Cada clave representa una palabra del texto, y cada valor representa la cantidad de veces que aparece una palabra en los datos originales. - Aplica una
ParDofinal a laPCollectionde los pares clave-valor y genera unaPCollectionde strings imprimibles apropiadas para escribir en un archivo de salida.
En la figura 1 se muestra un diagrama sobre cómo se estructura la canalización que contiene CountWords mediante transformaciones compuestas.
Java
Los parámetros y el valor de retorno de tu transformación compuesta deben coincidir con el tipo de ingreso inicial y el tipo de resultado final mostrados de la transformación completa. Por ejemplo, CountWords.apply acepta un ingreso PCollection<String> y muestra una PCollection<String>, aunque los datos intermedios de la transformación cambien de tipo varias veces:
static class CountWords
extends PTransform<PCollection<String>, PCollection<String>> {
@Override
public PCollection<String> apply(PCollection<String> lines) {
PCollection<String> words = lines.apply(
ParDo
.named("ExtractWords")
.of(new ExtractWordsFn()));
PCollection<KV<String, Integer>> wordCounts =
words.apply(Count.<String>perElement());
PCollection<String> results = wordCounts.apply(
ParDo
.named("FormatCounts")
.of(new DoFn<KV<String, Integer>, String>() {
@Override
public void processElement(ProcessContext c) {
c.output(c.element().getKey() + ": " + c.element().getValue());
}
}));
return results;
}
}
Cómo crear una transformación compuesta
Puedes crear tu propia transformación compuesta si creas una subclase de la clase Ptransform en el SDK de Dataflow y anulas el método de aplicación para especificar la lógica de procesamiento real.
Luego, puedes utilizar esta transformación de la misma forma en que lo harías con una transformación integrada del SDK.
Java
En los parámetros del tipo de clase PTransform, debes pasar los tipos PCollection que tu transformación toma como entrada y produce como salida.
Para tomar varias PCollections como entrada, o producir varias PCollections como salida, utiliza uno de los tipos de varias colecciones en el parámetro de tipo relevante.
En el siguiente código de ejemplo, se muestra cómo declarar una PTransform que acepte una PCollection de String como entrada y genere como salida una PCollection de Integer:
static class ComputeWordLengths
extends PTransform<PCollection<String>, PCollection<Integer>> {
...
}
Cómo anular el método apply
Dentro de tu subclase PTransform, debes anular el método apply. En apply es donde agregas la lógica de procesamiento de PTransform. La anulación de apply debe aceptar el tipo apropiado de PCollection de entrada como un parámetro, y debe especificar la PCollection de salida como el valor de retorno.
Java
En el siguiente código de ejemplo, se muestra cómo anular apply para la clase ComputeWordLengths que se declaró en el ejemplo anterior:
static class ComputeWordLengths
extends PTransform<PCollection<String>, PCollection<Integer>> {
@Override
public PCollection<Integer> apply(PCollection<String>) {
...
// transform logic goes here
...
}
Siempre y cuando anules el método apply en tu subclase PTransform para que acepte las PCollection de entrada apropiadas y muestre las PCollection de salida correspondientes, puedes incluir tantas transformaciones como desees. Estas transformaciones pueden ser transformaciones básicas, transformaciones compuestas, o alguna otra transformación incluida en las bibliotecas de los SDK de Dataflow.
Java
El usuario de la transformación no debe invocar el método apply de una PTransform de forma directa. En cambio, debes llamar al método apply en la PCollection misma con la transformación como argumento.
Esto permite que las transformaciones se aniden dentro de la estructura de tu canalización.
Si necesitas ayuda, visita nuestra