"java.lang.UnsupportedOperationException: leer Sammlung"

Question

Ich arbeite mit Spark-2.1.1 und Scala 2.11.8

ich meinen Code in Funken Shell ausführen. Dies ist der Code, den ich Ausführung

val read_file1 = sc.textFile("Path to file 1"); 

val uid = read_file1.map(line => line.split(",")).map(array => array.map(arr => { 
| if(arr.contains(":")) (array(2).split(":")(0), arr.split(":")(0)) 
| else (array(2).split(":")(0), arr)})) 

val rdd1 = uid.map(array => array.drop(4)).flatMap(array => array.toSeq).map(y=>(y,1)).reduceByKey(_+_) 

Meine Ausgabe dieses Codes ist:

((v67430612_serv78i, fb_201906266952256),1) 
((v74005958_serv35i, fb_128431994336303),1) 

jedoch für die Ausgänge zwei RDDs, wenn ich ausführen:

uid2.map(x => ((x._1, x._2), x._3)).join(rdd1).map(y => ((y._1._1, y._1._2, y._2._1), y._2._2)) 

ich der Fehler:

Warum bekomme ich dieser Fehler?

Hier sind Proben der Eingabedateien: -

Datei 1:

2017-05-09 21:52:42 , 1494391962 , p69465323_serv80i:10:450 , 7 , fb_406423006398063:396560, guest_861067032060185_android:671051, fb_100000829486587:186589, fb_100007900293502:407374, fb_172395756592775:649795 
2017-05-09 21:52:42 , 1494391962 , z67265107_serv77i:4:45 , 2:Re , fb_106996523208498:110066, fb_274049626104849:86632, fb_111857069377742:69348, fb_127277511127344:46246 

File 2:

fb_100008724660685,302502,-450,v300430479_serv73i:10:450,switchtable,2017-04-30 00:00:00  
fb_190306964768414,147785,-6580,r308423810_serv31i::20,invite,2017-04-30 00:00:00 

ich dies gerade bemerkt: Wenn ich die Ausführung

rdd1.take(10).foreach(println) or rdd1.first() 

Ich bekomme diese Nachricht vor dem Ausgang:

WARN Executor: Managed memory leak detected; size = 39979424 bytes, TID = 11 

Ich weiß nicht, ob das etwas mit dem Problem zu tun haben könnte ??

Noch ein Hinweis: Dieser Fehler tritt nur auf, wenn ich

res.first() 

für

uid2.map(x => ((x._1, x._2), x._3)).join(rdd1).map(y => ((y._1._1, y._1._2, y._2._1), y._2._2)) 

Auf tun

res.take(10).foreach(println) 

bekomme ich keine Ausgabe aber kein Fehler zurückgegeben entweder.

Answer 1

Sie haben trim die Leerzeichen in den Tupeln, die aus gespaltener Linie erstellt wurden, so dass nichts verbunden wurde, da sie nicht übereinstimmten. Wenn Sie also take aus einem leeren rdd versuchten, wurde eine Ausnahme ausgelöst.

Sie können folgende Lösung verwenden. Es funktioniert in meinem.

val read_file1 = sc.textFile("Path to file 1"); 

val uid = read_file1.map(line => line.split(",")).map(array => array.map(arr => { 
    if(arr.contains(":")) (array(2).split(":")(0).trim, arr.split(":")(0).trim) 
    else (array(2).split(":")(0).trim, arr.trim)})) 

val rdd1 = uid.map(array => array.drop(4)).flatMap(array => array.toSeq).map(y=>(y,1)).reduceByKey(_+_) 


val read_file2 = sc.textFile("Path to File 2"); 
val uid2 = read_file2.map(line => {var arr = line.split(","); (arr(3).split(":")(0).trim,arr(0).trim,arr(2).trim)}); 

val res = uid2.map(x => ((x._1, x._2), x._3)).join(rdd1).map(y => ((y._1._1, y._1._2, y._2._1), y._2._2)) 
res.take(10).foreach(println) 

Answer 2

Sie erhalten eine leere Sammlung nach dem join, geschieht es, wenn jetzt gibt es Tasten in RDDs entspricht. Entweder sind die Tasten nicht beschnitten, falsch geschnitten oder es gab überhaupt keine Übereinstimmungen. Ich schlage vor zu prüfen, ob es passende Schlüssel in Ihren Dateien/rdds gibt, zu überprüfen, ob die Daten korrekt extrahiert wurden und zu prüfen, ob Sie inner join statt left oder right outer join benötigen.