RDD - Transformation GroupByKey
The groupByKey() transformation is specialized for Key-Value Pair RDDs where each element is a tuple (key, value). It groups all values associated with the same key across the entire dataset into a single sequence iterator.
️ Performance Warning: High Shuffle Cost
Unlike reduceByKey(), groupByKey() does not perform a Map-side combine. Spark cannot aggregate or collapse values locally within partitions before transferring data.
- The Overhead: Every single key-value record in the entire RDD is serialized, written to disk, and transferred over the network (shuffled).
- Memory Pressure: Since all values for a key must be held in memory within a single executor's list, keys with huge numbers of values (e.g. tracking a massive "Active" status key) can easily cause spill-to-disk slowdowns or Out Of Memory (OOM) crashes.
graph TD
subgraph NoLocalCombine["groupByKey (No Local Combining - Slow & Heavy Network Overhead)"]
direction LR
P1["Partition 1:<br>(A, 1), (A, 2)"] -->|Shuffle ALL rows| N["Network Shuffle"]
P2["Partition 2:<br>(A, 3), (A, 4)"] -->|Shuffle ALL rows| N
N --> FR["Final Reducer:<br>(A, [1, 2, 3, 4])"]
end
style NoLocalCombine fill:#ffebee,stroke:#c62828,stroke-width:2px;
PySpark Code Examples
Setup Spark Session
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("RDD Transformation GroupByKey") \
.master("local[*]") \
.getOrCreate()
sc = spark.sparkContext
Example A: Basic Grouping
Let's group items belonging to the same product category:
# 1. RDD of key-value tuples representing (Category, ItemName)
inventory = sc.parallelize([
("Fruit", "Apple"),
("Veggie", "Carrot"),
("Fruit", "Banana"),
("Veggie", "Broccoli"),
("Fruit", "Cherry")
])
# 2. Group items by key (returns a ResultIterable)
grouped_inventory = inventory.groupByKey()
# 3. View the ResultIterable by converting values to a Python list
# mapValues applies a function only to the value portion (preserving keys!)
list_inventory = grouped_inventory.mapValues(list)
print("Grouped Inventory:")
print(list_inventory.collect())
# Output: [('Fruit', ['Apple', 'Banana', 'Cherry']), ('Veggie', ['Carrot', 'Broccoli'])]
Example B: Grouping User Transactions
Let's collect a list of all purchase amounts made by each user:
# 1. Input: (UserID, PurchaseAmount)
user_purchases = sc.parallelize([
("UserA", 120),
("UserB", 45),
("UserA", 15),
("UserC", 300),
("UserB", 10)
])
# 2. Group values by user keys
purchases_by_user = user_purchases.groupByKey().mapValues(list)
print("Transactions per User:")
print(purchases_by_user.collect())
# Output: [('UserA', [120, 15]), ('UserB', [45, 10]), ('UserC', [300])]
Important
Best Practice Rule: Avoid using groupByKey() if your goal is to perform aggregations (such as sum, average, count, min, or max). Always choose reduceByKey() or aggregateByKey() instead, as they aggregate data locally before shuffling!