RDD - Transformation ReduceByKey

The reduceByKey() transformation is specialized for Key-Value Pair RDDs where each element is a tuple (key, value). It aggregates the values of each key using an associative and commutative binary operator (such as addition or maximum).

Performance Optimization: Map-Side Combine

Unlike standard grouping operations (groupByKey()), reduceByKey() is highly optimized because it performs a Map-side combine (also known as local aggregation) inside each partition before shuffling data across the network:

1. Local Combine: Spark aggregates values matching the same key within each partition locally on the executor node.
2. Shuffle: Spark only transfers the locally aggregated totals over the network.
3. Final Reduce: Spark merges the partition-level aggregates to yield the final global result.

This minimizes network traffic, reducing shuffle sizes by up to 90% and speeding up distributed jobs significantly!

graph TD
    subgraph Partitions["Local Executors (Partition-level Aggregations)"]
        direction LR
        P1["Partition 1:<br>(Book, 20)<br>(Book, 45)"] -->|Local Combine| LC1["Local Total:<br>(Book, 65)"]
        P2["Partition 2:<br>(Book, 15)<br>(Book, 30)"] -->|Local Combine| LC2["Local Total:<br>(Book, 45)"]
    end

    subgraph Network["Network Shuffle & Reduce"]
        LC1 & LC2 -->|Network Shuffle| FR["Final Reducer"]
        FR -->|Global Merge| Out["(Book, 110)"]
    end

    style Partitions fill:#e8f5e9,stroke:#2e7d32,stroke-width:2px;
    style Network fill:#fff3e0,stroke:#e65100,stroke-width:2px;

PySpark Code Examples

Setup Spark Session

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("RDD Transformation ReduceByKey") \
    .master("local[*]") \
    .getOrCreate()

sc = spark.sparkContext

Example A: Basic Word Count

Let's count the frequency of words in an input list:

# 1. RDD of key-value tuples representing (Word, CountMultiplier)
word_pairs = sc.parallelize([
    ("spark", 1), ("rdd", 1), ("spark", 1), 
    ("pyspark", 1), ("rdd", 1), ("spark", 1)
])

# 2. Reduce values by key (sum counts)
word_counts = word_pairs.reduceByKey(lambda x, y: x + y)

# 3. View final results
print("Word Counts:")
print(word_counts.collect())
# Output: [('spark', 3), ('rdd', 2), ('pyspark', 1)]

Example B: Calculating Sum Total Sales per Department

Let's find the total revenue earned by each store category:

# 1. Input sales tuples: (StoreCategory, RevenueAmount)
sales = sc.parallelize([
    ("Electronics", 500),
    ("Books", 15),
    ("Electronics", 120),
    ("Clothing", 45),
    ("Books", 30),
    ("Clothing", 110)
], numSlices=2)

# 2. Aggregate sales amount by category keys
category_revenue = sales.reduceByKey(lambda x, y: x + y)

print("Revenue by Category:")
print(category_revenue.collect())
# Output: [('Electronics', 620), ('Books', 45), ('Clothing', 155)]

Example C: Finding Maximum Latency per API Path

Let's parse system logs and find the worst response latency (maximum value) for each individual API endpoint:

# 1. Input: (ApiPath, ResponseLatencyMS)
api_logs = sc.parallelize([
    ("/login", 120),
    ("/checkout", 850),
    ("/login", 410),
    ("/products", 300),
    ("/checkout", 990),
    ("/products", 150)
])

# 2. Find max latency by key
worst_latency = api_logs.reduceByKey(lambda latency1, latency2: max(latency1, latency2))

print("Worst Latencies:")
print(worst_latency.collect())
# Output: [('/login', 410), ('/checkout', 990), ('/products', 300)]