SparkR gapply - Funktion gibt eine mehrreihige R Datenrahmen

Question

Lassen Sie uns sagen, dass ich etwas ausführen wollen, wie folgt:

library(SparkR) 
... 
df = spark.read.parquet(<some_address>) 
df.gapply(
    df, 
    df$column1, 
    function(key, x) { 
     return(data.frame(x, newcol1=f1(x), newcol2=f2(x)) 
    } 
) 

, wo die Rückkehr der Funktion mehrere Zeilen hat. Um es klar zu sagen, die Beispiele in der Dokumentation (die leider viel von der Spark-Dokumentation widerspiegeln, wo die Beispiele trivial einfach sind) helfen mir nicht festzustellen, ob dies so gehandhabt wird, wie ich es erwarte.

Ich würde erwarten, dass das Ergebnis davon wäre, für k Gruppen im DataFrame mit n_k Ausgabezeilen pro Gruppe erstellt, dass das Ergebnis des Aufrufs gapply() Summe (1..k, n_k) Zeilen hätte , wobei der Schlüsselwert für jede der n_k Zeilen für jede Gruppe in Schlüssel k repliziert wird ... Das Schemafeld legt jedoch nahe, dass dies nicht so gehandhabt wird - in der Tat schlägt es vor, dass es entweder die Ergebnis in eine einzige Zeile geschoben.

Hoffentlich ist das klar, wenn auch theoretisch (Es tut mir leid, dass ich mein tatsächliches Codebeispiel nicht teilen kann). Kann jemand überprüfen oder erklären, wie eine solche Funktion tatsächlich behandelt wird?

Answer 1

Genaue Erwartungen in Bezug auf Ein- und Ausgabe sind klar in the official documentation angegeben:

jeder Gruppe eine Funktion Anwenden eines SparkDataFrame. Die Funktion soll auf jede Gruppe des SparkDataFrame angewendet werden und sollte nur zwei Parameter haben: Gruppierungsschlüssel und R data.frame entsprechend diesem Schlüssel. Die Gruppen werden ausgewählt aus SparkDataFrames Spalte (n). Die Ausgabe der Funktion sollte eine data.frame sein.

Schema gibt das Zeilenformat des resultierenden SparkDataFrame an. Es muss das Ausgabeschema der R-Funktion auf der Grundlage von Spark-Datentypen darstellen. Die Spaltennamen der zurückgegebenen data.frame werden vom Benutzer festgelegt. Unten ist die Datentypzuordnung zwischen R und Spark.

Mit anderen Worten: Ihre Funktion sollte ein key nehmen und data.frame der Zeilen zu diesem Schlüssel entspricht, und das Rück data.frame die Spark-SQL-Typen verwenden mit Schema als schema Argument versehen dargestellt werden kann. Es gibt keine Beschränkung hinsichtlich der Anzahl der Zeilen. Sie könnten zum Beispiel Identitätstransformation gelten wie folgt:

df <- as.DataFrame(iris) 

gapply(df, "Species", function(k, x) x, schema(df)) 

die gleiche Art und Weise wie Aggregationen:

gapply(df, "Species", 
    function(k, x) { 
    dplyr::summarize(dplyr::group_by(x, Species), max(Sepal_Width)) 
    }, 
    structType(
    structField("species", "string"), 
    structField("max_s_width", "double")) 
) 

obwohl in der Praxis Sie Aggregationen direkt auf DataFrame (groupBy %>% agg) bevorzugen sollten.