Multiple Aggregate operations on the same column of a spark dataframe

by

Scala:

You can for example map over a list of functions with a defined mapping from name to function:

import org.apache.spark.sql.functions.{col, min, max, mean}
import org.apache.spark.sql.Column

val df = Seq((1L, 3.0), (1L, 3.0), (2L, -5.0)).toDF("k", "v")
val mapping: Map[String, Column => Column] = Map(
  "min" -> min, "max" -> max, "mean" -> avg)

val groupBy = Seq("k")
val aggregate = Seq("v")
val operations = Seq("min", "max", "mean")
val exprs = aggregate.flatMap(c => operations .map(f => mapping(f)(col(c))))

df.groupBy(groupBy.map(col): _*).agg(exprs.head, exprs.tail: _*).show
// +---+------+------+------+
// |  k|min(v)|max(v)|avg(v)|
// +---+------+------+------+
// |  1|   3.0|   3.0|   3.0|
// |  2|  -5.0|  -5.0|  -5.0|
// +---+------+------+------+

or

df.groupBy(groupBy.head, groupBy.tail: _*).agg(exprs.head, exprs.tail: _*).show

Unfortunately parser which is used internally SQLContext is not exposed publicly but you can always try to build plain SQL queries:

df.registerTempTable("df")
val groupExprs = groupBy.mkString(",")
val aggExprs = aggregate.flatMap(c => operations.map(
  f => s"$f($c) AS ${c}_${f}")
).mkString(",")

sqlContext.sql(s"SELECT $groupExprs, $aggExprs FROM df GROUP BY $groupExprs")

Python:

from pyspark.sql.functions import mean, sum, max, col

df = sc.parallelize([(1, 3.0), (1, 3.0), (2, -5.0)]).toDF(["k", "v"])
groupBy = ["k"]
aggregate = ["v"] 
funs = [mean, sum, max]

exprs = [f(col(c)) for f in funs for c in aggregate]

# or equivalent df.groupby(groupBy).agg(*exprs)
df.groupby(*groupBy).agg(*exprs)

See also:

More Related Contents:

  1. Spark SQL: apply aggregate functions to a list of columns
  2. DataFrame join optimization – Broadcast Hash Join
  3. How to access element of a VectorUDT column in a Spark DataFrame?
  4. What is the maximum size for a broadcast object in Spark?
  5. How to loop through each row of dataFrame in pyspark
  6. Why does Spark think this is a cross / Cartesian join
  7. How do I add an persistent column of row ids to Spark DataFrame?
  8. Convert null values to empty array in Spark DataFrame
  9. How to exclude multiple columns in Spark dataframe in Python
  10. Why is predicate pushdown not used in typed Dataset API (vs untyped DataFrame API)?
  11. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  12. Finding duplicates from large data set using Apache Spark
  13. How to make good reproducible Apache Spark examples
  14. How to select the first row of each group?
  15. Querying Spark SQL DataFrame with complex types
  16. While writing to hdfs path getting error java.io.IOException: Failed to rename
  17. How to control partition size in Spark SQL
  18. How to import multiple csv files in a single load?
  19. Requirements for converting Spark dataframe to Pandas/R dataframe
  20. Retrieve top n in each group of a DataFrame in pyspark
  21. Provide schema while reading csv file as a dataframe
  22. Spark sql queries vs dataframe functions
  23. Joining Spark dataframes on the key
  24. How to split a list to multiple columns in Pyspark?
  25. How to explode multiple columns of a dataframe in pyspark
  26. PySpark: multiple conditions in when clause
  27. How does createOrReplaceTempView work in Spark?
  28. PySpark – get row number for each row in a group
  29. Multiple condition filter on dataframe
  30. How to return a “Tuple type” in a UDF in PySpark?

Leave a Comment