Definition
- HashAggregate – maintains an off-heap hash map from grouping key to mutable fixed-length buffer; fast and default for sum, count, avg, min, max.
- SortAggregate – sorts by grouping key, then aggregates runs of equal keys; handles buffers that are not mutable fixed-length.
- ObjectHashAggregate – hash map with arbitrary JVM object buffers for aggregates like
collect_listandcollect_set. - Automatic selection – Spark's planner picks based on whether all aggregate buffers support the fast mutable path.
To total the National Bank's spending per account, HashAggregate is like having a pigeonhole per account into which a clerk drops running totals instantly: fast, as long as each pigeonhole holds a simple fixed number. SortAggregate is like first sorting all slips into account order, then walking down the pile summing each account's run: slower but works when the thing being accumulated is awkward. ObjectHashAggregate is for when each pigeonhole must hold a growing list of items rather than a single number.
Understanding the Concepts
What are the three aggregation strategies?
Spark has three physical aggregation strategies. They differ in how they store and update the per-group buffer that accumulates partial results.
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
spark = SparkSession.builder.appName("NBI-Aggregation").getOrCreate()
txn = spark.createDataFrame([
("SAV2001", "Mumbai", 50000.00, "2026-01-15"),
("SAV2001", "Mumbai", 75000.00, "2026-01-16"),
("SAV2002", "Delhi", 12000.00, "2026-01-15"),
("CUR3001", "Chennai", 300000.00, "2026-01-16"),
("SAV2002", "Delhi", 8000.00, "2026-01-17")
], ["account_id", "branch", "amount", "txn_date"])
How does HashAggregate work?
HashAggregate is the fastest and preferred path. It keeps a hash map, backed by Tungsten off-heap memory, mapping each grouping key to a buffer of mutable, fixed-length values. For sum, count, min, max, and avg, the buffer is a few fixed-width slots updated in place.
# HashAggregate: fast path for sum, count, avg
result = txn.groupBy("branch").agg(
F.sum("amount").alias("total"),
F.count("*").alias("txn_count"),
F.avg("amount").alias("avg_amount")
)
result.explain()
result.show()
== Physical Plan ==
*(2) HashAggregate(keys=[branch], functions=[sum(amount), count(1), avg(amount)])
+- Exchange hashpartitioning(branch, 200)
+- *(1) HashAggregate(keys=[branch], functions=[partial_sum(amount), partial_count(1), partial_avg(amount)])
+- *(1) LocalTableScan
+-------+--------+---------+----------+
| branch| total|txn_count|avg_amount|
+-------+--------+---------+----------+
| Mumbai|125000.0| 2| 62500.0|
| Delhi| 20000.0| 2| 10000.0|
|Chennai|300000.0| 1| 300000.0|
+-------+--------+---------+----------+
How does SortAggregate work?
SortAggregate sorts the data by grouping key first (expensive), then scans the sorted stream aggregating runs of equal keys. It handles buffer types that the hash path cannot support.
# SortAggregate may appear for certain aggregate functions
# or when buffer types are not mutable fixed-length
# Force SortAggregate by disabling hash aggregation
spark.conf.set("spark.sql.execution.useObjectHashAggregateExec", "false")
# Example with string aggregation (may trigger SortAggregate)
result_sort = txn.groupBy("branch").agg(
F.first("account_id").alias("first_account")
)
result_sort.explain()
How does ObjectHashAggregate work?
ObjectHashAggregate handles aggregates whose buffer is an arbitrary JVM object, like collect_list and collect_set. It uses a hash map with object buffers and falls back to sort-based if the map grows too large.
# ObjectHashAggregate: for collect_list, collect_set
result_obj = txn.groupBy("branch").agg(
F.collect_list("amount").alias("all_amounts"),
F.collect_set("account_id").alias("unique_accounts")
)
result_obj.explain()
result_obj.show(truncate=False)
== Physical Plan ==
ObjectHashAggregate(keys=[branch], functions=[collect_list(amount), collect_set(account_id)])
+- Exchange hashpartitioning(branch, 200)
+- ObjectHashAggregate(keys=[branch], functions=[partial_collect_list(amount), partial_collect_set(account_id)])
+- LocalTableScan
+-------+------------------+--------------------+
|branch |all_amounts |unique_accounts |
+-------+------------------+--------------------+
|Mumbai |[50000.0, 75000.0]|[SAV2001] |
|Delhi |[12000.0, 8000.0] |[SAV2002] |
|Chennai|[300000.0] |[CUR3001] |
+-------+------------------+--------------------+
How does Spark choose the aggregation strategy?
Spark's planner chooses automatically based on the buffer types of the aggregate functions.
| Condition | Strategy |
|---|---|
| All buffers are mutable fixed-length | HashAggregate |
| Buffers are supported object types (collect_list, collect_set) | ObjectHashAggregate |
| Otherwise | SortAggregate |
# HashAggregate: sum, count, avg, min, max
txn.groupBy("branch").agg(F.sum("amount")).explain()
# Shows: HashAggregate
# ObjectHashAggregate: collect_list, collect_set
txn.groupBy("branch").agg(F.collect_list("amount")).explain()
# Shows: ObjectHashAggregate
# SortAggregate: when hash path not available
# (rare in modern Spark unless explicitly disabled)
How do you identify which strategy is being used?
Check the physical plan with explain(). The plan shows HashAggregate, ObjectHashAggregate, or SortAggregate nodes.
# Two-phase aggregation pattern:
# Phase 1 (partial, before shuffle): HashAggregate with partial_sum
# Phase 2 (final, after shuffle): HashAggregate with sum
txn.groupBy("account_id").agg(
F.sum("amount").alias("total"),
F.max("amount").alias("max_txn")
).explain()
== Physical Plan ==
*(2) HashAggregate(keys=[account_id], functions=[sum(amount), max(amount)])
+- Exchange hashpartitioning(account_id, 200)
+- *(1) HashAggregate(keys=[account_id], functions=[partial_sum(amount), partial_max(amount)])
+- *(1) LocalTableScan
What are the performance implications?
HashAggregate is the fastest because it uses fixed-width Tungsten memory with no sorting. SortAggregate is slower because sorting is O(n log n). ObjectHashAggregate is between the two but uses JVM heap for object buffers, which adds GC pressure.
# Performance ranking:
# 1. HashAggregate – fastest (fixed-width, off-heap, no sort)
# 2. ObjectHashAggregate – medium (object heap, GC pressure)
# 3. SortAggregate – slowest (full sort before aggregation)
# To stay on the fast path:
# - Use sum, count, avg, min, max (not collect_list)
# - Avoid unnecessary collect_list/collect_set
# - If you must use collect_list, expect ObjectHashAggregate
What are common mistakes?
The most common mistake is using collect_list when a simpler aggregate would suffice, pushing the plan off the fast HashAggregate path.
# WRONG: collect_list just to count distinct values
bad = txn.groupBy("branch").agg(
F.size(F.collect_list("account_id")).alias("count") # ObjectHashAggregate
)
# CORRECT: use countDistinct (stays on HashAggregate)
good = txn.groupBy("branch").agg(
F.countDistinct("account_id").alias("count") # HashAggregate
)
bad.explain() # ObjectHashAggregate
good.explain() # HashAggregate
What should a senior engineer know?
A senior engineer checks the physical plan to confirm which aggregation strategy is being used. They know that high-cardinality grouping keys or object-accumulating aggregates push off the fast hash path.
# Senior tips:
# 1. Check explain() for HashAggregate vs ObjectHashAggregate vs SortAggregate
# 2. Prefer sum/count/avg/min/max over collect_list for performance
# 3. If collect_list is needed, understand it uses ObjectHashAggregate
# 4. High cardinality + collect_list = heavy GC pressure
# 5. The partial → final two-phase pattern is normal and expected
Visual Overview
Aggregation Strategies: 1. HashAggregate (fastest): ┌─────────────────────────────────┐ │ Off-heap hash map (Tungsten) │ │ Key → [sum, count, avg] │ │ Mumbai → [125000, 2, 62500] │ │ Delhi → [20000, 2, 10000] │ │ Fixed-length, mutable, in-place │ └─────────────────────────────────┘ No sorting required. Spills gracefully. 2. SortAggregate (slower): ┌──────────────────────────────────┐ │ Step 1: Sort by key │ │ Chennai, 300000 │ │ Delhi, 12000 │ │ Delhi, 8000 │ │ Mumbai, 50000 │ │ Mumbai, 75000 │ │ Step 2: Aggregate runs │ │ Chennai → sum=300000 │ │ Delhi → sum=20000 │ │ Mumbai → sum=125000 │ └──────────────────────────────────┘ O(n log n) sort, then linear scan. 3. ObjectHashAggregate (medium): ┌──────────────────────────────────┐ │ JVM heap hash map │ │ Key → [List object] │ │ Mumbai → [50000, 75000] │ │ Delhi → [12000, 8000] │ │ Growing object buffers, GC risk │ └──────────────────────────────────┘ Falls back to sort if map too large.
Key Takeaway
Spark has three physical aggregation strategies. HashAggregate is the fast default: it keeps an off-heap hash map from each grouping key to a buffer of mutable, fixed-length values, and for sum, count, min, max, and average it updates those fixed slots in place as rows stream by, with no sorting, spilling to disk gracefully if the map outgrows memory. SortAggregate sorts the rows by grouping key first, then scans the sorted stream aggregating runs of equal keys, which tolerates high cardinality and buffer types the hash path cannot handle. ObjectHashAggregate sits in between, for aggregates whose buffer is an arbitrary JVM object like collect_list or collect_set.
Spark's planner chooses automatically based on the buffer type. It picks HashAggregate when every aggregate's buffer is mutable and fixed-length, ObjectHashAggregate when the buffers are supported object types, and SortAggregate otherwise. You can see exactly which one was chosen in the physical plan with explain().
At the National Bank of India this guided how I reasoned about aggregation performance. Our total-spend-per-account job was a simple sum, so the plan showed HashAggregate and I knew it was on the fast path. A risk feature that collected every transaction amount per account with collect_list showed ObjectHashAggregate, which was inherently heavier.
The practical lever is that high-cardinality grouping keys or object-accumulating aggregates push you off the fast hash path, so reducing cardinality and avoiding unnecessary collect_list keeps aggregation efficient.
Remember one sentence: HashAggregate is the fast path for fixed-length aggregates like sum and count, ObjectHashAggregate handles collect_list with JVM object buffers, and SortAggregate is the fallback that sorts first.
---