Merging Eingänge in verteilten Anwendung

Question

EINLEITUNG

Ich muss verteilte Anwendung schreiben, die für 3 Datensätze maximale Anzahl von eindeutigen Werten zählt. Ich habe keine Erfahrung in diesem Bereich und kenne keine Rahmenbedingungen. Meine Eingabe konnte sieht wie folgt:

u1: u2,u3,u4,u5,u6 
u2: u1,u4,u6,u7,u8 
u3: u1,u4,u5,u9 
u4: u1,u2,u3,u6 
... 

Dann beginnt der Ergebnisse sein sollte:

(u1,u2,u3), u4,u5,u6,u7,u8,u9 => count=6 
(u1,u2,u4), u3,u5,u6,u7,u8 => count=5 
(u1,u3,u4), u2,u5,u6,u9  => count=4 
(u2,u3,u4), u1,u5,u6,u7,u8,u9 => count=6 
... 

Also mein Ansatz ist es, zunächst jede zwei Datensätze zu verschmelzen, und dann verschmelzen jedes fusionierte Paar mit jedem einzelnen Aufzeichnung.

FRAGE

Kann ich eine solche Operation zu tun, wie (merge) auf mehr als einer Eingabereihe auf der gleichen Zeit in framewors wie hadoop/Funkenarbeits? Oder ist mein Ansatz falsch und ich sollte das anders machen?

Jeder Rat wird geschätzt.

Answer 1

Kann ich einen solchen Vorgang wie das Arbeiten (Zusammenführen) von mehr als einer Eingangszeile gleichzeitig in Framewors wie hadoop/spark ausführen?

Ja, Sie können.

Oder vielleicht ist mein Ansatz falsch und ich sollte dies anders machen?

Es hängt von der Größe der Daten ab. Wenn Ihre Daten klein sind, ist es schneller und einfacher, dies lokal zu tun. Wenn Ihre Daten riesig sind, mindestens Hunderte von GBs, besteht die allgemeine Strategie darin, die Daten in HDFS (verteiltes Dateisystem) zu speichern und eine Analyse mit Mapreduce/Spark durchzuführen.

Ein Beispiel Funken Anwendung in scala geschrieben:

object MyCounter { 
     val sparkConf = new SparkConf().setAppName("My Counter") 
     val sc = new SparkContext(sparkConf) 

     def main(args: Array[String]) { 
     val inputFile = sc.textFile("hdfs:///inputfile.txt") 
     val keys = inputFile.map(line => line.substring(0, 2)) // get "u1" from "u1: u2,u3,u4,u5,u6" 

     val triplets = keys.cartesian(keys).cartesian(keys) 
      .map(z => (z._1._1, z._1._2, z._2)) 
      .filter(z => !z._1.equals(z._2) && !z._1.equals(z._3) && !z._2.equals(z._3)) // get "(u1,u2,u3)" triplets 

     // If you have small numbers of (u1,u2,u3) triplets, it's better prepare them locally. 

     val res = triplets.cartesian(inputFile).filter(z => { 
      z._2.startsWith(z._1._1) || z._2.startsWith(z._1._2) || z._2.startsWith(z._1._3) 
     }) // (u1,u2,u3) only matches line starts with u1,u2,u3, for example "u1: u2,u3,u4,u5,u6" 
      .reduceByKey((a, b) => a + b) // merge three lines 
      .map(z => { 
      val line = z._2 
      val values = line.split(",") 
      //count unique values using set 
      val set = new util.HashSet[String]() 
      for (value <- values) { 
      set.add(value) 
      } 
      "key=" + z._1 + ", count=" + set.size() // the result from one mapper is a string 
     }).collect() 

     for (line <- res) { 
      println(line) 
     } 
     } 
    } 

1. Der Code ist nicht getestet. Und ist nicht effizient. Es kann einige Optimierung haben (zum Beispiel, entfernen Sie unnötige Map-Reduce-Schritte.)
2. Sie können die gleiche Version mit Python/Java neu schreiben.
3. Sie können die gleiche Logik Hadoop/MapReduce