Written by Ray Mattingly, Staff Engineering Manager, Big Data @ HubSpot

Background

This post is written for an audience with backgrounds in software engineering and data infrastructure. It assumes the reader is familiar with Apache Hadoop and Apache HBase.

At HubSpot, we store petabytes of HBase data on erasure-coded HDFS (the Hadoop Distributed File System), across several HBase clusters. Erasure coding (EC) prevents data loss by storing parity blocks computed from our data, rather than storing extra full copies of it. EC cuts our storage footprint in half compared with HDFS's default of keeping three full copies of everything. But this optimization comes with a reliability tax — every byte lives on exactly one DataNode at any given time. When that DataNode gets slow, every read that needs it gets slow too, and there is no replica to fail over to.

In practice, this meant our worst-case read latencies were measured in seconds rather than milliseconds. On one of our erasure-coded clusters, which I'll use as the case study throughout this post, a typical p99 read took about 10ms. Yet reads as slow as 5 seconds occurred nearly every day.

In this post I'll dig into how we updated the HDFS client to hedge erasure-coded reads by racing slow DataNodes with concurrent reconstructions via EC parity blocks, and what it has done for our tail latency. First, I'll proactively explain a few key terms.

Glossary

HDFS HDFS is the Hadoop Distributed File System, the storage layer of Apache Hadoop. It splits files into large blocks spread across a cluster of machines, and a NameNode keeps track of where each block lives.

DataNodes DataNodes are the processes that store those HDFS blocks on disk and serve them over the network.

HBase HBase is a distributed, scalable, big data store modeled after Google's Bigtable, built on top of HDFS.

Tables and Regions An HBase table stores rows sorted by row key. HBase splits each table into regions, each holding a contiguous range of row keys, so a single table can be spread across many machines.

RegionServers A RegionServer is the HBase process that serves reads and writes for a set of regions. RegionServers don't store data themselves; they read and write HFiles in HDFS through the HDFS client, so every disk read an HBase user waits on is, under the hood, an HDFS read.

Erasure Coding With classic replication, HDFS keeps three full copies of every block. This is the default, and most common, HDFS configuration. With erasure coding (EC), HDFS instead splits each block's data into fixed-size pieces called cells (1MB each in our setup) and computes parity cells from them using Reed-Solomon coding. These have nothing to do with HBase's Cell, which is a single stored value; in this post, "cell" always means a piece of erasure-coded HDFS data. Under the RS-6-3 policy, every 6 data cells get 3 parity cells, and any 6 of those 9 cells are enough to reconstruct the rest. A row of those 9 cells is called a stripe, and each cell in a stripe lives on a different DataNode.

Reconstruction When one cell of a stripe can't be read, the HDFS client reads other cells from the same stripe, any 6 of its 9, and decodes the missing cell's bytes from them. Upstream HDFS only reconstructs when a read has failed outright, such as a DataNode that is dead or a connection that has timed out, never just because a read is slow.

Hedged Reads A hedged read is a backup request. When a read has taken longer than a set threshold, the client starts a second attempt to fetch the same data from somewhere else, keeps the original running, and uses whichever result arrives first. HDFS has supported hedged reads for normally replicated files for years, sending the hedge to another replica. HDFS has never supported hedged reads of erasure-coded data.

The Problem

One Slow DataNode Stalls Every Read That Touches It

DataNodes don't usually fail cleanly. More often they get slow: a degraded disk, a saturated network interface, a host being drained, or a machine that has dropped off the network but hasn't been declared dead yet.

With replicated data, that's trivially survivable — HDFS's hedged reads will simply fetch another replica of the exact same data. With erasure-coded data, the slow DataNode holds the only copy of its cells. A read that needs one of those cells has no choice but to wait for it.

Striping makes this worse in a way that is easy to underestimate. HDFS lays each file out in 1MB cells, round-robin across 6 DataNodes, so every RegionServer reads from many DataNodes at once. This means that one slow DataNode will affect many RegionServers.

Requests fan out even further. A single multiGet spreads its rows across dozens of RegionServers, and each of those RegionServers reads from DataNodes all over the cluster, so one slow DataNode sits in the path of a large share of requests.

For a distributed system to perform reliably, predictably, and quickly, its worst case cannot be defined by its slowest node. At scale, some node is always slow or failing. Left alone, erasure coding leaves its worst-case performance at the mercy of any single DataNode.

Worst case, when a DataNode stops accepting connections, the clients will wait out a full 5-second timeout before they give up. And, remember, one DataNode causing 5-second timeouts will mean many RegionServers are experiencing these 5-second stalls. Recently, on our case-study cluster, we had an incident where a single slow DataNode caused every RegionServer to serve reads with at least 5 seconds of latency.

In the half hour around that stall, HBase clients logged more than 2,000 errors in a single customer-facing service reading from this cluster. One unhealthy DataNode, something that should be routine in a distributed system, had turned into thousands of failed requests for our customers.

Read Replicas Can't Route Around It

HBase has its own defense against a slow host: read replicas. An HBase table can keep secondary copies of each region open on other RegionServers. When the primary region doesn't answer quickly, the client sends the same read to a replica and takes whichever answer comes back first. But a read replica isn't a second copy of the data — it's merely a second RegionServer reading the same files from HDFS. With replicated data, the two RegionServers can read different copies of a block from different DataNodes, so a slow DataNode only slows down one of your RegionServers. With erasure-coded data, there's only one copy of each cell, so the replica's read goes to the same slow DataNode and waits just as long.

The Solution

Hedged reads are a well-worn way to cut tail latency in distributed systems, and HDFS has offered them for replicated files for years. But they've never been an option for erasure-coded data because each cell exists in exactly one place. There is no second copy to send the backup request to. Reconstruction has been there all along too, but only as a fallback once a read has already failed.

Our idea is to put the two together. When a read is slow, we start reconstructing its cell from parity live, as the hedge, and race the rebuild against the original read. As the rest of this post shows, that idea has cut our worst-case erasure-coded read latency dramatically.

Reconstructing Around Slow Cells

The basic idea is that, if a cell read takes longer than a threshold, we can read the stripe's other five data cells plus one parity cell, and race decoding the slow cell against the original read. We configure the threshold per cluster with dfs.client.hubspot.striped.read.hedge.threshold.millis; today it ranges from 250ms to 1,500ms depending on the cluster. Hedged reads are additive; the original read keeps running, and the fastest read (original or hedged) serves to decode the stripe — so there is minimal downside to kicking off a hedge at a reasonably high latency threshold.

Weighing the Tradeoff

Because nothing is free, I want to be very clear about the tradeoff that we've made here: a hedge trades extra work for lower latency. For erasure-coded data, each hedge is six extra reads (the stripe's five other data cells plus one parity cell, from six different DataNodes), and the client has to decode them. That's six times the disk IO and network traffic of the read it backs up, plus some CPU. The hedge threshold decides which reads are slow enough to be worth that cost. Set it too low and we hedge reads that would have finished a moment later anyway; set it too high and we leave latency on the table. With a well-chosen threshold, the trade pays off.

On our case-study cluster, we've found an appropriate hedge threshold. From October 30 to November 5, its DataNodes served about 650 million reads per day, and its RegionServers started about 660 hedges per day, roughly one per million reads. With this threshold, our hedges have a 90% win rate.

Of course, things change: a trade that's excellent today could be a bad one during tomorrow's cluster-wide incident. So we built guardrails into our hedging logic that continuously re-evaluate the tradeoff, and trip a circuit breaker when the value proposition of a hedge is too low.

Adding a Circuit Breaker

During a broad slowdown, from saturated disks or network cards, many reads cross the threshold at once, and every hedge piles more reads onto DataNodes that are already struggling. So we added a throttler that acts as a circuit breaker, which each RegionServer's HDFS client runs on its own. When it trips, reads simply wait for slow DataNodes, as they did before hedging existed.

We introduced four new configurations to control it:

• dfs.client.hubspot.striped.read.hedge.max.impaired.nodes
  • The number of distinct DataNodes that may look impaired at once before the breaker trips. A DataNode counts as impaired when one of its reads crosses the hedge threshold. This defaults to 2, so a third impaired DataNode inside the window turns hedging off.
• dfs.client.hubspot.striped.read.hedge.impaired.window.millis
  • The window, in milliseconds, over which impaired DataNodes are counted. This defaults to 60000, or one minute.
• dfs.client.hubspot.striped.read.hedge.cooldown.millis
  • How long, in milliseconds, hedging stays off once the breaker trips. This defaults to 30000.
• dfs.client.hubspot.striped.read.hedge.max.per.second
  • A hard cap on the number of hedges a client may start per second. This defaults to 200.

We run the defaults on every cluster today.

Hedging the Connection, Not Just the Transfer

A slow DataNode doesn't always look slow once you're talking to it. Often the slow part is getting connected at all: the TCP connect, the OP_READ_BLOCK handshake, and any NameNode re-fetch after a failure all happen before a single byte moves. A DataNode that won't accept connections holds a read for the full 5-second connect timeout without ever sending anything.

So for the reads that serve HBase Gets, the hedge timer includes connection time. A DataNode that won't answer at all gets hedged just like one that sends data slowly.

Evidence of Improvement

Measuring the Worst Case

Our key metric here is tail latency: the slowest read on each RegionServer, every minute. For a baseline, we used the seven weeks before hedging, September 1 through October 21, on our case-study cluster, which runs our most aggressive threshold of 250ms. Hedging rolled out slowly on that cluster from October 22 to 28, so our with-hedging numbers begin on the 28th and data between has been omitted.

From October 28 through November 8, the cluster had a median of 3 host-minutes per day with a read over 1s, down from a baseline median of 124 — a 98% reduction in slow-read host-minutes. Host-minutes with a read over 5 seconds, which had shown up on almost every day before hedging, fell by 98% on average, and the few remaining slow requests all trace back to unrelated failures or the hedging circuit breaker tripping.

Riding Out Real DataNode Failures

Real DataNode failures make the clearest case, and they also show why hedging connection setup matters. In the two failures below, on another of our clusters, we had two versions of hedging deployed: one that hedged based on only data transfer latency, and one that hedged on connection time as well. RegionServers running each version read from the same failing DataNodes in the same minutes:

With only data transfers hedged, up to a third of RegionServers still stalled on reads over 5 seconds, waiting out connection timeouts. With connections hedged too, not one did.

Conclusion

Hedging has been a big win for our worst-case EC reads. Here is the view this post started with, next to the same view, with hedging enabled, about a month later:

Erasure coding remains one of the best cost levers available to anyone running HBase on HDFS: RS-6-3 stores our data in half the disk that triple replication needs. Sadly, its tail latency — with no second copy to read when a DataNode is slow — made it a tough fit for live, latency-sensitive customer traffic. Hedged striped reads give that redundancy back: the parity was always there, and now the client uses it when a DataNode is slow. That makes EC much more of a slam-dunk win for a wider variety of use cases.

If you'd like to dig into the implementation, it's open source in our fork of Hadoop, on the hubspot-3.3.6 branch, as four commits. You can view them all together, or individually:

• Latency-triggered reconstruction for striped reads
• Stopping hedge-issued reads before a stripe returns
• Never decoding from, or returning, a read the hedge stopped
• Hedging reads around slow block reader creation

We hope to contribute this feature back to Apache Hadoop.

Are you ready to make an impact? Check out our careers page for your next opportunity! And to learn more about our culture, follow us on Instagram @HubSpotLife.