Migrar de MapReduce de App Engine a Apache Beam y Dataflow

Este tutorial está dirigido a los usuarios de MapReduce de App Engine. En él se explica cómo migrar de App Engine MapReduce a Apache Beam y Dataflow.

¿Por qué migrar?

MapReduce de App Engine es un modelo de programación para procesar grandes cantidades de datos de forma paralela y distribuida. Es útil para tareas grandes y de larga duración que no se pueden gestionar en el ámbito de una sola solicitud, como las siguientes:

  • Analizar registros de aplicaciones
  • Agregación de datos relacionados de fuentes externas
  • Transformar datos de un formato a otro
  • Exportar datos para realizar análisis externos

Sin embargo, MapReduce de App Engine es una biblioteca de código abierto mantenida por la comunidad que se basa en los servicios de App Engine y que Google ya no admite.

Por otro lado, Google ofrece asistencia completa para Dataflow, que proporciona funciones ampliadas en comparación con MapReduce de App Engine.

Casos de migración

A continuación, se muestran algunos ejemplos de casos en los que podría ser útil migrar de App Engine MapReduce a Apache Beam y Dataflow:

  • Almacena los datos de la aplicación de la base de datos de Datastore en un almacén de datos de BigQuery para procesarlos analíticamente con SQL.
  • Usa Dataflow como alternativa a MapReduce de App Engine para mantener o actualizar tu conjunto de datos de Datastore.
  • Crea copias de seguridad de partes de tu base de datos de Datastore en Cloud Storage.

¿Qué son Dataflow y Apache Beam?

Dataflow es un servicio gestionado para ejecutar una amplia variedad de patrones de tratamiento de datos. Apache Beam es un modelo de programación unificado que proporciona SDKs para definir flujos de trabajo de procesamiento de datos. Usa Apache Beam para crear flujos complejos por lotes y en tiempo real, y ejecútalos en Dataflow.

Empezar a usar Dataflow y Apache Beam

Para empezar, sigue la guía de inicio rápido que prefieras:

Crear y ejecutar un flujo de procesamiento

Cuando usas MapReduce de App Engine, creas clases de procesamiento de datos, añades la biblioteca MapReduce y, una vez que se han definido las especificaciones y los ajustes del trabajo, creas e inicias el trabajo en un solo paso mediante el método estático start() de la clase de trabajo adecuada.

En el caso de los trabajos de Map, se crean las clases Input y Output, así como la clase Map que realiza la asignación. En el caso de las tareas MapReduce de App Engine, se crean clases Input y Output, y se definen las clases Mapper y Reducer para las transformaciones de datos.

Con Apache Beam, las cosas se hacen de forma ligeramente diferente: se crea una pipeline. Utilizas conectores de entrada y salida para leer y escribir datos de tus fuentes y receptores de datos. Para implementar las transformaciones de datos, puedes usar transformaciones de datos predefinidas o escribir las tuyas propias. Después, cuando el código esté listo, despliega el flujo de procesamiento en el servicio Dataflow.

Convertir tareas MapReduce de App Engine en flujos de procesamiento de Apache Beam

En la siguiente tabla se muestran los equivalentes de Apache Beam de los pasos map, shuffle y reduce del modelo MapReduce de App Engine.

Java

MapReduce de App Engine Equivalente de Apache Beam
Mapa MapElements<InputT,OutputT>
Shuffle GroupByKey<K,V>
Reduce Combine.GroupedValues<K,InputT,OutputT>

Una práctica habitual es usar Combine.PerKey<K,InputT,OutputT> en lugar de GroupByKey y CombineValues.

Python

MapReduce de App Engine Equivalente de Apache Beam
Mapa beam.Map
Shuffle beam.GroupByKey
Reduce beam.CombineValues

Una práctica habitual es usar beam.CombinePerKey en lugar de beam.GroupByKey y beam.CombineValues.

Go

MapReduce de App Engine Equivalente de Apache Beam
Mapa beam.ParDo
Shuffle beam.GroupByKey
Reduce beam.Combine


En el siguiente código de ejemplo se muestra cómo implementar el modelo MapReduce de App Engine en Apache Beam:

Java

Tomado de MinimalWordCount.java: 
p.apply(TextIO.read().from("gs://apache-beam-samples/shakespeare/*"))

 // Apply a ParDo that returns a PCollection, where each element is an
 // individual word in Shakespeare's texts.
 .apply("ExtractWords", ParDo.of(new DoFn<String, String>() {
                   @ProcessElement
                   public void processElement(ProcessContext c) {
                     for (String word : c.element().split(ExampleUtils.TOKENIZER_PATTERN)) {
                       if (!word.isEmpty()) {
                         c.output(word);
                       }
                     }
                   }
                 }))

 // Apply the Count transform that returns a new PCollection of key/value pairs,
 // where each key represents a unique word in the text.
 .apply(Count.perElement())

 // Apply a MapElements transform that formats our PCollection of word counts
 // into a printable string, suitable for writing to an output file.
 .apply("FormatResults", MapElements.via(new SimpleFunction<KV<String, Long>, String>() {
                   @Override
                   public String apply(KV<String, Long> input) {
                     return input.getKey() + ": " + input.getValue();
                   }
                 }))

 // Apply a write transform that writes the contents of the PCollection to a
 // series of text files.
 .apply(TextIO.write().to("wordcounts"));

Python

Tomado de wordcount_minimal.py: 
# Read the text file[pattern] into a PCollection.
lines = p | ReadFromText(known_args.input)

# Count the occurrences of each word.
counts = (
    lines
    | 'Split' >> (beam.FlatMap(lambda x: re.findall(r'[A-Za-z\']+', x))
                  .with_output_types(unicode))
    | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
    | 'GroupAndSum' >> beam.CombinePerKey(sum))

# Format the counts into a PCollection of strings.
output = counts | 'Format' >> beam.Map(lambda (w, c): '%s: %s' % (w, c))

# Write the output using a "Write" transform that has side effects.
# pylint: disable=expression-not-assigned
output | WriteToText(known_args.output)

Go

Tomado de minimal_wordcount.go: 
// beam.Init() is an initialization hook that must be called on startup.
beam.Init()

// Create the Pipeline object and root scope.
p := beam.NewPipeline()
s := p.Root()
lines := textio.Read(s, "gs://apache-beam-samples/shakespeare/kinglear.txt")

// Invoke a ParDo transform on our PCollection of text lines.
// This ParDo invokes a DoFn (defined in-line) on each element that
// tokenizes the text line into individual words. The ParDo returns a
// PCollection of type string, where each element is an individual word in
// Shakespeare's collected texts.
words := beam.ParDo(s, func(line string, emit func(string)) {
    for _, word := range wordRE.FindAllString(line, -1) {
        emit(word)
    }
}, lines)

// Invoke the stats.Count transform on our PCollection of
// individual words. The Count transform returns a new PCollection of
// key/value pairs, where each key represents a unique word in the text.
// The associated value is the occurrence count for that word.
counted := stats.Count(s, words)

// Use a ParDo to format our PCollection of word counts into a printable
// string, suitable for writing to an output file. When each element
// produces exactly one element, the DoFn can simply return it.
formatted := beam.ParDo(s, func(w string, c int) string {
    return fmt.Sprintf("%s: %v", w, c)
}, counted)

// Invoke textio.Write at the end of the pipeline to write
// the contents of a PCollection (in this case, our PCollection of
// formatted strings) to a text file.
textio.Write(s, "wordcounts.txt", formatted)

Ventajas adicionales de Apache Beam y Dataflow

Si decides migrar tus trabajos de MapReduce de App Engine a las canalizaciones de Apache Beam, podrás disfrutar de varias funciones que ofrecen Apache Beam y Dataflow.

Programar tareas de Cloud Dataflow

Si conoces las colas de tareas de App Engine, puedes programar tus tareas periódicas con Cron. En este ejemplo se muestra cómo usar Cron de App Engine para programar tus pipelines de Apache Beam.

Hay varias formas adicionales de programar la ejecución de tu canalización. Puedes hacer lo siguiente:

Monitorizar tareas de Cloud Dataflow

Para monitorizar tus trabajos de MapReduce de App Engine, necesitas una URL alojada en appspot.com.

Cuando ejecutas tus canalizaciones con el servicio gestionado de Dataflow, puedes monitorizarlas mediante la práctica interfaz de usuario de monitorización basada en la Web de Dataflow. También puedes usar Cloud Monitoring para obtener más información sobre tus flujos de procesamiento.

Lectura y escritura

En Apache Beam, los lectores y escritores de MapReduce de App Engine se denominan fuentes y receptores de datos, o conectores de E/S.

Apache Beam tiene muchos conectores de entrada/salida para varios servicios de Google Cloud, como Bigtable, BigQuery, Datastore y Cloud SQL, entre otros. También hay conectores de entrada/salida creados por colaboradores de Apache Beam para servicios que no son de Google, como Apache Cassandra y MongoDB.

Siguientes pasos