Dynamo: Amazon's Highly Available Key-value Store

Reliability at massive scale is one of the biggest challenges we face at Amazon.com

challenges

scale, small and large components fail continuously and the way persistent state is managed in the face of these failures drives the reliability and scalability of the software systems.

motivation

Dynamo sacrifices consistency under certain failure scenarios. It makes extensive use of object versioning and application-assisted conflict resolution in a manner that provides a novel interface for developers to use.

solution using object version and assisted conflict resolution

Reliability is one of the most important requirements

motivation

operating Amazon’s platform is that the reliability and scalability of a system is dependent on how its application state is managed.

hypothesis

a particular need for storage technologies that are always available.

motivation, because Amazon consists of hundreds of services

Therefore, the service responsible for managing shopping carts requires that it can always write to and read from its data store, and that its data needs to be available across multiple data centers.

example use case for availability

Amazon’s software systems need to be constructed in a manner that treats failure handling as the normal case without impacting availability or performance

failure handling first implementation

Dynamo is used to manage the state of services that have very high reliability requirements and need tight control over the tradeoffs between availability, consistency, cost-effectiveness and performance.

Solution tradeoff

There are many services on Amazon’s platform that only need primary-key access to a data store

query pattern

A select set of applications requires a storage technology that is flexible enough to let application designers configure their data store appropriately based on these tradeoffs to achieve high availability and guaranteed performance in the most cost effective manner.

type of application that can use Dynamo

Dynamo uses a synthesis of well known techniques to achieve scalability and availability: Data is partitioned and replicated using consistent hashing [10], and consistency is facilitated by object versioning

hypothesis of how to achieve scalability and availability

The consistency among replicas during updates is maintained by a quorum-like technique and a decentralized replica synchronization protocol.

Consistency implementation

gossip based distributed failure detection and membership protocol

failure + service discovery implementation

served tens of millions requests that resulted in well over 3 million checkouts in a single day and the service that manages session state handled hundreds of thousands of concurrently active sessions

evaluation

relational database is a solution that is far from ideal. Most of these services only store and retrieve data by primary key and do not require the complex querying and management functionality offered by an RDBMS

Design motivation for not choosing RDBMS

available replication technologies are limited and typically choose consistency over availability.

limited replication of RDBMS

not easy to scale-out databases or use smart partitioning schemes for load balancing

reasons for not using RDBMS

simple query model and do not need any relational schema. Dynamo targets applications that need to store objects that are relatively small

query model motivation

Dynamo targets applications that operate with weaker consistency (the “C” in ACID) if this results in high availability. Dynamo does not provide any isolation guarantees and permits only single key updates.

motivation for dropping to weak consistency and no isolation

The tradeoffs are in performance, cost efficiency, availability, and durability guarantees.

efficiency tradeoffs

on-hostile and there are no security related requirements such as authentication and authorization

no need for security related

A common approach in the industry for forming a performance oriented SLA is to describe it using average, median and expected variance

common industry metrics

o address this issue, at Amazon, SLAs are expressed and measured at the 99.9 th percentile of the distribution. The choice for 99.9% over an even higher percentile has been made based on a cost-benefit analysis which demonstrated a significant increase in cost to improve performance

industry metrics are not good enough, Amazon aims for 99.9th percentile

availability can be increased by using optimistic replication techniques, where changes are allowed to propagate to replicas in the background, and concurrent, disconnected work is tolerated.

hypothesis of using optimistic replication

Dynamo is designed to be an eventually consistent data store; that is all updates reach all replicas eventually.

important solution of how Dynamo can achieve availability

any traditional data stores execute conflict resolution during writes and keep the read complexity simple [7]. In such systems, writes may be rejected if the data store cannot reach all (or a majority of) the replicas at a given time

traditional data store solution for conflicts

For a number of Amazon services, rejecting customer updates could result in a poor customer experience. For instance, the shopping cart service must allow customers to add and remove items from their shopping cart even amidst network and server failures.

example of why Dynamo uses read for conflict resolution

Data replication algorithms used in commercial systems traditionally perform synchronous replica coordination in order to provide a strongly consistent data access interface. To achieve this level of consistency, these algorithms are forced to tradeoff the availability of the data under certain failure scenarios.

technical problem of traditional db system

An important design consideration is to decide when to perform the process of resolving update conflicts,

important design consideration of WHEN to resolve conflicts

The next design choice is who performs the process of conflict resolution. This can be done by the data store or the application.

WHO to resolve conflicts

Systems like Pastry [16] and Chord [20] use routing mechanisms to ensure that queries can be answered within a bounded number of hops.

P2P mechanism for query

Oceanstore resolves conflicts by processing a series of updates, choosing a total order among them, and then applying them atomically in that order

how Oceanstore resolves conflicts

avoid routing requests through multiple nodes (which is the typical design adopted by several distributed hash table systems such as Chord and Pastry). This is because multihop routing increases variability in response times, thereby increasing the latency at higher percentiles.

Dynamo motivation for avoiding P2P approach

Systems like Ficus [15] and Coda [19] replicate files for high availability at the expense of consistency

expected tradeoff

These systems differ on their conflict resolution procedures. For instance, Coda and Ficus perform system level conflict resolution and Bayou allows application level resolution

difference in conflict resolutions

Antiquity is a wide-area distributed storage system designed to handle multiple server failures [23]. It uses a secure log to preserve data integrity, replicates each log on multiple servers for durability, and uses Byzantine fault tolerance protocols to ensure data consistency.

handle failures + data integrity + durability + consistency

Dynamo does not focus on the problem of data integrity and security and is built for a trusted environment.

Compared to Bigtable, Dynamo targets applications that require only key/value access with primary focus on high availability where updates are not rejected even in the wake of network partitions or server failures.

Dynamo only need key/value access

“always writeable” data store where no updates are rejected due to failures or concurrent writes

req1

Dynamo is built for an infrastructure within a single administrative domain where all nodes are assumed to be trusted

req2

do not require support for hierarchical namespace

req3

t require at least 99.9% of read and write operations to be performed within a few hundred milliseconds

req4

In addition to the actual data persistence component, the system needs to have scalable and robust solutions for load balancing, membership and failure detection, failure recovery, replica synchronization, overload handling, state transfer, concurrency and job scheduling, request marshalling, request routing, system monitoring and alarming, and configuration management

required solutions for storage system

The context encodes system metadata about the object that is opaque to the caller and includes information such as the version of the object. The context information is stored along with the object so that the system can verify the validity of the context object supplied in the put request.

system interface

One of the key design requirements for Dynamo is that it must scale incrementally.

key requirements

It applies a MD5 hash on the key to generate a 128-bit identifier, which is used to determine the storage nodes that are responsible for serving the key

how storage nodes is picked

the random position assignment of each node on the ring leads to non-uniform data and load distribution.

harder to rebalance

Second, the basic algorithm is oblivious to the heterogeneity in the performance of nodes.

bigger machines will get assigned multiple nodes

virtual node looks like a single node in the system, but each node can be responsible for more than one virtual node. Effectively, when a new node is added to the system, it is assigned multiple positions (henceforth, “tokens”) in the ring.

Dynamo implementation of consistent hashing

oad handled by this node is evenly dispersed across the remaining available nodes.

how load is balanced

The number of virtual nodes that a node is responsible can decided based on its capacity, accounting for heterogeneity in the physical infrastructure

accounts for heterogeneity

To achieve high availability and durability, Dynamo replicates its data on multiple host

needs for replication

In addition to locally storing each key within its range, the coordinator replicates these keys at the N-1 clockwise successor nodes in the ring.

what this means is if the ring is B -> C -> D, then the key not only stored in B but also C and D as well

preference list contains more than N nodes.

how replicas are used with partitions

When a customer wants to add an item to (or remove from) a shopping cart and the latest version is not available, the item is added to (or removed from) the older version and the divergent versions are reconciled later

data versioning use cases

Dynamo treats the result of each modification as a new and immutable version of the data. It allows for multiple versions of an object to be present in the system at the same time.

mechanism that guarantee eventual consistency under failure

system itself can determine the authoritative version (syntactic reconciliation).

1st method of conflict resolution

system cannot reconcile the multiple versions of the same object and the client must perform the reconciliation in order to collapse multiple branches of data evolution back into one (semantic reconciliation)

2nd method of conflict resolution

Using this reconciliation mechanism, an “add to cart” operation is never lost. However, deleted items can resurface

unwanted behavior of reconciliation mechanism

design applications that explicitly acknowledge the possibility of multiple versions of the same data (in order to never lose any updates).

applications need to be aware of multiple version for Dynamo only, not DynamoDB

Dynamo uses vector clocks [12] in order to capture causality between different versions of the same object.

because of branching in data versions

One can determine whether two versions of an object are on parallel branches or have a causal ordering, by examine their vector clocks.

check if two versions are related

In Dynamo, when a client wishes to update an object, it must specify which version it is updating

internal state of put()

Dynamo has access to multiple branches that cannot be syntactically reconciled, it will return all the objects at the leaves, with the corresponding version information in the context. An update using this context is considered to have reconciled the divergent versions and the branches are collapsed into a single new version.

conflict resolution interface

possible issue with vector clocks is that the size of vector clocks may grow if many servers coordinate the writes to an object

vector clocks size might grow too much

Dynamo employs the following clock truncation scheme: Along with each (node, counter) pair, Dynamo stores a timestamp that indicates the last time the node updated the data item. When the number of (node, counter) pairs in the vector clock reaches a threshold (say 10), the oldest pair is removed from the clock

truncation scheme for vector clock

There are two strategies that a client can use to select a node: (1) route its request through a generic load balancer that will select a node based on load information, or (2) use a partition-aware client library that routes requests directly to the appropriate coordinator nodes.

1st approach: cost steps to route request, client lighter 2st approach: client library decides how to route request directly to node but more dependency

To maintain consistency among its replicas, Dynamo uses a consistency protocol similar to those used in quorum systems. This protocol has two key configurable values: R and W. R is the minimum number of nodes that must participate in a successful read operation. W is the minimum number of nodes that must participate in a successful write operation

not strict quorum because of sloppy quorums and eventual consistency read and writes may involve different set of nodes

If Dynamo used a traditional quorum approach it would be unavailable during server failures and network partitions, and would have reduced durability even under the simplest of failure conditions

reduced durability if using strict quorum

if node A is temporarily down or unreachable during a write operation then a replica that would normally have lived on A will now be sent to node D. This is done to maintain the desired availability and durability guarantees.

hinted handoff to maintain desired availability and durability

Nodes that receive hinted replicas will keep them in a separate local database that is scanned periodically. Upon detecting that A has recovered, D will attempt to deliver the replica to A. Once the transfer succeeds, D may delete the object from its local store without decreasing the total number of replicas in the system

how hinted handoff works

Using hinted handoff, Dynamo ensures that the read and write operations are not failed due to temporary node or network failures. Applications that need the highest level of availability can set W to 1, which ensures that a write is accepted as long as a single node in the system has durably written the key it to its local store.

W to 1 = higher chance of accepted write

There are scenarios under which hinted replicas become unavailable

To handle this and other threats to durability, Dynamo implements an anti-entropy (replica synchronization) protocol to keep the replicas synchronized.

for permanent failures

detect the inconsistencies between replicas faster and to minimize the amount of transferred data, Dynamo uses Merkle trees [13].

each branch of Merkle tree can be checked independently without requiring nodes to download the entire tree

erkle trees help in reducing the amount of data that needs to be transferred while checking for inconsistencies among replica

Dynamo uses Merkle trees for anti-entropy as follows: Each node maintains a separate Merkle tree for each key range (the set of keys covered by a virtual node) it hosts. This allows nodes to compare whether the keys within a key range are up-to-date

make sure keys within a key range are up-to-date

The disadvantage with this scheme is that many key ranges change when a node joins or leaves the system thereby requiring the tree(s) to be recalculated

rebalance issues

it was deemed appropriate to use an explicit mechanism to initiate the addition and removal of nodes from a Dynamo ring.

adding nodes should be manually not automatically

A gossip-based protocol propagates membership changes and maintains an eventually consistent view of membership. Each node contacts a peer chosen at random every second and the two nodes efficiently reconcile their persisted membership change histories.

gossip-based protocol for service discovery

The mappings stored at different Dynamo nodes are reconciled during the same communication exchange that reconciles the membership change histories

how membership information is exchanged

his allows each node to forward a key’s read/write operations to the right set of nodes directly

membership facilitate read/write operations

he mechanism described above could temporarily result in a logically partitioned Dynamo ring. For example, the administrator could contact node A to join A to the ring, then contact node B to join B to the ring. In this scenario, nodes A and B would each consider itself a member of the ring, yet neither would be immediately aware of the other.

logical partition

Decentralized failure detection protocols use a simple gossip-style protocol that enable each node in the system to learn about the arrival (or departure) of other nodes.

deprecated mechanism for service discovery

When a new node (say X) is added into the system, it gets assigned a number of tokens that are randomly scattered on the ring. For every key range that is assigned to node X, there may be a number of nodes (less than or equal to N) that are currently in charge of handling keys that fall within its token range.

rebalancing nodes

by adding a confirmation round between the source and the destination, it is made sure that the destination node does not receive any duplicate transfers for a given key range

deduplication

he main reason for designing a pluggable persistence component is to choose the storage engine best suited for an application’s access patterns.

storage engine follows access pattern

The majority of Dynamo’s production instances use BDB Transactional Data Store.

In Dynamo, each storage node has three main software components: request coordination, membership and failure detection, and a local persistence engine. All these components are implemented in Java.

The request coordination component is built on top of an eventdriven messaging substrate where the message processing pipeline is split into multiple stages similar to the SEDA architecture [24]

Although it is desirable always to have the first node among the top N to coordinate the writes thereby serializing all writes at a single location, this approach has led to uneven load distribution resulting in SLA violations. This is because the request load is not uniformly distributed across objects.

If stale versions were returned in any of the responses, the coordinator updates those nodes with the latest version. This process is called read repair because it repairs replicas that have missed a recent update at an opportunistic time and relieves the anti-entropy protocol from having to do it

he coordinator for a write is chosen to be the node that replied fastest to the previous read operation which is stored in the context information of the request.

how coordinator is chosen among top N nodes

the value of N determines the durability of each object.

he values of W and R impact object availability, durability and consistency

However, this is not necessarily true here. For instance, the vulnerability window for durability can be decreased by increasing W. This may increase the probability of rejecting request

The involvement of multiple storage nodes in read and write operations makes it even more challenging, since the performance of these operations is limited by the slowest of the R or W replicas.

Dynamo provides the ability to trade-off durability guarantees for performance. In the optimization each storage node maintains an object buffer in its main memory. Each write operation is stored in the buffer and gets periodically written to storage by a writer thread.

not flushed yet

his scheme trades durability for performance. In this scheme, a server crash can result in missing writes that were queued up in the buffer.

This section discusses the load imbalance seen in Dynamo and the impact of different partitioning strategies on load distribution

he total number of requests received by each node was measured for a period of 24 hours - broken down into intervals of 30 minutes. In a given time window, a node is considered to be “inbalance”, if the node’s request load deviates from the average load by a value a less than a certain threshold (here 15%). Otherwise the node was deemed “out-of-balance”

how requests are monitored

The tokens of all nodes are ordered according to their values in the hash space. Every two consecutive tokens define a range. The last token and the first token form a range that "wraps" around from the highest value to the lowest value in the hash space.

St1 token range

when a new node joins the system, it needs to “steal” its key ranges from other nodes. However, the nodes handing the key ranges off to the new node have to scan their local persistence store to retrieve the appropriate set of data items.

ST1 - "scan" requires IO on a production node -> resource intensive

this significantly slows the bootstrapping process and during busy shopping season, when the nodes are handling millions of requests a day, the bootstrapping has taken almost a day to complete

ST1 - bootstrapping to heavy

when a node joins/leaves the system, the key ranges handled by many nodes change and the Merkle trees for the new ranges need to be recalculated

ST1 - merkle tree need to be recalculated

Finally, there was no easy way to take a snapshot of the entire key space due to the randomness in key ranges, and this made the process of archival complicated

ST1 - hard to snapshot?

The fundamental issue with this strategy is that the schemes for data partitioning and data placement are intertwined

In this strategy, the tokens are only used to build the function that maps values in the hash space to the ordered lists of nodes and not to decide the partitioning

ST2 - data partitioning

A partition is placed on the first N unique nodes that are encountered while walking the consistent hashing ring clockwise from the end of the partition.

ST2 - data placement

he primary advantages of this strategy are: (i) decoupling of partitioning and partition placement, and (ii) enabling the possibility of changing the placement scheme at runtime

Similar to strategy 2, this strategy divides the hash space into Q equally sized partitions

When a node leaves the system, its tokens are randomly distributed to the remaining nodes such that these properties are preserved

comparing these different strategies in a fair manner is hard as different strategies have different configurations to tune their efficiency.

The load balancing efficiency of each strategy was measured for different sizes of membership information that needs to be maintained at each node, where Load balancing efficiency is defined as the ratio of average number of requests served by each node to the maximum number of requests served by the hottest node

strategies were evaluated by T and Q

ategy 3 achieves the best load balancing efficiency and strategy 2 has the worst load balancing efficienc

he nodes gossip the membership information periodically and as such it is desirable to keep this information as compact as possible.

Faster bootstrapping/recovery: Since partition ranges are fixed, they can be stored in separate files,

ST3 faster bosstrapping/recovery

ii) Ease of archival: Periodical archiving of the dataset is a mandatory requirement for most of Amazon storage services. Archiving the entire dataset stored by Dynamo is simpler in strategy 3 because the partition files can be archived separately

the tokens are chosen randomly and, archiving the data stored in Dynamo requires retrieving the keys from individual nodes separately and is usually inefficient and slow.

for ST1, have to go through all nodes

he disadvantage of strategy 3 is that changing the node membership requires coordination in order to preserve the properties required of the assignment

because of Q equally sized partitions

The first is when the system is facing failure scenarios such as node failures, data center failures, and network partitions.

infra failure

The second is when the system is handling a large number of concurrent writers to a single data item and multiple nodes end up coordinating the updates concurrently.

congestion?

Semantic reconciliation introduces additional load on services, so it is desirable to minimize the need for it.

Experience shows that the increase in the number of divergent versions is contributed not by failures but due to the increase in number of concurrent writers

Write requests on the other hand will be coordinated by a node in the key’s current preference list. This restriction is due to the fact that these preferred nodes have the added responsibility of creating a new version stamp that causally subsumes the version that has been updated by the write request

picking write coordinator

In this scheme client applications use a library to perform request coordination locally. A client periodically picks a random Dynamo node and downloads its current view of Dynamo membership state.

client side load balance

An important advantage of the client-driven coordination approach is that a load balancer is no longer required to uniformly distribute client load. Fair load distribution is implicitly guaranteed by the near uniform assignment of keys to the storage nodes

bviously, the efficiency of this scheme is dependent on how fresh the membership information is at the client. Currently clients poll a random Dynamo node every 10 seconds for membership update

. A pull based approach was chosen over a push based one as the former scales better with large number of clients and requires very little state to be maintained at servers regarding clients

pull base scale better cause of less state to maintain

stale membership for duration of 10 seconds. In case, if the client detects its membership table is stale (for instance, when some members are unreachable), it will immediately refresh its membership information

refresh when stale is detected?

The latency improvement is because the client driven approach eliminates the overhead of the load balancer and the extra network hop that may be incurred when a request is assigned to a random node

client driven approach performs better then server driven

Each node performs different kinds of background tasks for replica synchronization and data handoff (either due to hinting or adding/removing node

necessary to ensure that background tasks ran only when the regular critical operations are not affected significantly. To this end, the background tasks were integrated with an admission control mechanism

admission control mechanism?

he admission controller constantly monitors the behavior of resource accesses while executing a "foreground" put/get operation.

operation. Monitored aspects include latencies for disk operations, failed database accesses due to lock-contention and transaction timeouts, and request queue wait times

metrics to monitor

Subsequently, it decides on how many time slices will be available to background tasks, thereby using the feedback loop to limit the intrusiveness of the background activitie

timeslice for background task

Dynamo exposes data consistency and reconciliation logic issues to the developers

Dynamo adopts a full membership model where each node is aware of the data hosted by its peers

membership model ~ service discovery?

his model works well for a system that contains couple of hundreds of nodes. However, scaling such a design to run with tens of thousands of nodes is not trivial because the overhead in maintaining the routing table increases with the system size

if extensive personalization techniques are used then customers with longer histories require more processing which impacts performance at the high-end of the distribution. An SLA stated in terms of mean or median response times will not address the performance of this important customer segment.

Therefore, nodes B, C, and D will offer to and upon confirmation from X transfer the appropriate set of keys.