Wie mehrere JavaRDDs gleichzeitig verarbeiten?

Question

Ich hatte eine große Datenmenge des Formats csv und ich muss einige RDD-Operationen für dieses Dataset durchführen, ohne die APIs DataFrames/Dataset und SparkSQL zu verwenden. Dazu habe ich jede Datenspalte in ein separates JavaRDD geladen.

Hier ist mein Beispieldatensatz:

id name address rank 
1001 john NY  68 
1002 kevin NZ  72 
1003 steve WA  64 

Hier ist der Code, den ich bisher versucht:

diese

JavaRDD<String> diskfile = sc.textFile("/Users/hadoop/Downloads/a.csv"); 
JavaRDD<String> idRDD=diskfile.flatMap(line -> Arrays.asList(line.split(",")[0])); 
JavaRDD<String> nameRDD=diskfile.flatMap(line -> Arrays.asList(line.split(",")[1])); 
JavaRDD<String> addressRDD=diskfile.flatMap(line -> Arrays.asList(line.split(",")[2])); 

nachdem ich reduceByKey sowohl addressRDD und nameRDD wie folgt angewandt:

JavaPairRDD<String,Integer> addresspair=address.mapToPair(t -> new Tuple2 <String,Integer>(t,1)).reduceByKey((x, y) -> x + y); 
JavaPairRDD<String,Integer> namepair=nameRDD.mapToPair(t -> new Tuple2 <String,Integer>(t,1)).reduceByKey((x, y) -> x + y); 

Problem:

Ich habe soryByVale (swap the key -values) auf addresspair angewendet und erhalte einen Adresswert (result), der am häufigsten aufgetreten ist. Jetzt muss ich alle erforderlichen Spalten der csv-Datei zurückgeben, die Adressenfeld als result enthält.

Answer 1

Sie können filter wie unten verwenden.

JavaRDD<String> filteredData = diskfile.filter(add -> add.contains(result)); 
filteredData.foreach(data -> { 
      System.out.println(data); 
     });