Crea transformaciones compuestas

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 podría estar compuesta por otras operaciones de transformación (es decir, puede que realice más de una ParDo, Combine o GroupByKey). Estas transformaciones se denominan transformaciones compuestas. Las transformaciones compuestas son útiles si quieres crear una transformación que se pueda volver a usar 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 el ejemplo de una transformación compuesta. CountWords es una subclase PTransform compuesta por varias transformaciones anidadas.

En su método apply, la transformación CountWords aplica las siguientes operaciones de transformación:

  1. Aplica un ParDo en la entrada PCollection de las líneas de texto, que produce una salida PCollection de palabras individuales.
  2. Aplica la transformación Count* en la biblioteca del SDK de Dataflow en la PCollection de palabras, lo que produce una PCollection de 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.
  3. Aplica una ParDo final a la PCollection de los pares clave-valor y genera una PCollection de 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.

La transformación CountWords es una transformación compuesta que usa dos operaciones ParDo y la transformación provista por el SDK llamada Count.
Figura 1: Desglose de la transformación compuesta CountWords

Java

Los parámetros y el valor de retorno de tu transformación compuesta deben coincidir con el tipo de entrada inicial y el tipo de resultado final mostrados de toda la transformación. Por ejemplo, CountWords.apply acepta una entrada 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;
    }
  }

Crea 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 usar esta transformación de la misma forma en la que lo harías con una transformación integrada del SDK.

Java

Para los parámetros de 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, usa uno de los tipos de varias colecciones en el parámetro de tipo relevante.

En la siguiente muestra de código, se señala 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>> {
    ...
  }

Anula el método apply

Dentro de tu subclase PTransform, deberás anular el método apply. apply es donde agregas la lógica de procesamiento de PTransform. La anulación de apply debe aceptar el tipo apropiado PCollection de entrada como un parámetro, y debe especificar la PCollection de salida como el valor de retorno.

Java

En la siguiente muestra de código, se señala cómo anular apply de la clase ComputeWordLengths declarada 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.