Eseguire la migrazione da App Engine MapReduce ad Apache Beam e Dataflow

Apache Beam e Dataflow ti consentono di modernizzare i job MapReduce legacy di App Engine con l'elaborazione unificata di flussi e batch su un'infrastruttura completamente gestita. In questo tutorial imparerai a tradurre i concetti di App Engine MapReduce in pipeline Apache Beam ed eseguirli su Dataflow.

Perché eseguire la migrazione?

App Engine MapReduce è un modello di programmazione per l'elaborazione di grandi quantità di dati in modo parallelo e distribuito. È utile per attività di grandi dimensioni e di lunga durata che non possono essere gestite nell'ambito di una singola richiesta, ad esempio:

  • Analisi dei log delle applicazioni
  • Aggregazione di dati correlati da origini esterne
  • Trasformare i dati da un formato all'altro
  • Esportare i dati per l'analisi esterna

Tuttavia, App Engine MapReduce è una libreria open source gestita dalla community che si basa sui servizi App Engine e non è più supportata da Google.

Dataflow, invece, è completamente supportato da Google e offre funzionalità estese rispetto ad App Engine MapReduce.

Casi di migrazione

Di seguito sono riportati alcuni esempi di casi in cui potresti trarre vantaggio dalla migrazione da App Engine MapReduce ad Apache Beam e Dataflow:

  • Archivia i dati dell'applicazione di database Datastore in un data warehouse BigQuery per l'elaborazione analitica utilizzando SQL.
  • Utilizza Dataflow come alternativa ad App Engine MapReduce per la manutenzione o gli aggiornamenti del tuo set di dati Datastore.
  • Esegui il backup di parti del database Datastore in Cloud Storage.

Che cos'è Dataflow e Apache Beam?

Dataflow è un servizio gestito per l'esecuzione di un'ampia varietà di pattern di elaborazione dati. Apache Beam è un modello di programmazione unificato che fornisce SDK per la definizione di workflow di trattamento dati. Utilizza Apache Beam per creare pipeline complesse sia per l'elaborazione batch che per lo streaming ed eseguirle su Dataflow.

Inizia a utilizzare Dataflow e Apache Beam

Per iniziare, segui la guida rapida che preferisci:

Creazione ed esecuzione di una pipeline

Quando utilizzi App Engine MapReduce, crei classi di trattamento dati, aggiungi la libreria MapReduce e, una volta definite le specifiche e le impostazioni del job, crei e avvii il job in un unico passaggio utilizzando il metodo statico start() nella classe di job appropriata.

Per i job Map, crei le classi Input e Output e la classe Map che esegue la mappatura. Per i job MapReduce di App Engine, crei le classi Input e Output e definisci le classi Mapper e Reducer per le trasformazioni dei dati.

Con Apache Beam le cose funzionano in modo leggermente diverso: crei una pipeline. Utilizzi i connettori di input e output per leggere e scrivere dalle origini e dai sink di dati. Utilizzi trasformazioni dei dati predefinite (o scrivi le tue) per implementare le trasformazioni dei dati. Poi, una volta pronto il codice, esegui il deployment della pipeline nel servizio Dataflow.

Conversione dei job MapReduce di App Engine in pipeline Apache Beam

La tabella seguente mostra gli equivalenti Apache Beam dei passaggi map, shuffle e reduce del modello MapReduce di App Engine.

Java

MapReduce di App Engine Equivalente Apache Beam
Mappa MapElements<InputT,OutputT>
Shuffle GroupByKey<K,V>
Riduci Combine.GroupedValues<K,InputT,OutputT>

Una pratica comune è utilizzare Combine.PerKey<K,InputT,OutputT> anziché GroupByKey e CombineValues.

Python

MapReduce di App Engine Equivalente Apache Beam
Mappa beam.Map
Shuffle beam.GroupByKey
Riduci beam.CombineValues

Una pratica comune è utilizzare beam.CombinePerKey anziché beam.GroupByKey e beam.CombineValues.

Vai

MapReduce di App Engine Equivalente Apache Beam
Mappa beam.ParDo
Shuffle beam.GroupByKey
Riduci beam.Combine


Il seguente codice campione mostra come implementare il modello MapReduce di App Engine in Apache Beam:

Java

Estratto da 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

Tratto da 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)

Vai

Estratto da 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 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)

Altri vantaggi di Apache Beam e Dataflow

Se scegli di eseguire la migrazione dei job MapReduce di App Engine alle pipeline Apache Beam, potrai usufruire di diverse funzionalità offerte da Apache Beam e Dataflow.

Pianificazione dei job Dataflow

Se hai familiarità con le code di attività di App Engine, puoi pianificare i job ricorrenti utilizzando Cron. Questo esempio mostra come utilizzare App Engine Cron per pianificare le pipeline Apache Beam.

Esistono diversi modi aggiuntivi per pianificare l'esecuzione della pipeline. Puoi:

Monitoraggio dei job Dataflow

Per monitorare i job MapReduce di App Engine, fai affidamento su un URL ospitato su appspot.com.

Quando esegui le pipeline utilizzando il servizio gestito Dataflow, puoi monitorarle utilizzando la comoda interfaccia utente di monitoraggio basata sul web di Dataflow. Puoi anche utilizzare Cloud Monitoring per ulteriori informazioni sulle tue pipeline.

Lettura e scrittura

In Apache Beam, i lettori e i writer di App Engine MapReduce sono chiamati origini dati e sink o connettori I/O .

Apache Beam dispone di molti connettori I/O per diversi Google Cloud servizi, come Bigtable, BigQuery, Datastore, Cloud SQL e altri. Esistono anche connettori I/O creati dai collaboratori di Apache Beam per servizi non Google, come Apache Cassandra e MongoDB.

Passaggi successivi