Aggregations & Grouping

Aggregation is always two moves: group the rows by one or more columns, then reduce each group to a summary. In Spark that's groupBy(...) followed by agg(...) (or a shortcut like .count()). Every row in a group is folded into a single output row.

def groupBy(cols: Column*): RelationalGroupedDataset
def agg(expr: Column, exprs: Column*): DataFrame
// shortcuts on the grouped dataset:
def count(): DataFrame     def sum(col: String*): DataFrame     def avg(col: String*): DataFrame

import org.apache.spark.sql.functions.count

orders
  .groupBy($"category")
  .agg(count("*").as("n"))
  .orderBy($"category")
  .show(false)

The workhorses. Inside one agg you can compute as many measures as you like — each becomes a column. This is how you turn a raw event table into a revenue summary.

def sum(e: Column): Column      def avg(e: Column): Column       def mean(e: Column): Column
def min(e: Column): Column      def max(e: Column): Column       def count(e: Column): Column
// all live in org.apache.spark.sql.functions

import org.apache.spark.sql.functions.{sum, avg, min, max, round}

orders.groupBy($"category").agg(
  sum($"amount").as("revenue"),
  round(avg($"amount"), 2).as("avg_amount"),
  min($"amount").as("lo"),
  max($"amount").as("hi")
).orderBy($"revenue".desc).show(false)

Counting unique values — how many distinct customers, products, or sessions. countDistinct is exact; approx_count_distinct trades a little accuracy (HyperLogLog) for big speed on huge cardinalities.

def countDistinct(expr: Column, exprs: Column*): Column
def approx_count_distinct(e: Column): Column
def approx_count_distinct(e: Column, rsd: Double): Column   // rsd = max relative error

import org.apache.spark.sql.functions.countDistinct

orders.agg(
  count("*").as("rows"),
  countDistinct($"user_id").as("buyers"),
  countDistinct($"category").as("categories")
).show(false)

Sometimes you don't want a number — you want the actual values rolled into an array. collect_list gathers every value in the group (with duplicates, unordered); collect_set gathers the distinct values. This is the standard way to denormalise "all products a customer bought" into one row.

def collect_list(e: Column): Column   // Array, keeps duplicates, no ordering guarantee
def collect_set(e: Column): Column    // Array of distinct values

import org.apache.spark.sql.functions.collect_list

orders.groupBy($"category")
  .agg(collect_list($"product").as("products"))
  .orderBy($"category").show(false)

pivot rotates distinct values of one column into new columns, filling each cell with an aggregate. It's how you build a cross-tab: category down the side, order status across the top, revenue in the cells.

def pivot(pivotColumn: String): RelationalGroupedDataset
def pivot(pivotColumn: String, values: Seq[Any]): RelationalGroupedDataset   // faster: values known
// used as: df.groupBy(...).pivot("col").agg(...)

orders.groupBy($"category")
  .pivot("status")
  .agg(sum($"amount"))
  .orderBy($"category").show(false)

The browser engine can't run pivot, but you can hand-roll the same cross-tab with CASE WHEN — which is exactly what Spark compiles a pivot down to. Run it:

// the CASE-WHEN form Spark generates under the hood
selectExpr(...)  // conceptually

rollup and cube add subtotal and grand-total rows to a grouped result. rollup(a, b) gives totals for each a plus one overall total; cube(a, b) gives every combination. The extra rows carry null in the grouped-out columns.

def rollup(cols: Column*): RelationalGroupedDataset   // hierarchical subtotals + grand total
def cube(cols: Column*): RelationalGroupedDataset      // all combinations of subtotals
def stddev(e: Column): Column   def variance(e: Column): Column   def corr(a: Column, b: Column): Column

orders.rollup($"category")
  .agg(sum($"amount").as("revenue"))
  .orderBy($"category".asc_nulls_last).show(false)

The statistical aggregates — stddev, variance, skewness, corr, covar_pop — summarise the shape of a numeric column, not just its total:

import org.apache.spark.sql.functions.{stddev, variance}

orders.agg(
  round(stddev($"amount"), 2).as("sd"),
  round(variance($"amount"), 2).as("var")
).show(false)