Codificación de datos

Cuando crees o generes datos de canalización, deberás especificar cómo se codifican y decodifican los elementos de tus PCollections desde y hacia las strings de bytes. Las strings de bytes se usan para el almacenamiento intermedio, la lectura desde fuentes y la escritura en receptores. Los SDK de Dataflow usan objetos denominados codificadores para describir cómo se deben codificar y decodificar los elementos de una PCollection dada.

Usa codificadores

Por lo general, es necesario que especifiques un codificador cuando leas datos en tu canalización desde una fuente externa (o crees datos de canalización a partir de datos locales) y, también, cuando envíes datos de canalización a un receptor externo. Consulta E/S de canalización para obtener más información.

Java

En el SDK de Dataflow para Java, el tipo Coder proporciona los métodos necesarios a fin de codificar y decodificar datos. El SDK de Cloud Dataflow para Java proporciona una serie de subclases Coder que funcionan con una variedad de tipos de Java estándar, como Integer, Long, Double, StringUtf8, TableRow de BigQuery y muchos más. Puedes encontrar todas las subclases Coder disponibles en el paquete com.google.cloud.dataflow.sdk.coders.

Cuando lees datos en una canalización, el codificador indica cómo interpretar los datos de entrada en un tipo de lenguaje específico, como Integer o String. Del mismo modo, el codificador indica cómo los tipos de lenguaje específicos en tu canalización deben escribirse en strings de bytes para un receptor de datos de salida o a fin de materializar datos intermedios en tu canalización.

Los SDK de Dataflow establecen un codificador para cada PCollection en una canalización, incluidos los generados como resultado de una transformación. La mayoría de las veces, los SDK de Dataflow pueden inferir el codificador correcto para una PCollection de salida de forma automática. Consulta Infiere un codificador para obtener más información.

Ten en cuenta que los codificadores no siempre tienen una relación 1 a 1 con los tipos. Por ejemplo, el tipo Integer puede tener varios codificadores válidos y los datos de entrada y salida pueden usar diferentes codificadores Integer. Una transformación puede tener datos de entrada de tipo Integer que usen BigEndianIntegerCoder y datos de salida de tipo Integer que usen VarIntCoder.

Java

Puedes establecer un Coder de forma explícita cuando ingresas o generas una PCollection. Cuando aplicas las transformaciones Read o Write de tu canalización, estableces el Coder con una llamada al método .withCoder.

Por lo general, estableces el Coder cuando el codificador para una PCollection no se puede inferir de forma automática o cuando quieres usar un codificador diferente al predeterminado de tu canalización. En el siguiente código de ejemplo, se lee un conjunto de números de un archivo de texto y se establece un Coder de tipo TextualIntegerCoder para la PCollection resultante:

  PCollection<Integer> numbers =
     p.begin()
      .apply(TextIO.Read.named("ReadNumbers")
                        .from("gs://my_bucket/path/to/numbers-*.txt")
                        .withCoder(TextualIntegerCoder.of()));

Puedes establecer el codificador para una PCollection existente con el método PCollection.setCoder. Ten en cuenta que no puedes llamar a setCoder en una PCollection que ya finalizó (p. ej., si llamas a .apply en él).

Puedes obtener el codificador para una PCollection existente con el método getCoder. Este método fallará con IllegalStateException si no se estableció un codificador y no se puede inferir para la PCollection dada.

Inferencia de codificador y codificadores predeterminados

Los SDK de Dataflow requieren un codificador por cada PCollection en tu canalización. Sin embargo, la mayoría de las veces, no es necesario que especifiques un codificador de forma explícita, como para una PCollection intermedia producto de una transformación en el medio de tu canalización. En esos casos, los SDK de Dataflow pueden inferir un codificador apropiado de las entradas y salidas de la transformación que se usa para producir la PCollection.

Java

Cada objeto Pipeline tiene un CoderRegistry. El CoderRegistry representa una asignación de los tipos de Java a los codificadores predeterminados que la canalización debe usar para las PCollections de cada tipo.

De forma predeterminada, el SDK de Dataflow para Java infiere el Coder de los elementos de una PCollection de salida de manera automática con el parámetro de tipo del objeto de función de la transformación, como DoFn. En el caso de ParDo, por ejemplo, un objeto de función DoFn<Integer, String> acepta un elemento de entrada de tipo Integer y produce un elemento de salida de tipo String. En ese caso, el SDK de Dataflow para Java inferirá, de forma automática, el Coder predeterminado de la PCollection<String> de salida (en la canalización predeterminada del CoderRegistry, que sería StringUtf8Coder).

Los codificadores predeterminados y CoderRegistry

Cada objeto de Pipeline posee un objeto de CoderRegistry, que asigna tipos de lenguajes al codificador predeterminado que la canalización debe usar para esos tipos. Puedes usar CoderRegistry tú mismo a fin de buscar el codificador predeterminado de un tipo específico o registrar un nuevo codificador predeterminado para un tipo determinado.

Java

CoderRegistry contiene una asignación predeterminada de los Coder a los tipos estándar de Java con respecto a todas las Pipeline que crees con el SDK de Dataflow para Java. En la siguiente tabla, se muestra la asignación estándar:

Tipo de Java Coder predeterminado
Double DoubleCoder
Instant InstantCoder
Integer VarIntCoder
Iterable IterableCoder
KV KvCoder
List ListCoder
Map MapCoder
Long VarLongCoder
String StringUtf8Coder
TableRow TableRowJsonCoder
Void VoidCoder
byte[] ByteArrayCoder
TimestampedValue TimestampedValueCoder

Puedes usar el método CoderRegistry.registerStandardCoders a fin de establecer asignaciones predeterminadas para cualquier CoderRegistry dado.

Busca un codificador predeterminado

Java

Puedes usar el método CoderRegistry.getDefaultCoder a fin de determinar el Coder predeterminado para un tipo de Java. Puedes acceder al CoderRegistry para una Pipeline determinada con el método Pipeline.getCoderRegistry. Esto te permite determinar (o establecer) el Coder predeterminado para un tipo de Java por cada canalización: es decir, “para esta canalización, verifica que los valores Integer se codifiquen con BigEndianIntegerCoder”.

Establece el codificador predeterminado para un tipo

Java

A fin de establecer el Coder predeterminado para un tipo de Java de una canalización en particular, debes obtener y modificar el CoderRegistry de la canalización. Usa el método Pipeline.getCoderRegistry para obtener el objeto CoderRegistry y, a continuación, usa el método CoderRegistry.registerCoder a fin de registrar un Coder nuevo para el tipo de Java de destino.

En el siguiente código de ejemplo, se demuestra cómo establecer un Coder predeterminado, en este caso BigEndianIntegerCoder, para los valores Integer de una canalización.

  PipelineOptions options = PipelineOptionsFactory.create();
  Pipeline p = Pipeline.create(options);

  CoderRegistry cr = p.getCoderRegistry();
  cr.registerCoder(Integer.class, BigEndianIntegerCoder.class);

Anota un tipo de datos personalizado con un codificador predeterminado

Java

Si tu programa de canalización define un tipo de datos personalizado, puedes usar la anotación @DefaultCoder para especificar el codificador que se debe usar con ese tipo. Por ejemplo, supongamos que tienes un tipo de datos personalizado para el que deseas usar SerializableCoder. Puedes usar la anotación @DefaultCoder de la siguiente manera:

  @DefaultCoder(AvroCoder.class)
  public class MyCustomDataType {
    ...
  }

Si creaste un codificador personalizado para que coincida con tu tipo de datos y quieres usar la anotación @DefaultCoder, la clase del codificador debe implementar un método de fábrica estático Coder.of(Class<T>).

  public class MyCustomCoder implements Coder {
    public static Coder<T> of(Class<T> clazz) {...}
    ...
  }

  @DefaultCoder(MyCustomCoder.class)
  public class MyCustomDataType {
    ...
  }
¿Te ha resultado útil esta página? Enviar comentarios:

Enviar comentarios sobre...

Si necesitas ayuda, visita nuestra página de asistencia.