HDFS Erasure Coding — Cut Storage Overhead in Half at the Cost of Reconstruction IO

Why it matters

Cold data dominates the storage footprint of most analytical clusters. Historical logs, archived events, and old snapshots collectively account for the bulk of petabytes on production HDFS. Storing three copies of data that gets read once a month is wasteful when the primary durability requirement is data preservation, not read throughput.

Cutting storage overhead in half turns a hundred-petabyte cluster into a fifty-petabyte cluster with the same effective data volume. That directly saves capital expenditure on disks and reduces the physical footprint of the cluster. For cloud deployments, the savings are even more direct: less object storage means less monthly billing.

The architecture

The default EC scheme in HDFS is Reed-Solomon with six data cells and three parity cells, written as RS(6,3). Data is laid out in stripes: six consecutive data cells of a fixed size (1 MB under the default policy), from which three parity cells are computed using Galois-field arithmetic. The nine cells of one stripe land on nine different DataNodes, ideally in nine different racks, and a file is many such stripes written across that same set of nine nodes. The unit the NameNode tracks is therefore a block group, not a block.

The mathematical property of Reed-Solomon is that any six of the nine cells are sufficient to reconstruct the original data. This means up to three cell losses can be tolerated without losing data, which is the same tolerance as three-way replication (any two of three replicas can be lost). Durability is equivalent; the difference is in the space cost. RS(6,3) uses fifty percent overhead (three parity cells for six data cells), where three-way replication uses two hundred percent overhead (two extra copies for one original).

How it works end to end

Reading an erasure-coded file works simply if all cells are available: the client reads the six data cells in parallel from the six DataNodes hosting them and reassembles the block. There is no penalty compared to reading a replicated block, except that the read now involves six network connections instead of one.

Reading with cell loss is where the cost shows up. If any of the six data cells is missing, the client must fetch some of the parity cells and perform Reed-Solomon decoding to reconstruct the missing data cell. This is CPU-intensive and generates extra network traffic. On busy clusters, reconstruction storms during a large-scale failure can noticeably degrade cluster performance.

Writing is even more expensive. Every write requires computing the parity cells and distributing all nine cells to nine DataNodes. Compare this to replication, where the client streams data once and it is forwarded through the pipeline; EC writes involve more CPU and more distinct network destinations. This is why EC is a poor fit for hot write paths.

Reed-Solomon, and the arithmetic that motivates it

Three-way replication is a brute-force erasure code: to survive two losses it stores two extra whole copies, a 200% space tax on every byte. Reed-Solomon reaches the same fault tolerance by storing parity instead of copies. Take k data units, compute m parity units from them over a finite field (GF(2^8) in practice, so every unit is a byte vector and encoding is a matrix multiply), and distribute all k+m. The code is maximum distance separable: any k of the k+m units are sufficient to recover all k data units. Losing any m of them is survivable; losing m+1 is not.

Decoding is the same machinery run backwards. The generator matrix has the property that any k of its rows form an invertible k-by-k submatrix, so the reader takes the k units it still has, inverts the corresponding submatrix, and multiplies. Nothing special happens when the missing unit is data rather than parity - both are just rows.

The overhead follows directly from k and m, and so does the minimum sane cluster size, because every unit of a stripe must land on a distinct DataNode:

RS-6-3 is the default because it beats replication on both axes at once - half the space, one more tolerated failure. Wider schemes save more but need a wider cluster and read more to repair. XOR-2-1 is the degenerate single-parity case: parity is a plain XOR, which is nearly free to compute, but it survives only one loss and is best treated as a test or small-cluster option.

Striped versus contiguous layout

There are two ways to apply an erasure code to a filesystem, and the choice determines almost everything else about how EC behaves.

Contiguous layout keeps blocks whole. Block 0 sits entirely on one DataNode, block 1 on another, and parity blocks are computed across k such blocks. Locality is preserved - a task scheduled next to a block still reads it from one local disk - and conversion from replicated to coded data can run as a lazy background job. The cost is that a file needs at least k blocks before the scheme applies at all, and the encoder has to read k full blocks off the network to produce parity.

Striped layout, which is what HDFS shipped, splits the byte stream into small fixed-size cells and rotates them across the k data nodes. A file shorter than k blocks - even shorter than one block - still gets coded, parity is computed on the client as the data streams past, and a single sequential read pulls from k DataNodes at once, so one reader can exceed the throughput of a single disk or NIC.

The bill for striping is locality. No DataNode holds a contiguous run of the file, so there is no such thing as a node-local read of an EC block, and every read is a fan-out. That single consequence is behind most of the operational advice later in this article.

Cells, stripes, and block groups - how a file maps onto nodes

Three units stack, and conflating them is the usual source of confusion.

A cell is the striping unit, 1 MB in every built-in policy (the -1024k suffix in the policy name is literally the cell size). A stripe is k data cells plus the m parity cells computed from them - 9 MB of cells carrying 6 MB of data under RS-6-3. A block group is the whole run of stripes written across one fixed set of k+m DataNodes, and it is the object the NameNode records.

Logical bytes are dealt round-robin. Under RS-6-3, bytes 0 to 1 MB go to node 1, the next megabyte to node 2, and so on to node 6; byte 6 MB wraps back to node 1 as the first cell of stripe 2. Each DataNode therefore accumulates the cells at its own index - node 1 holds cells 0, 6, 12, ... - and those cells concatenated form that node's internal block. An internal block fills up at dfs.blocksize, so with the standard 128 MB block a full RS-6-3 block group holds 6 x 128 MB = 768 MB of logical data in nine internal blocks of 128 MB each.

The NameNode does not track nine independent blocks. Striped block IDs come from a reserved ID range with the index of the internal block encoded in the ID, so one block-group entry plus an index yields every member. That keeps the metadata cost of a block group close to that of a single block - a point worth remembering when the small-file arithmetic shows up later, because it does not mean small EC files are cheap.

Placement is unchanged in spirit but tighter in practice: the nine internal blocks want nine failure domains. See HDFS rack awareness for how the topology is expressed; the EC-specific point is that on a three-rack cluster a rack loss takes three cells at once, which is exactly the tolerance of RS-6-3 and leaves no margin.

The write path - k+m streams instead of a pipeline

A replicated write is a chain: the client streams packets to one DataNode, which forwards to the second, which forwards to the third. The client holds one connection and the cluster does the fan-out. That machinery is covered in the HDFS write pipeline.

An erasure-coded write cannot work that way, because no two DataNodes in the group receive the same bytes. The client opens a DFSStripedOutputStream, which runs one streamer per internal block - nine concurrent connections to nine DataNodes for RS-6-3 - and drives them itself. It buffers a full stripe of k cells, computes the m parity cells locally, then hands each of the k+m cells to its own streamer. Encoding is client CPU, and the fan-out is client-side connection management.

Two consequences fall out. First, a machine writing many EC files at once holds a lot more sockets and a lot more buffer memory than the replicated equivalent, and its CPU is now in the data path; an under-provisioned edge node becomes the bottleneck for the whole ingest. Second, failure handling is different in kind. There is no pipeline recovery and no replacement node mid-write: the client can lose up to m streamers and simply keep going, finishing the file with fewer than k+m internal blocks. The NameNode notices the group is short and schedules reconstruction afterwards. That is more forgiving than a pipeline stall, but it means a write that looked healthy can leave repair work behind it.

The last stripe of a file is usually partial. Parity is computed as if the missing data cells were zero-filled, but the padding is not stored - the final internal blocks are simply shorter. Space is not wasted at the tail of a large file; it is wasted at the head of a small one, which is a different problem.

Reads - the healthy path and the degraded path

When all k data cells are available the parity is never touched. The client resolves the block group, works out which cells of the requested byte range live where, and issues concurrent reads to the relevant data nodes. For a large sequential scan that means six streams instead of one, which is good for aggregate bandwidth and bad in two specific ways: six times the connection count per reader, and exposure to the slowest of six nodes rather than one. Tail latency on an EC read is the maximum over the group, so one node with a sick disk slows every reader touching that group.

A degraded read is what happens when one of the needed data cells is unavailable - node down, disk failed, DataNode restarting. The client fetches enough parity cells to make up k units in total, decodes the missing data cell in its own process, and returns the bytes. Correctness is unaffected; the read is silently more expensive. It now touches more nodes, transfers more bytes than the caller asked for, and burns client CPU on the decode.

The operational sting is that degraded reads are invisible in application-level metrics. A rolling restart across the cluster puts every reader into degraded mode for a while, and the symptom is a diffuse latency and CPU rise on the client side rather than an error anywhere. On a replicated cluster the same restart is absorbed by simply reading a different replica, at no extra cost at all.

Reconstruction cost, and why ISA-L matters

Losing a DataNode costs far more under EC than under replication, and the ratio is exactly k. To restore one lost replica, HDFS copies one surviving replica: one byte read, one byte over the wire, per byte lost. To restore one lost internal block, a DataNode must read k cells for every cell it rebuilds - six bytes read from six different nodes, per byte restored, under RS-6-3 - and then decode them. A node holding 20 TB of EC data does not generate 20 TB of repair traffic; it generates on the order of 120 TB, drawn from across the cluster.

The scheduling side is familiar: the NameNode's redundancy monitor finds block groups missing internal blocks and dispatches reconstruction tasks, which run on DataNodes rather than on the NameNode. The throttling side is where EC needs its own thinking. A reconstruction task is not one transfer, so counting it as one against dfs.namenode.replication.max-streams understates it badly; HDFS weights EC reconstruction work with dfs.datanode.ec.reconstruction.xmits.weight so that a single reconstruction consumes a share of the node's transfer budget proportional to the reads it actually performs. There are further DataNode-side knobs for the striped-read thread pool, its buffer size, and its timeout - the ones to reach for when reconstruction either starves or tramples foreground traffic.

The CPU side is where ISA-L comes in. Reed-Solomon over GF(2^8) in pure Java is slow enough to become the limiting factor on both encode and reconstruct. Intel's Intelligent Storage Acceleration Library implements the same arithmetic with vector instructions and is an order-of-magnitude class improvement, so HDFS prefers a native coder and falls back to the Java one. This is a silent fallback: a cluster with no ISA-L on the DataNodes still works, just slowly, and nothing in the logs shouts about it. Check it explicitly with hadoop checknative, which reports ISA-L availability, and note the coder preference is configurable per codec via io.erasurecode.codec.rs.rawcoders (native first, Java as fallback). Verify it on every node - a handful of hosts missing the library become the slow tail of every reconstruction storm.

Locality loss and what compute frameworks feel

MapReduce, Spark, Hive and every other HDFS consumer were built on an assumption that striping deletes: that a split has a home, and the scheduler can put the task on a node that already holds the bytes. With a striped block group, no node holds a contiguous run of the file. There is no node-local read to schedule, and short-circuit local reads - the path that bypasses the DataNode entirely and reads the block file directly - simply do not apply to erasure-coded data.

The practical effect is that every byte of an EC table scan crosses the network, where the replicated equivalent could serve a large fraction of it locally. That is usually fine: the reads are parallel across k nodes and aggregate bandwidth is good. It stops being fine when the cross-rack fabric is the constraint. Node-local traffic that used to never leave the host now hits the top-of-rack switch and, since the group is spread across racks by design, a real share of it crosses the spine. Clusters with heavily oversubscribed uplinks feel EC as a network problem long before they feel it as a storage win.

Rack-aware placement therefore stops being purely a durability setting and becomes a performance one, and the same is true of any caching layer in front of the data. The rule of thumb: EC is cheap for data you scan occasionally and expensive for data you scan repeatedly, and the difference is measured on the network, not on the disks.

Policies, and turning EC on for a directory

EC is configured as a policy - a codec, a k, an m, and a cell size - applied to a directory. The built-ins are named by their parameters: RS-6-3-1024k, RS-3-2-1024k, RS-10-4-1024k, XOR-2-1-1024k, plus RS-LEGACY-6-3-1024k using the older pure-Java coder. In Hadoop 3.x the built-in policies are disabled by default except the system default named by dfs.namenode.ec.system.default.policy, so anything other than RS-6-3 needs an explicit enable before it can be assigned.

hdfs ec -listPolicies                       # names, params, ENABLED/DISABLED
hdfs ec -enablePolicy  -policy RS-3-2-1024k # cluster-wide, once

hdfs ec -setPolicy -path /warehouse/archive -policy RS-6-3-1024k
hdfs ec -getPolicy -path /warehouse/archive
hdfs ec -unsetPolicy -path /warehouse/archive   # new files revert to replicated

hdfs ec -verifyClusterSetup                 # enough nodes/racks for enabled policies?
hdfs ec -listCodecs                         # which coders are actually loaded
hdfs ec -addPolicies -policyFile custom.xml # non-standard k, m or cell size

hdfs fsck /warehouse/archive -files -blocks # block groups, missing internal blocks

Two properties of -setPolicy catch people out. It is inherited: subdirectories created afterwards pick up the policy, and a directory can hold a mix of replicated and coded files if the policy was set partway through its life. And it applies only to new files - existing data is not rewritten. Converting an existing dataset means copying it through a directory that carries the policy, which is a DistCp job, not a metadata change, and it costs a full read and re-write of the data.

Some filesystem operations are simply unavailable on striped files: append, truncate, concat and setReplication are rejected. Anything that appends - open log files, streaming sinks, anything doing incremental writes - has to stay replicated and be coded later, once it is closed and cold.

Where EC hurts - small files and hot data

The small-file case is worse than most people expect, and the arithmetic is unforgiving. A file smaller than one cell still gets a block group, with one data internal block and m full parity internal blocks, because parity cannot be smaller than the data it protects. Under RS-6-3, a 128 KB file occupies 128 KB of data plus three 128 KB parity blocks: 4x overhead, worse than the 3x replication it replaced. Break-even against replication needs roughly k cells of data - about 6 MB under RS-6-3 - and the full 1.5x saving only arrives for files large enough to fill stripes properly.

Metadata gets worse at the same time rather than better. A small EC file is one block group with k+m internal blocks where a small replicated file was one block; NameNode pressure moves in the wrong direction. If the cluster already has a small-files problem, EC makes it worse, not better - compact first, then code.

Hot data is the other bad fit, for reasons that are now all mechanical rather than economic: every read is a k-way fan-out with the slowest node setting the latency; any node restart puts readers into decoding degraded reads; writes cost client CPU and k+m sockets; appends are unavailable; and a node failure produces k-fold repair traffic while the workload is still running.

What is left is a clean fit and a large one: cold, large, immutable, write-once data. Archived partitions, aged event logs, retained raw feeds, old snapshots - the petabytes that exist for compliance and the occasional backfill. Set the policy on those directories, leave the active tables replicated, and size the network for a reconstruction storm rather than for the steady state.