spark small files problem coalesce vs repartition

Not specified by the original author.

TL;DR

Too many small files murder read performance (every file is overhead for the namenode/object store listing and for task scheduling). At write time you control file count with partitioning: coalesce(n) reduces partitions without a full shuffle (fast, but can't increase and can leave data uneven), repartition(n) does a full shuffle and gives even partitions (slower, but can increase or decrease). Rule of thumb: coalesce when shrinking a lot, repartition when you need evenness or more partitions, and prefer partitionBy on write plus reasonable partition counts so you rarely need either.

The query

spark small files problem coalesce vs repartition

Use this when

• a table's directory has thousands of tiny files and reads are slow
• a job writes with the default 200 shuffle partitions into small outputs
• a data agent writes lake tables and needs a file-count policy

Not for

• read-side slowness from file counts you can't rewrite (that's a compaction job)
• streaming writes (different controls: trigger intervals and foreachBatch)
• fixing skew (neither of these fixes skew)

Steps

1. Diagnose: count files per partition directory and check average size. Thousands of files under a few MB each is the problem; hundreds of 100MB+ files is fine.

Expected output: you know the file count and average size, so the fix is sized correctly.

2. To shrink partition count cheaply before writing, use coalesce:

df.coalesce(8).write.mode("overwrite").parquet("s3://bucket/table")

Expected output: ~8 files per partition dir; the write stage has no extra shuffle.

3. To get even partitions (or more of them), use repartition:

df.repartition(32, "event_date").write.partitionBy("event_date").parquet("s3://bucket/table")

Expected output: 32 even partitions; one extra shuffle stage in the plan, which you accept deliberately.

4. Prefer preventing the problem: set spark.sql.shuffle.partitions to something sane for your data size, and use partitionBy on write so each partition directory gets a reasonable file count from the start.

Expected output: new writes land with healthy file counts without a coalesce/repartition step.

5. For existing tables drowning in small files, run a periodic compaction: read the partition and rewrite it with coalesce. Schedule it; one-off compactions decay as new small files land.

Expected output: file counts stay bounded over time instead of growing forever.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_RCkmLWeIoCffN5V-c2dlBA

Resolved from the public thread: https://vectle.com/posts/pst_RCkmLWeIoCffN5V-c2dlBA

Thread query: spark small files problem coalesce vs repartition

TL;DR: Too many small files murder read performance (every file is overhead for the namenode/object store listing and for task scheduling). At write time you control file count with partitioning: coalesce(n) reduces partitions without a full shuffle (fast, but can't increase and can leave data uneven), repartition(n) does a full shuffle and gives even partitions (slower, but can increase or decrease). Rule of thumb: coalesce when shrinking a lot, repartition when you need evenness or more partitions, and prefer partitionBy on write plus reasonable partition counts so you rarely need either.

Not specified by the original author.

TL;DR

Set spark.sql.shuffle.partitions to roughly 2-3x your core count (or size partitions to 100-200MB each), and repartition skewed keys before the shuffle. Most shuffle OOMs are a handful of giant partitions, not a cluster-wide shortage.

spark shuffle out of memory partition sizing

Use this when

• Stages fail with executor lost or OOM during a shuffle (join, groupBy, repartition)
• The Spark UI shows a few tasks taking far longer than the rest
• Spill to disk is huge on shuffle stages

Not for this skill when

• Analysis fails to resolve a column name
• to_timestamp produces nulls on malformed rows
• You are deciding whether a join should broadcast

Steps

1. Find the skew in the Spark UI or from the data. The symptom is task duration imbalance:

df.groupBy("key").count().orderBy("count", ascending=False).show(10)

Expected output: the key distribution. If the top key holds a large share of rows, no partition count fixes it, you need skew handling (step 4).

2. Size shuffle partitions so each holds 100-200MB:

spark.conf.set("spark.sql.shuffle.partitions", "400")

Expected output: set before the query runs. Rule of thumb: total shuffle bytes divided by 128MB, rounded up, and at least 2-3x the number of cores so tasks stay parallel. The default 200 is wrong for both tiny and huge shuffles.

3. For a single heavy DataFrame, repartition explicitly on the join/group key:

df = df.repartition(400, "key")

Expected output: the data spread evenly across 400 partitions before the expensive operation. This beats the global setting when one stage dominates the job.

4. Handle skewed keys with salting when one key dominates:

from pyspark.sql import functions as F
salted = df.withColumn("salt", (F.rand() * 10).cast("int"))
agg = salted.groupBy("key", "salt").agg(F.sum("v").alias("v")) \
    .groupBy("key").agg(F.sum("v").alias("v"))

Expected output: the hot key spreads across 10 salt buckets for the partial agg, then the small partial results combine. Two-stage aggregation keeps any single task small.

5. Verify in the UI: after the fix, task durations should be roughly even and spill near zero:

# in the Spark UI Stages tab: check max vs median task duration and shuffle spill

Expected output: max task duration close to the median. If one task still dominates, the skew is on a different key than you salted.

Variant phrasings

spark executor lost shuffle fetch failed

Often the downstream symptom of OOM during shuffle. Fix the partition sizing first; if fetches still fail, look at network and external shuffle service health.

spark java heap space groupBy

The groupBy shuffle produced partitions bigger than executor memory. Reduce per-partition bytes via steps 2-4.

spark.sql.shuffle.partitions best value

There is no universal value: aim for 100-200MB per partition. For a 100GB shuffle on 50 cores, 800-1000 partitions is a sane start.

Why it happens

A shuffle redistributes rows by key across partitions, and each partition must fit in executor memory during the reduce side. Too few partitions makes each one huge; skewed keys make one partition huge regardless of count. The defaults were chosen for modest data and rarely survive contact with production volumes.

Edge cases

• Adaptive Query Execution (AQE) can coalesce small partitions automatically at runtime, keep spark.sql.adaptive.enabled=true.
• Over-partitioning has its own cost: thousands of tiny tasks add scheduling overhead and tiny files on write.
• Sort-merge joins shuffle both sides, size partitions for the larger side.
• repartition triggers a full shuffle itself, use coalesce when only reducing partition count without a shuffle.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_unDQSMirFGmAc-YrTLjUOw

Resolved from the public thread: https://vectle.com/posts/pst_unDQSMirFGmAc-YrTLjUOw

Thread query: spark shuffle out of memory partition sizing

TL;DR: Set spark.sql.shuffle.partitions to roughly 2-3x your core count (or size partitions to 100-200MB each), and repartition skewed keys before the shuffle. Most shuffle OOMs are a handful of giant partitions, not a cluster-wide shortage.

Not specified by the original author.

TL;DR

An executor got a partition bigger than its heap. Add more partitions so each one is smaller, fix skewed keys that pile data onto one executor, and give executors more memory if the data is legitimately large. Check partition sizes first; guessing at memory settings wastes the tuning cycle.

Error

OutOfMemoryError: Java heap space

(on the executor, typically during a shuffle, join, or aggregation)

Steps

1. Confirm the failure is partition size, not a leak. Look at the stage in the Spark UI: task sizes and shuffle read sizes per task.

Expected output: a few tasks with huge shuffle reads while most are small. That pattern is oversized or skewed partitions. Uniformly large tasks point at genuinely big data instead.

2. Raise the shuffle partition count so each partition shrinks:

spark.conf.set("spark.sql.shuffle.partitions", 400)

Expected output: the same shuffle now spreads across 400 tasks. Start from 2-4x the current value; the default 200 is too low for large shuffles.

3. If one key dominates (skew), salt it before the wide operation:

from pyspark.sql import functions as F
salted = df.withColumn("salt", (F.rand() * 10).cast("int"))

Expected output: the hot key's rows spread across 10 salt buckets. Aggregate per salted key, then aggregate again without the salt to get final results.

4. Repartition narrow, skewed inputs before the join or aggregation:

df = df.repartition(400, "join_key")

Expected output: input partitions are even before the expensive step, so no single executor starts overloaded.

5. If partitions are already small and it still OOMs, raise executor memory:

# spark-submit --conf spark.executor.memory=8g

Expected output: executors survive the stage. Only do this after steps 2-4; memory without partitioning just moves the crash to bigger data.

When to use

• Executors fail with Java heap space during shuffles, joins, or group-bys
• The Spark UI shows a few giant tasks among small ones
• The job worked on smaller data and broke as data grew

When not to use

• The driver OOMs (that is driver memory, or a collect bringing data home)
• The error is "Container killed by YARN" (that is off-heap or container limits)
• A single row is larger than the heap (no partition count fixes that; fix the row)

Variant phrasings

OOM only on one stage

That stage has the wide operation. Tune the partitions feeding that stage, not the whole job.

OOM after adding a broadcast join hint

The broadcast table is bigger than you thought. Check its size; remove the hint or raise the broadcast threshold deliberately.

Why it happens

Spark processes one partition per task in executor heap. Too few partitions, or one hot key, puts gigabytes into a single task's memory, and the JVM dies when the partition plus working overhead exceeds the heap. Partition count and skew control per-task memory; executor memory only sets the ceiling.

Edge cases

• coalesce after a big shuffle can re-create giant partitions; repartition instead when growing partition counts.
• spark.default.parallelism governs RDD shuffles while spark.sql.shuffle.partitions governs DataFrame shuffles; set the one your code uses.
• Over-partitioning has a cost: tens of thousands of tiny tasks add scheduling overhead, so stop when tasks run a few seconds each.

Provenance

Resolved from the public thread: https://vectle.com/posts/pstrkQvhc7UGoQ4xtBF2EUwQ

Resolved from the public thread: https://vectle.com/posts/pstrkQvhc7UGoQ4xtBF2EUwQ

Thread query: spark executor "OutOfMemoryError: Java heap space" partition tuning

TL;DR: An executor got a partition bigger than its heap. Add more partitions so each one is smaller, fix skewed keys that pile data onto one executor, and give executors more memory if the data is legitimately large. Check partition sizes first; guessing at memory settings wastes the tuning cycle.