Spark DataFrame: функция множественного агрегирования для нескольких столбцов

Question

У меня есть список функций агрегирования, псевдонимов и других, таких как конфигурация JSON, например

{
    "aggregation": [{
        "alias_column_name1": {
            "sum": "<columnName1>"
        }
    }, {
        "alias_column_name2": {
            "sum": "<columnName1>"
        }
    }]
}

В настоящее время я выполняю то же самое с помощью следующего кода:

val col1:Column = sum(<dataframeName>(<columnName1>)).alias(<alias_column_name1>)
val col2:Column = sum(<dataframeName>(<columnName2>)).alias(<alias_column_name2>)
dataframe.groupby(..).agg(col1, col2)

Но у меня много агрегацииКонфигурация, и я хочу передать список таких в методе агрегации, как

val colList = List[Column](col1, col2)
dataframe.groupby(..).agg(colList)

Как я могу добиться того же?Спасибо

Версии:

Scala : 2.11
Spark : 2.2.2
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.2.2"
libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.2.2"
libraryDependencies += "org.apache.spark" %% "spark-hive" % "2.2.2"

Answer 1

Отдельный список столбцов и функций

Допустим, у вас есть список функций:

import org.apache.spark.sql.Column
import org.apache.spark.sql.functions._

val funs: Seq[Column => Column] = Seq(sum _, min _, max _)

и список столбцов

val cols: Seq[Column] = Seq($"y", $"z")

и набор данных

val df = Seq((1, 2, 3), (1, 4, 5) ).toDF("x", "y", "z")

, вы можете объединить

val exprs = for { c <- cols; f <- funs} yield f(c)

и затем

df.groupBy($"x").agg(exprs.head, exprs.tail: _*)

То же самое можно сделать в PySpark:

from pyspark.sql import functions as F

funs = [F.sum, F.min, F.max]
cols = ["y", "z"]

df = spark.createDataFrame([(1, 2, 3), (1, 4, 5)], ("x", "y", "z"))

df.groupBy("x").agg(*[f(c) for c in cols for f in funs])

Предопределенный список операций для каждого столбца

Если вы хотите начать с предопределенного набора псевдонимов, столбцов и функций, как показано в вашем вопросе, он можетпроще просто перестроить его на

trait AggregationOp {
  def expr: Column
}

case class FuncAggregationOp(c: Column, func: Column => Column, alias: String
    ) extends AggregationOp {
  def expr = func(c).alias(alias)
}

val ops: Seq[AggregationOp] = Seq(
   FuncAggregationOp($"y", sum _, "alias_column_name1"),
   FuncAggregationOp($"z", sum _, "alias_column_name2")
)
val exprs = ops.map(_.expr)

df.groupBy($"x").agg(exprs.head, exprs.tail: _*)

Вы можете легко настроить это для других случаев:

case class StringAggregationOp(c: String, func: String, alias: String
    ) extends AggregationOp {
  def expr = org.apache.spark.sql.functions.expr(s"${func}(`${c}`)").alias(alias)
}

val ops: Seq[AggregationOp] = Seq(
   StringAggregationOp("y", "sum", "alias_column_name1"),
   StringAggregationOp("z", "sum", "alias_column_name2")
)

Эквивалент Python может выглядеть примерно так:

from collections import namedtuple
from pyspark.sql import functions as F

class AggregationOp(namedtuple("Op", ["c", "func", "alias"])):
    def expr(self):
        if callable(self.func):
            return self.func(self.c).alias(self.alias)
        else:
            return F.expr("{func}(`{c}`)".format
                (func = self.func, c = self.c)).alias(self.alias)

ops = [
    AggregationOp("y", "sum", "alias_column_name1"),
    AggregationOp("z", "sum", "alias_column_name2")
]

 df.groupBy("x").agg(*[op.expr() for op in ops])

Смежный вопрос :

• Несколько операций агрегирования с одним и тем же столбцом кадра данных искры
• SparkSQL: применить агрегатфункции к списку столбцов