A Spark job writes 40,000 tiny Parquet files a day — what caused it and how do you fix it?

## The arithmetic behind the file count Spark's file writer is simple and its output count is predictable. Each task writes to the output path, and when the write is partitioned by columns it opens a separate file for each distinct combination of those column values it encounters. So: ``` files ≈ (number of in-memory partitions) × (distinct partitionBy value combinations seen per task) ``` A job whose last shuffle produced the default 200 partitions, writing with `partitionBy("dt", "country")` where the batch spans 200 date/country combinations, produces up to 200 × 200 = 40,000 files — each a few hundred kilobytes. Nothing has gone wrong internally; the layout is exactly what was asked for. ## Why it hurts - **Read planning.** Every subsequent Spark job that reads this dataset must list and open all those files. Listing dominates on object stores, where each `LIST` and `HEAD` is a network round trip; the metadata phase can take longer than the scan. - **Wasted parallelism on read.** Spark packs small files together toward `spark.sql.files.maxPartitionBytes` and charges each file an opening cost via `spark.sql.files.openCostInBytes`, so tiny files raise per-task overhead relative to real work. - **Lost compression and statistics.** Parquet's dictionary encoding, run-length encoding and row-group min/max statistics all work better over more rows. A 200 KB Parquet file carries proportionally more footer and schema overhead and gives predicate pushdown almost nothing to prune with. - **Storage-layer cost.** On HDFS every file and block consumes NameNode heap; on object storage you pay per request. Either way, small files are a tax paid by every reader forever. ## Diagnosis Count files per directory (`hdfs dfs -count` or an object-store listing) and check the mean size — anything well under ~100 MB for Parquet is a candidate. Then look at the job: what was the partition count going into the write (visible as the write stage's task count in the Spark UI), and how many distinct `partitionBy` values did the batch contain? Multiplying the two should reproduce the observed file count, which confirms the mechanism rather than guessing at it. ## The primary fix: align in-memory partitioning with write partitioning ```python (df.repartition("dt", "country") .write.partitionBy("dt", "country") .mode("overwrite") .parquet("s3://lake/events/")) ``` Hash-partitioning by the same columns puts all rows for a `(dt, country)` pair into one task, so that task writes exactly one file for that directory. File count drops from 40,000 to 200. This is the standard idiom and the answer an interviewer is listening for. Two refinements: - If one directory is much larger than the rest, a single file becomes unwieldy. Use `.option("maxRecordsPerFile", 5000000)` (or the session-level `spark.sql.files.maxRecordsPerFile`) to cap file size — Spark rolls to a new file when the cap is hit. - If every directory holds several hundred MB, use `repartition(400, "dt", "country")` so each directory gets a handful of properly sized files rather than one enormous one. Be aware that `repartition` by the partition columns concentrates each group in one task, so a skewed group produces a slow task. That is a deliberate trade: you are buying good file layout with some write-stage imbalance. ## What not to do `coalesce(1)` before the write is the classic wrong fix. `coalesce` is a narrow transformation with no stage boundary, so the single-partition limit propagates up the entire fused stage — the scan, the filters and the projections all run in one task. You will trade 40,000 small files for a job that takes hours or dies on memory. ## Prevention and the layout question If the `partitionBy` columns are high-cardinality — `user_id`, an hour-plus-minute timestamp, an event UUID — no amount of repartitioning saves you, because the directory count itself is the problem. Partition on the coarsest column your query predicates actually filter on, usually a date, and use clustering (`repartitionByRange` on a secondary column, or a table format's clustering feature) for finer pruning rather than more directories. For a streaming job the same arithmetic applies per micro-batch, and there the answer is usually a periodic **compaction** job that rewrites yesterday's directories into properly sized files, plus a longer trigger interval so each batch writes more. Adaptive Query Execution helps indirectly: with `spark.sql.adaptive.enabled` on (the default since Spark 3.2) and its coalescing feature, an over-provisioned post-shuffle stage may already be merged down before the write, so the multiplier is smaller than the configured 200. It is a mitigation, not a substitute for aligning the write partitioning.