SparkR gapply - функция возвращает многорядные R dataframe

Question

Допустим, я хочу, чтобы выполнить что-то выглядит следующим образом:

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)) 
    } 
) 

где возвращение функции имеет несколько строк. Чтобы быть ясным, примеры в документации (которые, к сожалению, вторят большей части документации Spark, где примеры тривиально просты) не помогают мне определить, будет ли это обрабатываться, как я ожидаю.

Я ожидал бы, что результатом этого было бы для групп k, созданных в DataFrame с выходными строками n_k для каждой группы, что результат вызова gapply() имел бы сумму (1..k, n_k) строк , где ключевое значение реплицируется для каждой из строк n_k для каждой группы в ключе k ... Однако поле схемы подсказывает мне, что это не так, как это будет обрабатываться - на самом деле это предполагает, что он либо захочет результат вставляется в одну строку.

Надеюсь, это ясно, хотя и теоретическое (извините, что я не могу поделиться своим примером с фактическим кодом). Может ли кто-нибудь проверить или объяснить, как эта функция будет фактически рассматриваться?

Answer 1

Точные ожидания в отношении ввода и вывода четко указаны в the official documentation:

Применить функцию к каждой группе в SparkDataFrame. Эта функция должна применяться к каждой группе SparkDataFrame и должна иметь только два параметра: групповой ключ и R data.frame, соответствующий этому ключу. Группы выбираются из столбцов (ов) SparkDataFrames. Выходной сигнал функции должен быть data.frame.

Схема определяет формат строки получаемого SparkDataFrame. Он должен представлять схему вывода функции R на основе типов данных Spark. Имена столбцов возвращенных data.frame устанавливаются пользователем. Ниже приведено сопоставление типов данных между R и Spark.

Другими словами ваша функция должна принимать key и data.frame строк, соответствующих этому ключу и вернуть data.frame, которые могут быть представлены с использованием типов Спарк SQL со схемой, представленной в качестве аргумента schema. Нет ограничений на количество строк. Например, вы можете применить тождественное преобразование следующим образом:

df <- as.DataFrame(iris) 

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

точно так же, как агрегаты:

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")) 
) 

хотя на практике следует отдавать предпочтение агрегирования непосредственно на DataFrame (groupBy %>% agg).