TopolearnAggregations power almost every report and metric built with Spark.
df.groupBy("country", "year").agg(...) groups rows and computes one output row per group. Pass aggregate expressions from pyspark.sql.functions:
F.sum,F.avg,F.min,F.max,F.count("*"),F.count("col")(non-null only)F.countDistinct("user")(exact, shuffles more) orF.approx_count_distinct("user")(fast HyperLogLog estimate)F.collect_list,F.collect_set,F.first,F.percentile_approx- conditional aggregates like
F.sum(F.when(F.col("type") == "refund", F.col("amount")).otherwise(0)), orF.count_if(cond)
Name results with .alias(). Without a groupBy, df.agg(F.max("x")) aggregates the whole DataFrame.
groupBy(...).pivot("month", ["jan", "feb"]).sum("sales") turns values into columns. Listing the pivot values explicitly avoids an extra job to discover them.
Spark aggregates in two phases: partial aggregation within each partition, a shuffle of the partial results by key, then final aggregation. That's why grouping is efficient even on huge data, but also why a single extremely common key (skew) can make one task much slower than the rest.
After groupBy, output rows are in no particular order; add orderBy if you need one.
Example
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.master("local[2]").getOrCreate()
sales = spark.createDataFrame(
[("A", "sale", 10.0, "u1"), ("B", "sale", 5.0, "u2"), ("A", "refund", -3.0, "u1"),
("B", "sale", 7.0, "u3"), ("A", "sale", 4.0, "u4")],
"shop STRING, kind STRING, amount DOUBLE, user STRING",
)
summary = (
sales.groupBy("shop")
.agg(F.count("*").alias("n"),
F.sum("amount").alias("net"),
F.sum(F.when(F.col("kind") == "refund", F.col("amount")).otherwise(0)).alias("refunds"),
F.countDistinct("user").alias("users"))
.orderBy("shop")
)
print([tuple(r) for r in summary.collect()])
# [('A', 3, 11.0, -3.0, 2), ('B', 2, 12.0, 0.0, 2)]
wide = sales.groupBy("shop").pivot("kind", ["sale", "refund"]).sum("amount").orderBy("shop")
print([tuple(r) for r in wide.collect()]) # [('A', 14.0, -3.0), ('B', 12.0, None)]Remember
groupBy(...).agg(...)with functions frompyspark.sql.functions- Aggregation is partial per partition, then shuffled and merged
- Give
pivotits values explicitly to save a job
Watch out for
- Assuming groupBy output is sorted by key.