Definir una transformación de ParDo

En un post anterior sobre el funcionamiento de Apache Beam hemos mencionado algunos tipos de transformaciones que este modelo utiliza, entre los que están ParDo, GroupByKey, Flatten y Partition.

En este artículo, utilizaremos uno de ellos: ParDo. Veamos, pues, cómo se hace una transformación de ParDo.

Definir una transformación de ParDo

La transformación de ParDo nos permite aplicar nuestro código Python sobre colecciones de datos. Para definir esta transformación tenemos que definir una clase de Python que herede de beam.DoFn y codificar nuestras operaciones en el método process.

¿Cómo generamos esta clase de ParDo? Pues generamos un class, donde ponemos el nombre que queramos. Al manifestarle que debe heredar beam.DoFn, lo que estamos diciéndole al algoritmo es que la clase que se va a implementar es una transformación de Apache Beam.

Para que después Apache Beam ejecute esa transformación de forma correcta, tenemos que implementar el método def process, al que le pasamos self y element. En element va a recibir cada uno de los elementos de la colección que va obteniendo, que en este caso sería una fila:

#Definir una transformación de ParDo
class ComputeWordLengthFn (beam.DoFn):
      def process (self, element):
            return [len (element)]

#Definir una transformación de ParDo
#Create a PCollection
with beam.Pipeline (options = PipelineOptions ()) as p:
      words = (
            p | beam.Create ([
                  'hello',
                  'welcome',
                  'nice',
                  'work',
]))
word_lengths = words | beam.ParDo (ComputeWordLengthFn ())

En el caso de que nuestro código sea muy sencillo y podamos meterlo en una función lambda, podemos usar la transformación FlatMap para aplicarla sobre cada elemento de nuestra PCollection:

#Definir una transformación de ParDo
#Create a PCollection
with beam.Pipeline (options = PipelineOptions ()) as p:
      words = (
            p | beam.Create ([
                  'hello',
                  'welcome',
                  'nice',
                  'work',
]))
word_lengths = words | beam.FlatMap (lambda word: [len (word)])

Y si es aún más sencillo, como en este caso, en el que solo debemos aplicar la función len sobre cada elemento, podemos usar la transformación Map, que aplica la función indicada a cada elemento, como puede ser un len o un print.

#Definir una transformación de ParDo
#Create a PCollection
with beam.Pipeline (options = PipelineOptions ()) as p:
      words = (
            p | beam.Create ([
                  'hello',
                  'welcome',
                  'nice',
                  'work',
]))
word_lengths = words | beam.Map (len)

#Definir una transformación de ParDo
#Create a PCollection
with beam.Pipeline (options = PipelineOptions ()) as p:
      words = (
            p | beam.Create ([
                  'hello',
                  'welcome',
                  'nice',
                  'work',
]))
word_lengths = words | beam.Map (len)
output = word_lengths | beam.Map (print)

#Definir una transformación de ParDo
#Create a PCollection
with beam.Pipeline (options = PipelineOptions ()) as p:
      output = (
            p | beam.Create ([
                  'hello',
                  'welcome',
                  'nice',
                  'work',
            ])
            | beam.Map (len)
            | beam.Map (print)
)

Ahora que has entendido cómo definir una transformación en ParDo, es la hora de seguir formándote en alguna de las numerosas temáticas del mundo del Big Data. En KeepCoding te ofrecemos la posibilidad de aprender con grandes expertos, que te guiarán a través de la teoría y la práctica para que, en unos pocos meses, te transformes en un gran profesional IT. Échale un vistazo al temario de nuestro Big Data, Inteligencia Artificial & Machine Learning Full Stack Bootcamp y descubre esta formación intensiva. ¡Solicita ahora mismo más información y da el salto que cambiará tu futuro!