RDD - Repartition & Coalesce

In Apache Spark, partitions are the basic unit of parallel processing. The speed and efficiency of your cluster depends directly on how your data is distributed across these partitions.

Sometimes, you need to adjust the number of partitions of an existing RDDeither to increase parallelism or to reduce storage overhead before saving. Spark offers two primary methods for this: repartition() and coalesce().

This guide details the mechanical and performance differences between these two methods, accompanied by illustrative PySpark examples.

The Key Mechanical Differences

graph TD
    subgraph RepartitionFlow["repartition() - Full Network Shuffle"]
        direction LR
        P1["Part 1"] --> S1["Shuffle Manager"]
        P2["Part 2"] --> S1
        S1 --> NP1["New Part 1 (Balanced)"]
        S1 --> NP2["New Part 2 (Balanced)"]
        S1 --> NP3["New Part 3 (Balanced)"]
    end

    subgraph CoalesceFlow["coalesce() - Local Merging (No Shuffle)"]
        direction LR
        PP1["Part 1"] --> CP1["New Part 1"]
        PP2["Part 2"] --> CP1
        PP3["Part 3"] --> CP2["New Part 2"]
    end

    style RepartitionFlow fill:#ffebee,stroke:#c62828,stroke-width:2px;
    style CoalesceFlow fill:#e8f5e9,stroke:#2e7d32,stroke-width:2px;

Setting Up Spark Session (For Code Examples)

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Day01 Partitioning") \
    .master("local[*]") \
    .getOrCreate()

sc = spark.sparkContext

1. Using repartition(numPartitions)

Use repartition() when you want to increase the number of partitions to scale up your cluster parallelization, or when your data is highly skewed (some partitions are massive, others are empty) and you want to force an even distribution.

Code Example:

# 1. Create a small RDD with 2 partitions
raw_rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8], numSlices=2)
print("Initial Partitions:", raw_rdd.getNumPartitions()) # 2
print("Initial Layout:", raw_rdd.glom().collect())
# Output: Initial Layout: [[1, 2, 3, 4], [5, 6, 7, 8]]

# 2. Increase partitions to 4 using repartition()
# This triggers a complete network shuffle to distribute elements evenly
repartitioned_rdd = raw_rdd.repartition(4)

print("
--- After Repartition ---")
print("New Partitions count:", repartitioned_rdd.getNumPartitions()) # 4
print("Repartitioned Layout:", repartitioned_rdd.glom().collect())

# Typical Output:
# Repartitioned Layout: [[1, 5], [2, 6], [3, 7], [4, 8]]
# Note: Data has been completely reshuffled into equal-sized blocks!

2. Using coalesce(numPartitions)

Use coalesce() when you want to decrease the number of partitions. Instead of shuffling data across the network, coalesce keeps data on the worker node local disks and simply groups adjacent partitions together.

• Primary Use Case: When writing results to a database or filesystem. If your RDD has 200 partitions, saving it will generate 200 small part-* files, creating heavy HDFS metadata bloat. Coalescing to 1 or 4 partitions before saving ensures a small, clean file list.

Code Example:

# 1. Create a small RDD with 4 partitions
large_partition_rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8], numSlices=4)
print("Initial Partitions:", large_partition_rdd.getNumPartitions()) # 4
print("Initial Layout:", large_partition_rdd.glom().collect())
# Output: Initial Layout: [[1, 2], [3, 4], [5, 6], [7, 8]]

# 2. Decrease partitions to 2 using coalesce()
# Spark merges local partitions (e.g. 0 & 1, 2 & 3) without network shuffle
coalesced_rdd = large_partition_rdd.coalesce(2)

print("
--- After Coalesce ---")
print("New Partitions count:", coalesced_rdd.getNumPartitions()) # 2
print("Coalesced Layout:", coalesced_rdd.glom().collect())

# Typical Output:
# Coalesced Layout: [[1, 2, 3, 4], [5, 6, 7, 8]]
# Note: Elements from original partitions are combined locally without global network transfers!

3. What happens if you try to coalesce to a larger partition count?

By design, coalesce is built to avoid network shuffles. Therefore, it cannot increase the number of partitions. If you call .coalesce(N) where N is larger than the current partition count, Spark will simply ignore the call and retain your existing partition configuration.

rdd = sc.parallelize([1, 2, 3], numSlices=2)

# Attempt to increase partitions to 4 using coalesce
failed_coalesce_rdd = rdd.coalesce(4)

print("Attempted Coalesce Partitions:", failed_coalesce_rdd.getNumPartitions()) 
# Output: Attempted Coalesce Partitions: 2  (No change occurred!)