Reduzieren einer Liste von Fallklassen auf eine Anzahl der Fallklassen

Question

Ich habe derzeit eine Gruppe RDD der Form ((id, code), (list of events with keys id and code)). Unten sehen Sie die ID 000406106-01, der Code ist 496, und die einzelnen Ereignisse sind jeweils Fallklasse. Was ich zu tun hoffte, war eine RDD der Form ((id, code), count of events) zu erhalten. Im Wesentlichen wollte ich das CompactBuffer Objekt der Ereignisse in eine Anzahl der Ereignisse reduzieren. Irgendwelche Vorschläge?

ID   CODE    EVENT1              EVENT2 
((000406106-01,496),CompactBuffer(Diagnostic(000406106-01,Sun Apr 16 02:24:00 UTC 2006,496), Diagnostic(000406106-01,Fri Jul 20 15:30:00 UTC 2012,496), Diagnostic(000406106-01,Tue Dec 23 17:00:00 UTC 2014,496), Diagnostic(000406106-01,Wed Jan 06 20:45:00 UTC 2010,496), Diagnostic(000406106-01,Fri Mar 04 16:30:00 UTC 2011,496), Diagnostic(000406106-01,Sun Aug 04 04:51:00 UTC 2013,496), Diagnostic(000406106-01,Fri Mar 11 16:00:00 UTC 2011,496), Diagnostic(000406106-01,Tue Jul 10 13:45:00 UTC 2012,496), Diagnostic(000406106-01,Wed Jun 15 20:00:00 UTC 2005,496), Diagnostic(000406106-01,Tue Dec 29 13:30:00 UTC 2009,496), Diagnostic(000406106-01,Fri Jul 13 13:30:00 UTC 2012,496), Diagnostic(000406106-01,Thu Jul 26 03:40:00 UTC 2007,496), Diagnostic(000406106-01,Mon Jun 13 14:45:00 UTC 2005,496), Diagnostic(000406106-01,Wed Dec 24 18:00:00 UTC 2014,496), Diagnostic(000406106-01,Thu Mar 03 15:45:00 UTC 2011,496), Diagnostic(000406106-01,Wed Dec 31 15:00:00 UTC 2014,496), Diagnostic(000406106-01,Sat Jul 26 04:39:00 UTC 2008,496), Diagnostic(000406106-01,Thu Dec 31 20:30:00 UTC 2009,496))) 

Was ich suche:

 ID  CODE COUNT 
((000406106-01,496), 20) 

Edit: Aus Gründen der Klarheit ist hier, wie die RDD oben erzeugt wird:

val grpDiag = diagnostic.groupBy(diag => (diag.id, diag.code)) 

Wo Diagnose ist ein ungruppierten RDD von die obigen Daten.

Answer 1

Wenn das zweite Element des Tupels ist ein CompactBuffer und alles, was Sie brauchen, ist seine Länge ein mapValues mit _.size sollten Sie das gewünschte Ergebnis geben:

rdd.mapValues(_.size) 

Generell sollten Sie nur die Gruppierung vermeiden eine count zu finden und verwenden Sie reduceByKey als Ersatz:

val diagnostics: RDD[Diagnostic] = ??? 
diagnostics.map(d => ((d.id, d.code), 1L)).reduceByKey(_ + _)