Why are there too many small files after a Spark write?

Summary

• Each Spark write task writes its own file or files, and a file cannot be larger than the task that wrote it, so many small tasks produce many small files.
• Common causes: a high shuffle partition count, adaptive coalescing that ignores the 64 MB advisory size by default, partitioned writes where each task holds many partition values, and frequent small appends or streaming micro-batches.
• Fix it at write time with coalesce, repartition or the REBALANCE hint, and fix existing files with Delta OPTIMIZE or Iceberg rewrite_data_files.
• Documented target sizes differ: Delta auto compaction uses 128 MB, Iceberg rewrites toward 512 MB, and Databricks autotunes 256 MB up to 1 GB by table size.
• On Databricks, predictive optimization runs OPTIMIZE on Unity Catalog managed tables.

Why are there too many small files after a Spark write?
Every Spark write task writes its own output, so the number of files follows how the data is partitioned when it reaches the writer. The Spark SQL hints guide says its partitioning hints "control the number of output files". The Iceberg docs add that Spark cannot write a file larger than a Spark task, and that a task writing to several partitions creates even smaller files.

What causes small files?

Cause	What the docs say
Many shuffle partitions	Delta's MERGE docs say spark.sql.shuffle.partitions (default 200) "determines the number of output files", and raising it generates a larger number of smaller files
Adaptive coalescing favors parallelism	spark.sql.adaptive.coalescePartitions.parallelismFirst defaults to true, so Spark ignores the 64 MB advisory size and respects only the 1 MB minimum partition size
Partitioned writes	Delta's MERGE docs say "every shuffle task can write multiple files in multiple partitions"
Frequent appends or streaming	Data written from a streaming process is "typically small", which "can cause the table metadata to track lots of small files" (Iceberg docs)

How do I write fewer, larger files?

# Narrow dependency: merges existing partitions without a shuffle
df.coalesce(10).write.mode("overwrite").parquet("/output")

# Adds a shuffle so upstream work still runs in parallel
df.repartition(10).write.mode("overwrite").parquet("/output")

# Partitioned output: group rows by the partition columns first
df.repartition("year", "month").write.partitionBy("year", "month").mode("append").parquet("/output")

The coalesce docs say going from 1000 to 100 partitions causes no shuffle, but a drastic coalesce such as one partition can leave the computation on fewer nodes, and repartition() avoids that at the cost of a shuffle. For SQL, the REBALANCE hint rebalances output partitions "so that every partition is of a reasonable size", is "useful when you need to write the result of this query to a table, to avoid too small/big files", and is ignored if AQE is off.
To see how many tasks will write, count the partitions that hold rows:

from pyspark.sql import functions as F

df.select(F.spark_partition_id()).distinct().count()

Each of those partitions is one write task, and a file cannot be larger than the task that writes it. You can also change adaptive coalescing:

spark.conf.set("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")

With this setting false, coalescing respects spark.sql.adaptive.advisoryPartitionSizeInBytes (64 MB), which the docs recommend on busy clusters to avoid many small tasks. Iceberg's docs warn that the advisory size measures shuffle data, while files on disk are compressed and columnar, so a file ends up smaller than its task. Pick the partition count by measuring output files, not by dividing data size by a target.

How do I compact files that already exist?

Delta Lake (compaction since 1.2.0, Python and Scala APIs since 2.0) and Iceberg each have a command:

OPTIMIZE delta.`/path/to/delta/table`;
OPTIMIZE delta_table_name WHERE date >= '2017-01-01';

CALL catalog_name.system.rewrite_data_files('db.sample');

from delta.tables import *
deltaTable = DeltaTable.forPath(spark, pathToTable)
deltaTable.optimize().executeCompaction()

Delta Lake 3.1.0 and above can also reduce small files automatically. Auto compaction (delta.autoOptimize.autoCompact) runs after a successful write, synchronously on the cluster that performed it, and compacts only files that were not compacted before. Optimized writes (delta.autoOptimize.optimizeWrite) are off by default and work best for partitioned tables; the docs say they remove the need for coalesce(n) or repartition(n) before a write.
Delta's bin-packing is idempotent, so a second run on the same data has no effect. Iceberg's rewrite_data_files uses a bin-pack strategy by default and accepts a where filter. For a plain Parquet directory without a table format, compaction means rewriting the data with Spark using the techniques above.

What file size should I aim for?

The docs give per-engine values rather than one ideal size:

Where	Documented value
Spark file reads	spark.sql.files.maxPartitionBytes packs up to 128 MB into one partition
Delta auto compaction (Delta Lake 3.1.0 and above)	128 MB target when set to true
Iceberg rewrite_data_files	target-file-size-bytes, default 512 MB
Databricks autotuning	256 MB below 2.56 TB, growing to 1 GB at 10 TB and above

What does Databricks do about small files?

Databricks automatically tunes file sizes for Unity Catalog managed tables, and predictive optimization automatically runs OPTIMIZE, VACUUM and ANALYZE on them. Auto compaction and optimized writes reduce small files but do not fully replace OPTIMIZE, and for tables over 1 TB Databricks recommends scheduling it. If you use optimized writes, Databricks recommends not running coalesce(n) or repartition(n) just before the write. Databricks also recommends liquid clustering over partitions or ZORDER.

FAQs

Do many small files cost money?

Amazon S3 charges for LIST requests at the same rate as PUT, COPY and POST requests. The Spark tuning guide adds that listing a job's input can take a very long time against object stores like S3 when there are many directories.

Can I compact only part of a table?

Yes. Delta accepts a partition predicate with WHERE, and Iceberg's rewrite_data_files accepts where, which selects every file that may contain matching data.

Does maxRecordsPerFile reduce small files?

No. spark.sql.files.maxRecordsPerFile caps the records written to one file (the default of 0 means no limit), so a low value splits output into more files, not fewer.