Pinterest: sharding the MySQL fleet

01 · CONTEXT

A single database that could not keep up with the product

This case comes from a Pinterest engineering post from August 2015 about how the company split its data across many MySQL servers. The numbers here are from the time of the post. The approach finished rolling out in early 2012.

In 2011 growth was too fast: around September, the entire infrastructure was over capacity. The NoSQL technologies in use ended up failing catastrophically, and the many MySQL read replicas caused annoying bugs, especially with caching. The team decided to redesign the whole storage layer.

As of the post, the interest graph already had more than 50 billion Pins across one billion boards. That is the volume the design had to sustain.

02 · REQUIREMENTS

What the design had to guarantee

The requirements were few and clear: all user content accessible at all times, and updates generally best-effort. With that, the team was willing to give up strong consistency in exchange for simplicity and disponibilidade.

Spreading data across several databases has a cost: you cannot use joins, foreign keys or indexes that span databases. Another decision was to avoid moving data item by item; if balancing is needed, you move an entire virtual node. In production, you read and write only on the master, because replicas lag and that causes strange bugs.

The storage choice was conservative: MySQL, because it is mature, instead of new auto-scaling technologies. It already delivered ordering, range selection and row-level transactions, which was what the product used. It is a practical example of database selection driven by maturity and the queries required.

03 · THE FLEET

Many small databases on few servers

The central point is separating the logical shard from the physical server. The system started with eight EC2 servers, each with one MySQL instance and several databases inside it. Each database is a shard of the data, and data that lands in a shard never leaves it. Extra capacity comes from moving whole shards to other machines.

Each server has master-master replication to a backup host, and production talks only to the master. A configuration table says which machines serve which shards; it only changes when moving shards or replacing a host, and it lived in ZooKeeper.

1. The service looks up the configuration to find out which server serves the shard it is looking for.

2. Once the server is found, the read or write goes straight to the range's master, and only to it.

3. The master replicates to the backup, which exists for recovery and high availability, not for production reads.

4. If the master dies, a script promotes the backup and a new machine is prepared to take its place.

04 · THE ID

An identifier that already says where the data lives

To distribute and later find the data, Pinterest created a 64-bit ID that carries the shard, the data type and its position in the table. It is a way of embedding the particionamento rule in the identifier itself, without consulting a central index.

The ID is assembled with bit shifts over these fields. In the post's example Pin, the decomposition shows it lives in shard 3429, and the shard configuration indicates the server where that database is.

1. The application receives a Pin's ID and extracts the shard from it, without querying any database.

2. With the shard in hand, it looks up the configuration to find out which server serves that database.

3. Connects to the right master and reads the row from the shard's database by local ID.

On writes the path is the reverse. The application picks a shard (preferably the one of the board the Pin is going into), inserts the JSON into the objects table, MySQL returns the auto-incremented local ID and, with shard, type and local ID, the system composes the full ID.

A side effect: no separate ID generator is needed. The local ID together with shard and type is already a unique identifier.

05 · DATA MODEL

Objects, mappings and application-level join

There are two kinds of data. Objects, such as Pins, users, boards and comments, have an auto-incremented local ID and a JSON blob with all the data. To edit, the system reads, modifies and writes the JSON inside a MySQL transaction. Since new fields are just fields in the JSON, no ALTER on the table is needed.

Mappings link one object to another, such as a board to the Pins on it. They are unidirectional, have a sequence column that provides ordering, and their indexes live on the shard of the source ID.

Listing a board's Pins is therefore a join at the application layer: first the IDs from the mapping, then the objects. This allows cache of the mapping separately from the object; the object cache lived in a memcache cluster and the mapping cache in a redis cluster.

06 · GROWING AND TRADE-OFFS

How the fleet grows and what is lost

The post describes three ways to gain capacity. The easiest is to upgrade the machines. The second is to open new ranges: the ID has room for far more shards than were created at the start, so it was enough to bring up new servers with the next range and let them receive the new objects. The third is to move shards to new machines. This is horizontal scaling applied to databases.

To move shards, you create a new master-master pair that replicates from the original server. Once replication finishes, you change the configuration, and each server now serves half of the ranges.

1. The original server is at its limit and serves a large range of shards.

2. A new server is created and starts replicating from the original.

3. When replication finishes, the configuration is updated with the new split.

4. Each server now serves only half of the range, with half of the load.

Some objects need to be found by something that is not an ID, such as the Facebook ID, the email or an IP address. For them there is the mod shard: the key goes through a hash and modulo of the total number of shards. In return, properties such as locality are lost, and all shards must be created up front and the key built on your own.

1. A login arrives with an external identifier that has no embedded shard.

2. The service hashes the key, finds the mod-shard server and reads the account's internal ID.

3. With the internal ID, the path goes back to the normal design: the shard is embedded in the ID.

The system is best-effort and does not provide atomicity, isolation or consistency in all cases. When a master dies, scripts promote the backup and prepare a replacement machine, and as of the post there was still no automatic failover. For migrating off the old database, the tip was to repeat the transfer until what arrives at the new system is minimal or zero. This setup had been in production for 3.5 years.