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

Questo tutorial è rivolto agli utenti di App Engine MapReduce. Mostra come eseguire la migrazione dall'utilizzo di App Engine MapReduce ad Apache Beam e 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 provenienti da origini esterne
  • Trasformare i dati da un formato all'altro
  • Esportazione dei 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.

Richieste 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 del database Datastore in un data warehouse BigQuery per l'elaborazione analitica utilizzando SQL.
  • Utilizza Dataflow come alternativa ad App Engine MapReduce per la manutenzione e/o gli aggiornamenti del 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 dei dati. Apache Beam è un modello di programmazione unificato che fornisce SDK per definire i flussi di lavoro di elaborazione dei dati. Utilizza Apache Beam per creare pipeline complesse sia in batch che in 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 elaborazione dei dati, aggiungi la libreria MapReduce e, una volta definite le specifiche e le impostazioni del job, crei e avvii il job in un solo passaggio utilizzando il metodo start() statico sulla classe di job appropriata.

Per i job di mappatura, crea i job Input e Output e il job Map che esegue la mappatura. Per i job MapReduce di App Engine, crei 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 di output per leggere e scrivere dalle origini e dalle destinazioni dei dati. Utilizza trasformazioni di dati predefinite (o scrivi le tue) per implementare le trasformazioni di dati. Una volta che il codice è pronto, esegui il deployment della pipeline nel servizio Dataflow.

Conversione di job MapReduce di App Engine in pipeline Apache Beam

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

Java

MapReduce di App Engine Equivalente di 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> instead of GroupByKey e CombineValues.

Python

MapReduce di App Engine Equivalente di 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 di Apache Beam
Mappa beam.ParDo
Shuffle beam.GroupByKey
Riduci beam.Combine


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

Java

Tratto 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

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

Ulteriori 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 Cloud Dataflow

Se hai dimestichezza con le code di attività di App Engine, puoi pianificare i job ricorrenti utilizzando Cron. Questo esempio dimostra 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 Cloud Dataflow

Per monitorare i job MapReduce di App Engine, devi disporre di 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 gli scrittori di App Engine MapReduce sono chiamati origini dati e destinazioni o connettori I/O.

Apache Beam dispone di molti connettori I/O per diversi servizi Google Cloud, 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