Uber Schemaless: datastore on top of MySQL

01 · CONTEXT

Why build your own datastore

In early 2014, Uber was running out of database space because of trip growth, and the team concluded it could no longer store enough trip data in Postgres. The mission became implementing the company's next generation of database technology.

The new trip storage had five main requirements. The first was to add capacity linearly with more servers, something the team's Postgres did not offer. The others were write availability, notification of dependent systems when data changes, secondary indexes, and operational trust.

The write requirement had a precedent: Uber already used a Redis buffer to replay writes that failed in Postgres. The problem is that, while the trip was in Redis, it could not be read from Postgres and functions such as billing were lost. Schemaless needed a similar mechanism, favoring write disponibilidade over read-your-writes semantics.

For the notification requirement, the old processing handled several components of the trip at the same time, which was error-prone: if a step failed, everything had to be repeated. The team wanted to separate these steps into isolated stages, triggered by data changes.

The team analyzed common alternatives, such as Cassandra, Riak, and MongoDB, among others, and the final decision came down to operational trust in the system that would hold critical trip data. In the systems the team compared, indexes over many values become slow because they use scatter-gather across all nodes.

The result was a key-value store that holds JSON without strict schema validation, hence the name. It combines append-only partitioned MySQL, buffered writes to support failing MySQL masters, triggers for change notification, and global indexes. The design is inspired by FriendFeed and the operational focus by Pinterest.

02 · DATA MODEL

Immutable cells in a sparse map

Schemaless is described as a persistent, sparse, three-dimensional, append-only hash map, similar to Google's Bigtable. The smallest entity is the cell, which is immutable: once written, it cannot be overwritten or deleted.

The cell is a JSON blob referenced by a row key, a column name, and a ref key. The row key is a UUID, the column is a string, and the ref key is an integer. To update a cell, you write a new one with a higher ref key, and the most recent is the one with the highest ref key.

There is no predefined or enforced schema: rows do not need to share column names, and those names are defined entirely by the application. Typically, the application groups related data in the same column, which gathers data that changes together and lets the schema change on its side without database downtime.

In the Mezzanine service, which stores trips, each column has a role. BASE holds basic data, such as the driver UUID and the time of day. STATUS holds the payment state, with one cell for each billing attempt. NOTES holds notes from the driver or from an operations employee, and FARE ADJUSTMENTS holds cells when the fare is adjusted. Splitting into columns avoids data races and reduces the volume written on updates.

03 · ARCHITECTURE

Workers, storage nodes, and shards

Schemaless has two kinds of nodes: workers and storage nodes. Clients talk to the workers through HTTP endpoints, and the workers route requests to the storage nodes, aggregate the results, and handle background jobs. The two types are separate so that each part can scale independently.

The data is divided into a fixed number of shards, typically 4096, mapped to storage nodes. A cell goes to a shard by its row key. The storage nodes of a shard form a cluster with one master and two minions, and the minions are spread across several datacenters, for redundancy in case of a catastrophic datacenter outage.

1. The client service sends the request over HTTP to a worker, which routes it to the storage nodes.

2. The worker uses the row key to find the cell's shard and sends the write to the cluster's master.

3. After updating the master, the storage node replicates the change asynchronously to the minions.

4. For reads, the default is the master, so the client sees the result of its own write; reading from a minion is a per-request option.

In an interview, it's worth saying which read source you would use in each case and which consistency that gives the client.

04 · FAILURES

Reads, writes, and buffered writes

When a worker is slow or fails, a library in the client tries other hosts and retries the requests. This is safe because writes in Schemaless are idempotentes, and every request can be retried.

For reads, the cluster serves requests as long as one node is available. With the master available, Schemaless always returns the most recent data. Without it, part of the data may not have reached the minion and the result may be stale, although in production the replication latency is usually under one second. The workers use a circuit breaker on connections to the storage nodes to detect a faulty node, and the read fails over to another node.

For writes, a minion going down affects nothing, because writes go to the master. If the master is down, Schemaless still accepts the write and persists it to disk on another randomly chosen master, which is similar to the hinted handoff of Dynamo and Cassandra. The cost is that subsequent reads do not see those writes until the master comes back or a minion is promoted.

Having a single master per shard has advantages and disadvantages. The advantage is the total ordering of writes in each shard, which matters for triggers, since they can read a shard's data from any node and keep the same processing order. The disadvantage is that, with the master down, writes go somewhere else where they cannot be read; in exchange, Schemaless tells the client that the cell is not immediately available for reading.

There is also the risk of MySQL's asynchronous replication: the write is lost if the master persists it and fails before replicating to the minions. Buffered writes reduce this chance by writing the data to more than one cluster. The worker writes the request to two clusters, first the secondary and then the primary, and the client is only told of success if both writes succeed.

1. The worker receives the write and stores it first in the secondary cluster and then in the primary.

2. The secondary master is chosen at random and the write goes to a special buffer table.

3. The primary master persists the cell, and the client is only told of success if both writes succeed.

4. If the primary master crashes before replication reaches the minions, the secondary master serves as a temporary backup of the data.

5. A background job monitors the primary minions and only deletes the cell from the buffer table when it shows up there.

With the secondary cluster, all data is written to at least two hosts, and the number of secondary masters is configurable. Buffered writes rely on idempotence: if a cell with the same row key, column, and ref key already exists, the write is rejected. If several identical writes are buffered, only one succeeds and the others are rejected when the primary cluster comes back.

05 · STORAGE

MySQL as the engine of each shard

Schemaless is a relatively thin layer on top of MySQL, which routes requests to the right database. MySQL's indexes and the InnoDB cache give fast query performance for cells and secondary indexes.

Each shard is a separate MySQL database. Each database has a table for the cells, the entity table, and a table for each secondary index, in addition to auxiliary tables.

The entity table has an added_id column, an auto-increment integer that is the primary key. With it as the primary key, MySQL writes the cells linearly on disk, and the triggers use it as a single pointer to extract cells in insertion order. A composite index on row key, column, and ref key makes it possible to find cells efficiently.

The cell body is the JSON object stored as a compressed MySQL blob. After experimenting with several encodings and compression algorithms, the team settled on MessagePack and ZLib for their speed and compression size.

06 · TRIGGERS AND INDEXES

From the change log to the per-shard index

Because cells are immutable and new versions are appended, each cell represents a change, and the values of an instance form a change log. It is possible to listen to these changes and trigger functions from them, as in an event bus like Kafka. This feature is triggers.

Uber's example is billing: a trigger fires when the BASE column is written in the Mezzanine instance.

1. When the trip ends, the BASE column is written in the trips instance.

2. The billing service, which fires on the BASE column, picks up the cell and tries to charge the trip to the card.

3. The billing result, success or failure, is written back to the STATUS column.

4. This way billing is decoupled from trip creation, and Schemaless acts as an asynchronous event bus.

Indexes are defined over fields of the JSON blobs. The query is efficient because it only needs to go to a single shard, and Schemaless lets you denormalize cell data directly into the index, trading storage for fast queries. In Mezzanine, a secondary index makes it possible to find a driver's trips, with the creation time and the city denormalized.

The only requirement of an index is to have one field designated as the shard field, which determines which shard the entry is written to or read from and which must be provided when querying. At the time of the article, this field had to be immutable. It should have good distribution.

Indexes are eventually consistent: the index update does not happen in the same transaction as the cell write, and the cell and index entries normally do not sit in the same shard. Consistent indexes would require 2PC on writes, with significant overhead. This cost is avoided, but users may see stale data, although most of the time the lag stays well below 20ms.