Too Many Small Files

An easy Spark interview practice problem on DataDriven. Write the Spark code and run it against the tables, with instant feedback.

Domain

Spark

Difficulty

easy

Seniority

L5

Problem

A daily export pipeline reads 200 GB of transaction data, keeps only the flagged high-value records (transactions over $500), and writes them as Parquet to S3. Downstream Athena queries on the exported table take 45 seconds for a simple COUNT(*), and S3 holds 2,000 Parquet files averaging 1 MB each even though the job already sets spark.sql.shuffle.partitions to 2000. Read the transactions table, keep the flagged records, and fix the write so the output lands as a handful of right-sized files.

Two thousand files. One megabyte each. Athena says no.

Expected output

Worked solution and explanation

What this really is

This is a partition-count problem wearing an Athena costume. The count is fine, the data is tiny (2 GB), and nothing about the query is slow: what is slow is opening 2,000 objects in S3 to read 2 GB. The number of files Spark writes equals the number of partitions in the final stage, not whatever spark.sql.shuffle.partitions happens to be. The trap is reading spark.sql.shuffle.partitions=2000 in the config and 'fixing' it there. Lower it and nothing changes, because a filter is a narrow transformation and never shuffles, so that config is never consulted on this path. The 2,000 files are the 2,000 input read partitions carried straight through the filter to the write. And none of it licenses touching the content: the export is still every column of every transaction over $500, the 12 flagged rows with no product_id included, and a fix that tidies them away ships 116 records where the table should hold 128.

Why the obvious knob does nothing

spark.sql.shuffle.partitions only governs the post-shuffle partition count for wide transformations: joins, groupBy, distinct, a repartition by key. This job has none of those on the write path. Read the physical plan: the FileScan feeds a Filter on total_amount, and the filtered rows go straight to the write with the same partitioning. There is an Exchange in the plan only because of an incidental sort, and even so the fan-out to 2,000 tiny files is set by the partition count reaching the writer. Changing a config that the plan never reaches is why the naive fix below leaves you with 2,000 files.

# 'The config says 2000, so lower it.'
spark.conf.set("spark.sql.shuffle.partitions", "10")

flagged = spark.table("transactions").filter(col("total_amount") > 500)
flagged.write.mode("overwrite").parquet("s3://exports/flagged_transactions/")
# Still 2,000 files. The filter never shuffled, so shuffle.partitions was ignored.

The fix

Step 1: Size the output to the data

You have 2 GB after the filter. Athena and every Parquet-on-S3 engine want files in the 128 MB to 1 GB range so per-file open and footer-read overhead stays negligible. At ~200 MB per file that is 10 files, which is the target partition count going into the writer.

Step 2: Reduce partitions with coalesce, not repartition

You are only shrinking the partition count, not rebalancing keys, so use coalesce(10). coalesce is a narrow dependency: it merges adjacent partitions on the executors that already hold them, with no shuffle across the network. repartition(10) would get you 10 evenly sized files too, but it pays for a full shuffle of the 2 GB you did not need to move.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.getOrCreate()

# After filter: 2 GB of data. Target file size ~200 MB -> 10 files.
# coalesce(N) merges partitions without a full shuffle (narrow dependency).
# repartition(N) would trigger a full shuffle: unnecessary here since we
# are only reducing partition count, not redistributing keys.

flagged = (
    spark.table("transactions")
    .filter(col("total_amount") > 500)
    .coalesce(10)   # 2 GB / 10 = 200 MB per file, Athena-friendly
)

flagged.write \
    .mode("overwrite") \
    .parquet("s3://exports/flagged_transactions/")

coalesce(10)

Narrow dependency, no shuffle. Merges the existing 2,000 partitions into 10 on the executors that already hold them. Cheap. Can leave the 10 partitions uneven if the upstream partitions were skewed, which is harmless here because Athena reads every file anyway.

repartition(10)

Wide dependency, full shuffle of all 2 GB across the network to produce 10 evenly sized partitions. More expensive, but the right call when you need uniform file sizes, for example to support partition pruning or when one merged partition would be far larger than the rest.

Common follow-up questions

• The 2 GB is skewed so `coalesce(10)` leaves one 1.5 GB partition and nine tiny ones. What now? (Tests whether they know coalesce cannot rebalance and that `repartition(10)` (a shuffle) is the price of uniform file sizes.)
• The export is partitioned by `txn_date` with dozens of dates per run. How do you keep file sizes sane per date? (Probes `repartition('txn_date')` or `maxRecordsPerFile`, and understanding that one file per partition-directory can itself be too small or too large.)
• How would you retrofit this so the layout stays healthy without editing the job each time volumes change? (Looks for `maxRecordsPerFile`, a periodic compaction job, or Adaptive Query Execution coalescing shuffle partitions automatically.)