HDFS Small Files Problem — Why Millions of Tiny Files Kill NameNodes

Why it matters

The small files problem is silent until it is catastrophic. A pipeline that produces one file per minute per source runs happily for months, until the day the NameNode heap crosses its allocation and garbage collection latency spikes. Then every RPC gets slower, then RPCs start timing out, and finally the NameNode restarts under memory pressure and comes back to the same problem in a few days.

Preventing small files is far cheaper than cleaning them up. A single line of pipeline code that batches output before writing is worth a hundred emergency ops calls later.

The architecture

Every file in HDFS consumes one inode entry in the NameNode's block map, plus one entry per block. For a small file with a single block, this is 300 bytes of NameNode heap. Multiply by tens of millions of files and the NameNode heap becomes dominated by metadata for files that hold almost no data.

DataNodes handle small files less badly because the storage is not the bottleneck; the block map on the NameNode is. But small files still hurt: block replication traffic for a million tiny blocks is dominated by protocol overhead, and disk seeks per byte read are much worse for many small files than for fewer large ones.

How it works end to end

The classic consolidation technique is Hadoop Archive (HAR). A HAR file is a single HDFS file that contains many logical files inside it, with an internal index. From the NameNode perspective it is one file with a few blocks. From the client perspective (using the har:// scheme) it is a virtual directory of files. HAR is simple and non-invasive; existing readers can be pointed at the archive with almost no code change.

SequenceFile is another approach: a binary key-value container where each key is a filename and each value is the file contents. This is friendlier to MapReduce and Spark because splits fall naturally on record boundaries. Avro and Parquet extend this idea with schema and columnar layout, and are the modern preferred choice for most data pipelines.

HBase turns the small file problem into a totally different pattern: HBase writes to memstores, flushes to HFiles, and periodically compacts HFiles into larger ones. Millions of small rows become a handful of HFiles per region. If your access pattern is by key, HBase can be the right home for what would otherwise be small files.

The constraint is object count, not bytes

The NameNode keeps the entire namespace resident in RAM, and two in-memory structures carry the weight. The namespace tree holds one inode object per file and one per directory, carrying the name, owner, group, permission bits, timestamps, replication factor and the ordered list of blocks the file owns. The block map holds one block object per block, keyed by block ID, carrying the block length, the generation stamp and an array of the DataNode storages currently holding a replica.

Nothing in either structure scales with the number of bytes in the file. A 4 KB file and a 128 MB file cost exactly one inode and one block object each. That is the whole problem in a sentence: HDFS charges you per namespace object, and a tiny file buys the same metadata as a full block for a rounding error of data. The petabyte ceiling on a NameNode is really a ceiling on how many things it is tracking, and you can hit it with a few terabytes if those terabytes arrive as a hundred million files.

Directories are not free either. An empty directory is a full inode. A pipeline that lays out /events/src=NNN/dt=YYYY-MM-DD/hr=HH/ across 200 sources creates roughly 1.75 million directory objects a year before it has written a single file into them.

Replicas are cheaper than people expect. The replica list lives inside the single block object, so replication factor 3 does not triple NameNode heap. It does triple block report volume, which is a separate and very real cost covered below. Erasure coding pushes the other way and makes small-file metadata strictly worse, because a small EC file becomes one block group of k+m internal blocks where a replicated file was one block - the arithmetic is worked through in HDFS erasure coding, and the practical rule from it is compact first, then code.

The arithmetic - turning a file count into a heap number

Object count is a formula you can run on a whiteboard:

objects = files + directories + blocks
blocks  = sum over files of ceil(file_size / dfs.blocksize)   # never less than 1

The commonly cited planning figure is roughly 150 bytes of NameNode heap per namespace object. Treat that as an order-of-magnitude approximation rather than a measurement: the true per-object cost moves with the JVM (compressed ordinary object pointers on or off), object header size, path-component length, ACLs and extended attributes, snapshot diffs, and how many storages sit in each block's replica array. Published estimates range from about that number to several times it. It is good enough to decide whether you need 8 GB or 80 GB of heap, and not good enough to justify a capacity plan to four significant figures - measure your own cluster with the offline image viewer and a heap histogram. The 300-byte figure quoted earlier in this article is the same approximation applied to a single-block file, which costs two objects: one inode plus one block.

Hold the data volume constant at 200 TB and vary only the file size, and the cost of geometry becomes obvious:

Same bytes, same replication, same disks - a 60x swing in the one resource that cannot be scaled horizontally. And heap size is not even the hard wall. A 60 GB heap densely packed with tiny long-lived objects is a garbage collector's worst case: full-GC pauses stretch into seconds, and a multi-second pause is exactly what makes a ZooKeeper session expire and trips an unwanted failover. Long before OutOfMemoryError, the cluster becomes unavailable in bursts. See HDFS High Availability for how that interacts with the failover machinery.

Block reports - the second-order cost

Every DataNode periodically sends a full block report per block pool: the complete list of block IDs and generation stamps for every replica it stores. The default interval is six hours (dfs.blockreport.intervalMsec, 21600000 ms). Between full reports it sends incremental reports as blocks are created and deleted.

The size of that report scales with replica count, not with bytes. A DataNode holding 100 TB in 256 MB blocks reports roughly 400,000 replicas. The same 100 TB arriving as 1 MB files is 100 million replicas - a report 250 times larger to build, serialize, ship and process. Multiply by a thousand DataNodes.

The failure mode is not bandwidth, it is lock contention. The NameNode holds the namespace write lock while it reconciles a report against its block map, so a report that takes seconds to process is seconds during which no client RPC makes progress. On a small-file cluster this shows up as a periodic latency wave every six hours, and at the bad end as RPC queue overflow, client timeouts, and DataNodes marked stale because their heartbeats could not be serviced.

Two settings shorten the lock holds without reducing the underlying replica count. dfs.blockreport.split.threshold makes a DataNode send one RPC per storage directory instead of one enormous RPC, so the NameNode processes the report in digestible pieces. dfs.blockreport.initialDelay staggers the first report after a restart with a random offset, so a thousand DataNodes do not all report in the same second. Both are worth setting on any large cluster, and neither is a fix - they smooth a load that should not exist.

Startup and failover - the recovery time nobody measured

Three phases of NameNode startup stretch with object count, and only the first is obvious.

fsimage load. The image is a serialized snapshot of the namespace tree, so both its on-disk size and its parse time scale with the number of objects. Tens of millions of objects is minutes of pure deserialization before the process is useful for anything. Edit log replay then applies everything since the last checkpoint, and small-file pipelines are transaction factories - one create, one or more block allocations and one close per file. The checkpoint mechanics that bound this are covered in HDFS NameNode checkpointing; the small-file contribution is simply that the transaction rate tracks file count, not data volume. Safe mode is the last phase: the NameNode refuses to leave read-only state until the fraction of expected blocks that have been reported crosses dfs.namenode.safemode.threshold-pct, which means waiting for a full block report from essentially every DataNode - see HDFS Safe Mode.

A cluster with a few million objects restarts in a couple of minutes. One carrying 400 million can spend the better part of an hour in image load plus block-report intake. That number is your worst-case recovery time, and on a small-file cluster it is almost always larger than the number written in the DR plan, because the plan was written when the namespace was a tenth the size. Measure it on the standby during a maintenance window rather than discovering it during an incident.

HA does not remove this cost so much as pre-pay it: the standby already holds the namespace and the block map, so a failover skips the reload entirely. But the standby's heap and GC profile are identical to the active's, so an object count that is too large for the heap does not have a healthy failover target - it has a second machine with the same disease. The failover time budget itself is broken down in the HA article; what small files add to it is GC pause length.

Why jobs get slow - one task per split

The metadata story is only half the damage. The other half lands on every job that reads the data.

A split never spans a file. The standard input format produces at least one split per file, so 200,000 tiny files produce at least 200,000 map tasks. Raising mapreduce.input.fileinputformat.split.minsize merges nothing here, because the knob merges within a file and there is nothing inside a 1 MB file to merge.

Launch overhead dominates. Container allocation, JVM start, classloading and framework initialisation cost on the order of a second per task even on a warm cluster. If the task then reads 1 MB and exits, well over 90 percent of its wall clock is overhead. Two hundred thousand tasks at one second each is roughly 55 hours of cluster CPU spent doing nothing but starting JVMs - and the scheduler is saturated with task churn the entire time, which slows every other job on the cluster too.

Sequential I/O disappears. Reading a 256 MB block is a long streaming scan the disk and the OS readahead are built for. Reading a 1 MB file is a getBlockLocations RPC to the already-overloaded NameNode, a TCP connection to a DataNode, a seek, a short read and a close. Per-byte throughput collapses, and the read path pounds the exact component that is already the bottleneck.

Frameworks offer painkillers. CombineFileInputFormat with mapreduce.input.fileinputformat.split.maxsize packs multiple files into one split in MapReduce. Spark does the same through spark.sql.files.maxPartitionBytes together with spark.sql.files.openCostInBytes, which models each file's fixed open cost as a notional byte count so the planner knows that four small files cost more than their bytes suggest. These cut the task count, which is the biggest single win available without rewriting data. They do not reduce the NameNode RPCs, the seeks, or a single byte of heap.

Where small files actually come from

Nobody sets out to write a hundred million files. They fall out of writer parallelism multiplied by a clock, and neither of those terms is proportional to data volume - which is why low-volume tables usually have the worst file-to-byte ratio in the warehouse.

Streaming sinks with short windows. A structured streaming job writes at least one file per output partition per trigger. Two hundred shuffle partitions on a 30-second trigger is 576,000 files per day for that one table, whether it received a terabyte or a megabyte.

Per-partition writes. A Spark job with 400 tasks writing a dynamically partitioned table across 50 partition values can emit 20,000 files in one run: the cross product of task count and partition cardinality, entirely independent of how much data there was.

Hourly batch outputs. A job with fixed parallelism creates exactly its parallelism in files every hour, forever. Twenty-four runs a day at 32 reducers is 280,000 files a year from one job.

Ingest imports. A Sqoop-style import with -m 8 produces eight files per table per run by construction. Schedule it every 15 minutes against a small dimension table and you have generated a quarter of a million objects a year to store a few gigabytes.

Collectors that roll on time instead of size. Log shippers and CDC sinks configured to roll every N minutes produce a file per interval per source regardless of throughput, including empty files for intervals with no events. Add the debris - _SUCCESS markers, .crc sidecars, and per-task temporary output left behind by failed commits - and the object count runs ahead of the file count you think you have.

Container formats and row-group sizing

The first structural remedy is to put many logical records inside one HDFS file, which is what every serious storage format is for. The three named earlier deserve more than a name each, because the choice between them is a real one and one of them carries a sizing trap that recreates the problem it was meant to solve.

SequenceFile

A flat binary key/value container, splittable at periodic sync markers so a single large one still parallelizes properly. It remains the cleanest answer to the literal problem of "I have a million small files whose names are meaningful": key the record by original path, value the file bytes, and the whole set collapses to a handful of blocks. Block compression is supported. There is no schema evolution and no columnar projection, so it is an archival and plumbing format rather than an analytics one.

Avro

Row-oriented with the schema written into the file header, splittable at sync markers, with real schema evolution. The right choice for write-heavy ingest and for records that are always consumed whole - it appends cheaply and decodes fast.

Parquet, and the row group that everyone gets wrong

Parquet is columnar, and the unit that matters is the row group. parquet.block.size (128 MB by default) should be set to match dfs.blocksize so that a row group never straddles a block boundary and forces a reader to fetch the tail from another DataNode. A Parquet file with 5 MB row groups is a small-file problem wearing a costume: you pay the footer, the per-column chunk metadata and the dictionary pages, and you get none of the scan efficiency that justified them, because the per-file footer read starts to dominate the actual column read.

The caveat that undoes most format migrations: a format does not fix a cadence. Writing one 2 MB Parquet file every 30 seconds is worse than writing one 2 MB text file, because Parquet's per-file metadata is larger and the reader pays a footer parse per file. Container formats only help when you also batch enough records to fill the container.

HAR archives - know them, rarely use them

Hadoop Archives are introduced above as the classic consolidation answer, and on paper they are exactly that. In practice almost nobody runs them, and it is worth understanding both halves of that - the mechanism is genuinely elegant and the reasons it lost are specific.

A HAR is a directory named something.har containing a _masterindex, an _index, and one or more part- files holding the concatenated bytes of the originals. To the NameNode the entire archive is a few files and a few blocks. Clients read it through the har:// scheme, which resolves a logical path through the two index files into an offset and length inside a part file.

The appeal is real: an archive is built by a MapReduce job, the originals are untouched until you choose to delete them, and no reader needs its data rewritten. Archiving last year's directory tree can reclaim tens of millions of objects in an afternoon.

The reasons it lost anyway are equally real. Archives are immutable - you cannot add, remove or modify a file inside one, so any change means a full rebuild. Every read pays two index lookups before it touches data, and the index files become hot spots under concurrency. har:// is a second-class citizen outside the raw FileSystem API: Hive, Spark SQL and every table format handle it awkwardly or not at all, so archiving a table generally means it stops being queryable the normal way. And it only solves the metadata half of the problem - the archive still contains as many logical files, and split behaviour over a HAR is poor, so job performance does not recover the way it does after a real compaction.

The practical guidance: recognise HAR because it appears in older runbooks and certification material, and reach for it only for cold data that nobody queries with SQL. For anything live, compact into Parquet or move the data out of HDFS.

Compaction - the remedy that actually ships

Compaction is the general cure: a scheduled job that reads many small files from a directory and writes a few right-sized ones in their place. The details that separate a working compaction from a data-loss incident are all in the swap.

1. read   /warehouse/events/dt=2026-08-01/*        (N small files)
2. write  /warehouse/_compacting/events/dt=2026-08-01/   (staging path)
3. verify record count and byte total against the input
4. swap   rename partition dir aside, rename staging dir in
5. delete the aside directory (to trash, not skipTrash)
6. touch  _COMPACTED marker so the job is idempotent

Staging output somewhere else means a failed job leaves no half-written partition. Rename inside one filesystem is a metadata-only NameNode operation and is atomic per path, but there is no atomic operation that swaps a directory's contents and deletes the old set together - so you rename aside, rename in, then delete. Readers that list the directory after the swap see only the new set. Readers already in flight are a race: block deletion is asynchronous - the NameNode queues the blocks and DataNodes remove them on a later heartbeat command - so an open stream usually finishes, but a reader that reaches a block after the DataNode has acted on that command gets a BlockMissingException. That window is why the aside directory belongs in trash rather than being purged immediately, and why long-running queries and compaction of the same partition should not be scheduled against each other.

Only compact closed partitions. Compacting a partition that is still receiving writes is the classic way to duplicate or lose rows. Make a partition eligible one watermark interval after its window closes, compact it exactly once, and leave a marker file so re-running the job is cheap and safe.

Budget it honestly. Compaction reads and rewrites every byte it touches, so a daily compaction of a table ingesting 2 TB/day costs 2 TB of reads plus 2 TB of writes amplified by the replication factor - and it competes for cluster capacity with the pipelines that created the fragments. That cost is the argument for fixing the writer instead. Compaction alone is a treadmill: the source keeps refilling what you keep merging. The Hive-specific mechanics (CONCATENATE, INSERT OVERWRITE with reducer control, and the ACID compactor service) are worked through in the Hive small-file problem and Hive compaction.

Tuning the writer side

The cheapest fix in the whole article is usually one line in the job that writes the data, and unlike compaction it keeps paying.

Spark. coalesce(n) before a write merges output partitions without a shuffle, which is cheap - but it also caps the parallelism of the stage feeding it, so a coalesce(1) at the end of a heavy job serializes the whole computation onto one task. repartition(n) pays a full shuffle and returns evenly sized files. For a dynamically partitioned write, repartition on the partition columns so every partition value lands in one task and produces one file, which turns the 400 x 50 cross product above into 50 files. Better still, enable adaptive execution (spark.sql.adaptive.enabled with spark.sql.adaptive.coalescePartitions.enabled) and let the engine collapse small shuffle partitions at runtime instead of hand-tuning a constant that goes stale.

Streaming. Raise the trigger interval. Moving from a 30-second to a 5-minute trigger is a straight 10x reduction in files for identical data, and the latency change is acceptable far more often than teams assume. When it genuinely is not, serve the low-latency read from a store built for it and land HDFS files from a separate batch job on a sane cadence.

Hive. hive.merge.mapfiles, hive.merge.mapredfiles and hive.merge.tezfiles, together with hive.merge.smallfiles.avgsize and hive.merge.size.per.task, add an automatic merge step after any query whose average output file falls below the threshold. Details and the DISTRIBUTE BY pattern are in the Hive article linked above.

Guard rail. Put a name quota on ingest directories with hdfs dfsadmin -setQuota. A runaway writer then fails with a clear quota error against its own path instead of quietly consuming the NameNode heap that every other team depends on - see HDFS quotas.

Architectural answers when tuning runs out

Federation - shard the namespace

Federation runs multiple independent NameNodes, each owning its own namespace and its own block pool, over one shared pool of DataNodes. Heap pressure becomes per-namespace, so the ingest tree that generates 40,000 files a minute can be isolated from the warehouse tree. It buys headroom rather than immunity: a single hot namespace can still exhaust its own heap, cross-namespace rename stops being atomic, and quotas and snapshots become per-namespace concerns. Client-side mount tables (ViewFS) or a server-side router tier stitch the tree back together - full treatment in HDFS Federation and Router-Based Federation.

Ozone - the successor built around this limit

Ozone is the answer to the question "what would you build if the object-count ceiling were the first design constraint". It splits metadata in two: the Ozone Manager owns the key namespace, and the Storage Container Manager owns containers - and the container, a multi-gigabyte unit holding many blocks, is the thing SCM replicates and tracks. The individual block stops being an object in a central process's memory, and both managers keep their state in an on-disk key-value store rather than entirely in the JVM heap, so the working set is no longer bounded by what fits in RAM. That is the structural difference; the request path, Ratis replication and S3 gateway are covered in the Ozone article. It is not free - Ratis quorums and container state machines are new failure modes, and the ecosystem is less battle-tested than HDFS.

Object storage - move the problem to someone else

S3, GCS and ADLS have no NameNode, so there is no heap to exhaust and per-object metadata is the provider's problem. The costs relocate rather than vanish: listing a prefix holding millions of objects is slow and paginated, per-request pricing turns a million tiny GETs into a visible line item, and the absence of atomic rename breaks the rename-based commit protocols that Hadoop jobs were built on - which is precisely why specialised committers and table formats like Iceberg and Delta exist. Small files stop being an availability risk for the cluster and remain a performance problem for every query.

Detection - find it before it is an outage

Every command below is cheap enough to run from cron, and the resulting trend is what turns a 2 a.m. page into a capacity ticket filed six weeks early.

# Cluster-wide picture: capacity, live nodes, and the two counts that matter
hdfs dfsadmin -report | head -20

# Average block size is the smoking gun: Total size / Total blocks.
# Run it against a subtree, not / , on a busy production NameNode.
hdfs fsck /warehouse/events -blocks | tail -20

# Per-tree object counts: DIR_COUNT  FILE_COUNT  CONTENT_SIZE  PATHNAME
hdfs dfs -count -h /warehouse/events

# Same, plus quota headroom: NAME_QUOTA  REM_NAME_QUOTA  SPACE_QUOTA  REM_SPACE_QUOTA
hdfs dfs -count -q -h /warehouse/events

# Which table is manufacturing objects? Rank subtrees by file count.
for d in $(hdfs dfs -ls -d /warehouse/*/* | awk '{print $8}'); do
  hdfs dfs -count "$d"
done | sort -k2 -nr | head -20

# Offline, zero load on the NameNode: a real size distribution from a
# checkpointed fsimage, parsed on any other host.
hdfs oiv -i fsimage_0000000000123456789 -o /tmp/ns.tsv -p Delimited

The single most useful number is average block size, which fsck prints directly in its summary. Healthy sits close to dfs.blocksize. Anything under about 10 MB across a large tree is a small-file problem already in progress, and the ratio degrades long before the heap alarm fires. hdfs dfs -count gives directory count, file count and content size for a subtree in one line, so files-per-gigabyte per table is a trivial nightly check to record and trend.

Be careful with fsck / on a production NameNode - it walks the entire namespace and adds genuine load to the process you are trying to protect. The offline image viewer is the safe alternative for a full census: it parses a checkpointed fsimage on a separate host with zero impact on the running NameNode, and its delimited output loads straight into Hive or any local SQL engine, which makes "files under 1 MB grouped by parent directory" a query rather than a project.

On the metrics side, trend FilesTotal and BlocksTotal from the NameNode's FSNamesystem JMX bean next to MemHeapUsedM and total GC pause time. The alert worth having is not heap percentage on its own - it is object count growth rate measured against the heap ceiling, because that is the only signal that tells you how many weeks of runway remain. Pair it with a per-directory file-count alert on ingest paths and a name quota underneath as the hard backstop, and the small-file problem becomes a scheduling decision instead of an incident.