DynamoDB Architecture

https://miafish.wordpress.com/2015/03/12/dynamo-db/

Techniques Used in Dynamo Systems

Problem	Technique	Advantage
dataset partitioning	Consistent Hashing	Incremental, possibly linear scalability in proportion to the number of collaborating nodes.
highly available writes	Vector Clock or Dotted-Version-Vector Sets, reconciliation during reads	Version size is decoupled from update rates.
handling temporary failures	Sloppy Quorum and Hinted Handoff	Provides high availability and durability guarantee when some of the replicas are not available.
recovering from permanent failures	anti-entropy using Merkle tree	Can be used to identify differences between replica owners and synchronize divergent replicas pro-actively.
membership and failure detection	gossip-based membership protocol and failure detection	Avoids having a centralized registry for storing membership and node liveness information, preserving symmetry.

Consistent hashing
Vector Clock
Sloppy Quorum and Hinted handoff:
In this method during writes if we find that the destination nodes are down we store a “hint” of the updated value on one of the alive nodes. Then when these down nodes come back up the “hints” are pushed to them thereby making the data consistent

anti-entropy using Merkle tree:
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. In this scheme, two nodes exchange the root of the Merkle tree corresponding to the key ranges that they host in common. Subsequently, using the tree traversal scheme described above the nodes determine if they have any differences and perform the appropriate synchronization action. 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
gossip-based membership protocol

http://systemdesigns.blogspot.com/2016/01/dynamodb.html

2.1 Design Consideration

数据复制(Data replication)算法传统上执行同步的副本(replica)协调，以提供一个强一致性的数据访问接口。为了达到这个水平的一致性，在某些故障情况下，这些算法被迫牺牲了数据可用性。众所周知，在早期的备份(replicated)数据库，当网络故障时，强一致性和高可用性不可能性同时实现。因此，系统和应用程序需要知道在何种情况下可以达到哪些属性。
对于容易出现的服务器和网络故障的系统，可使用积极复制的技术来提高系统的可用性，其变化可以在后台传播到副本，同时，并发和断开(disconnected)是可以容忍的。这种方法的挑战在于，它会导致更改冲突，而这些冲突必须检测并协调解决。这种协调冲突的过程引入了两个问题：何时协调它们，谁协调它们。Dynamo被设计成最终一致性(eventually consistent)的数据存储，即所有的更新操作，最终达到所有副本。

1）何时去协调更新操作冲突----是否应该在读或写过程中协调冲突。

因为Dynamo的目标是一个永远可写(always writable)的数据存储。对于Amazon许多服务来讲，拒绝客户的更新操作可能导致糟糕的客户体验。例如，即使服务器或网络故障，购物车服务必须让客户仍然可向他们的购物车中添加和删除项。这项规定迫使我们将协调冲突的复杂性推给读，以确保写永远不会拒绝。

2）谁执行协调冲突的过程。

这可以通过数据存储或客户应用程序。如果冲突的协调是通过数据存储，它的选择是相当有限的。在这种情况下，数据存储只可能使用简单的策略，如最后一次写入获胜(last write wins)，以协调冲突的更新操作。另一方面，客户应用程序，因为应用知道数据方案，因此它可以基于最适合的客户体验来决定协调冲突的方法。

this paper focuses on the core distributed systems techniques used in Dynamo: partitioning, replication, versioning, membership, failure handling and scaling. Table 1 presents a summary of the list of techniques Dynamo uses and their respective advantages.”

2.2.1 Partition Algorithm

Dynamo虚节点思想解决扩容问题。首先最简单最容易想到的就是根据资源数目对数据进行哈希分布，比如算 出一个哈希值，然后对资源数取模。这种简单处理的结果就是当资源数变化的时候，每个数据重新取模后，其分布方式都可能变化，从而需要迁移大量的数据。
举个简单的例子来说明一下，假设我的数据是自然数(1-20)，资源现在是三台主机(A，B，C)，采用取模分配方式，那么分配后A主机的数据为 (1，4，7，10，13，16，19)，B为(2，5，8，11，14，17，20) C(3，6，9，12，15，18) 现在增加一台主机D，重新分布后的结果是A(1，5，9，13，17) B(2，6，10，14，18) C(3，7，11，15，19) D(4，8，12，15，20) 。
可以看到，有大量的数据需要从一台主机迁移到另外一台主机。这个迁移过程是很消耗性能的。需要找到一种方式来尽可能减少对现存数据的影响(没有影响当然也不可能，那说明新添加的主机没有数据)。
Dynamo 采用的是 consistent hashing 来解决这个问题的。 那么我们先来了解一下什么是consistent hashing。先想象一个圆，把它看成是一个首尾相接的数轴，现在我们的数据，自然数，已经分布到这个圆上了，我们可以把我们的资源采用某种方式，随机的分布到这个圆上(图1-1)。

图1-1

现在我们让每一个资源负责它和上一个资源之间的数据，就是说A来负责区间(C，A)，B来负责区间(A，B)，C负责区间(B，C)。采用这种策略，当我 们增加一个资源主机的时候，比如H，那么我们只需要影响新节点相邻的节点A所负责的范围(只需要将A中(G, H)这个区间的数据迁移到H上)就可以了。
因为资源节点是随机分布到数据圆上的，所以当资源节点的数量足够多的时候，可以认为每个节点的负载基本是均衡的。这是原始的consistent hashing 。关于consistent hashing如果还有疑问，建议课后阅读这篇文章：http://www.paperplanes.de/2011/12/9/the-magic-of-consistent-hashing.html

但是很遗憾，Dynamo并没有采用这个模型。这个理想的理论模型跟现实之间有一个问题，在这个理论模型上，每个资源节点的能力是一样的。可现实世界，我们使用的资源却各有不同，于是Dynamo 使用了虚节点的方法，把上面的A B C等都想象成一个逻辑上的节点。一台真实的物理节点可能会包含几个虚节点(逻辑节点)，也可能只包含一个，看机器的性能而定 。我们可以把那个数据圆分成Q等份(每一个等份就是一个虚节点)，这个Q要远大于我们的资源数 。
现在假设我们有S个资源，那么每个资源就承担Q/S个等份。 当一个资源节点离开系统的时候，它所负责的等份要重新均分到其他资源节点上，一个新节点加入的时候，要从其他的节点”偷”到一定数额的等份。
这个策略下，当一个节点离开系统的时候，虽然需要影响到很多节点，但是注意，迁移的数据总量只是离开那个节点的数据量。 同样，一个新节点的加入，迁移的数据总量也只是一个新节点的数据量。 之所以有这个效果是因为Q的存在，使得增加和减少机器的时候不需要对已有的数据做重新哈希 。这个策略的要求是Q>>S。

2.2.2 Replication

解决了data partitioning的问题，接下来就是数据可靠性的问题。一般工业界认为比较安全的备份数是3份。还要考虑一个问题，各个节点间数据备份是同步还是异步。 假设我们要求写请求总是尽可能的成功，那么我们的策略是写任何一个节点成功就认为成功。节点之间的数据通过异步形式达成一致，这个时候读请求可能读不到最新写进去的信息。
比如我们一个数据在A B C 三个节点各存一份(系统中有三个备份的时候，下面的讨论都是基于这个假设的)，那么当写A成功后，另外一个进程从节点C读数据，这个时候C还没收到最新的数据，只能给读请求一个较老的版本。这个可能会带来大问题;同样，如果我们希望读请求总能读到正确的数据，那我们的策略是写的时候要等A B C三个节点都写成功了才认为写成功 。这个时候写请求可能要耗较多的时间，甚至根本不能完成(如果有节点不可达)也就是说，系统的一致性，可靠性，原子性，隔离性的问题(ACID)是无法同时达到的。只能在其中做出取舍 。
Dynamo 的处理方式是把这个选择权交给用户，这就是它的N W R模型。N代表N个备份，W代表要写入至少W份才认为成功，R表示至少读取R个备份。配置的时候要求W+R > N。 因为W+R > N， 所以 R > N-W 这个是什么意思呢?就是读取的份数一定要比总备份数减去确保写成功的倍数的差值要大 。也就是说，每次读取，都至少读取到一个最新的版本。从而不会读到一份旧数据。 当我们需要高可写的环境的时候(例如，amazon的购物车的添加请求应该是永远不被拒绝的)我们可以配置W = 1 如果N=3 那么R = 3。 这个时候只要写任何节点成功就认为成功，但是读的时候必须从所有的节点都读出数据。 如果我们要求读的高效率，我们可以配置 W=N R=1。这个时候任何一个节点读成功就认为成功，但是写的时候必须写所有三个节点成功才认为成功 。

2.2.3 Data Versioning

Dynamo中，最重要的是要保证写操作的高可用性，即“Always Writeable”，这样就不可避免的牺牲掉数据的一致性。如上所述，Dynamo中并没有对数据做强一致性要求，而是采用的最终一致性(eventual consistency)。若不保证各个副本的强一致性，则用户在读取数据的时候很可能读到的不是最新的数据。Dynamo中将数据的增加或删除这种操作都视为一种增加操作，即每一次操作的结果都作为一份全新的数据保存，这样也就造成了一份数据会存在多个版本，分布在不同的节点上。这种情况类似于版本管理中的多份副本同时有多人在修改。多数情况下，系统会自动合并这些版本，一旦合并尝试失败，那么冲突的解决就交给应用层来解决。这时系统表现出来的现象就是，一个GET(KEY)操作，返回的不是单一的数据，而是一个多版本的数据列表，由用户决定如何合并。这其中的关键技术就是Vector Clock。流程图如图2-3。

图2-3 对象的版本随时间演变

一个Vector Clock可以理解为一个<节点编号，计数器>对的列表。每一个版本的数据都会带上一个Vector Clock。通过对比两份不同数据的Vector Clock就能发现他们的关系。所以应用层在读取数据的时候，系统会连带这Vector Clock一同返回；在操作数据的时候也需要带上数据的Vector Clock一同提交。

2.2.4 Failure Detection

故障检测和处理往往都是分布式系统的重点和难点，尤其是对于像Dynamo这种对可用性要求很高的系统。上边说了，通过设定N，W，R参数来保证读写的正确。一旦出现读写失败的情况，都会触发故障处理机制。Dynamo中将故障分为两类，一类是临时性的故障，另一类是持久的故障。分别对应两种不同的处理方式。
以N=3为例，如果在一次写操作时发现节点A挂了，那么本应该存在A上的副本就会发送到D上，同时在D中会记录这个副本的元信息（MetaData）。其中有个标示，表明这份数据是本应该存在A上的，一旦节点D之后检测到A从故障中恢复了，D就会将这个本属于A的副本回传给A，之后删除这份数据。Dynamo中称这种技术为“Hinted Handoff”。另外为了应对整个机房掉线的故障，Dynamo中应用了一个很巧妙的方案。每次读写都会从”Preference List”列表中取出R或W个节点。那么只要在这个列表生成的时候，让其中的节点是分布于不同机房的，自然数据就写到了不同机房的节点上。

“Hinted Handoff”的方式在少量的或是短暂的机器故障中表现很好，但是在某些情况下仍然会导致数据丢失。如上所说，如果节点D发现A重新上线了，会将本应该属于A的副本回传过去，这期间D发生故障就会导致副本丢失。为了应对这种情况，Dynamo中用了基于 Merkle Tree的Anti-Entrpy系统，关于Merkle Tree的资料，参考Dynamo中链接: Merkle, R. A digital signature based on a conventional encryption function. Proceedings of CRYPTO, pages 369– 378. Springer-Verlag, 1988 。

1）简单的存取模式，只支持KV模式的数据存取，同时特定于小于1M的数据；
2）高可用性，即便在集群中部分机器故障，网络中断，甚至是整个机房下线，仍能保证用户对数据的读写；
3）高可扩展性，除了能够跨机房部署外，动态增加，删除集群节点，同时对正常集群影响很小。
4）数据的高可用性大于数据的一致性，短时间的数据不一致是可以容忍的，采用最终一致性来保证数据的高可用。
5）服务于内网，数据间没有隔离。
6）服务保证条约，在Amazon一切皆服务的原则下，每个模块都要为其使用者提供服务时间保证，比如在每秒500个请求的压力下，99.9%的请求要在300ms内返回。

http://systemdesigns.blogspot.com/2016/01/cassandra-vs-dynamo.html
Cassandra是一个open source, distributed, decentralized, elastically scalable, highly available, fault-tolerant, tuneably consistent, column-oriented的数据库。

大家可能会问，Cassandra与DynamoDB有什么关系呢？为什么要放在Dynamo扩展篇里面讲呢？

实际上这俩大名鼎鼎的nosql数据库体系的确有着千丝万缕的关系。有人说，Cassandra is the daughter of Amazon DynamoDb and Google Bigtable. Cassandra的data model 衍生自Bigtable，而它的distributed design则参考了Dynamo，他们都采用了一种叫做distributed hash table(DHT)的P2P结构,与传统的基于Sharding的数据库集群相比，Cassandra可以几乎无缝地加入或删除节点，非常适于对于节点规模变化比较快的应用场景。

我希望通过将Cassandra与DynamoDb对比，给大家带来一些思考。DynamoDb的设计是否完美无缺？实际应用中，它可能面对哪些问题呢？

Cassandra项目一开始是为了Facebook inbox search服务的，旨在提供一个scaled storage system以解决快速增长的数据问题。由Avinash Lakshman主持编写，这哥以前在Amazon工作，是DynamoDb的作者之一，所以我们不难理解为什么Cassandra借用了那么多Dynamo的特性了，一脉相承嘛！

好，下面来详细介绍Cassandra。

2. Cassandra 

2.1 Data Model

首先是data model，这是Cassandra与Dynamo最大的不同点所在。Dynamo使用的是key-value pair store, 而Cassandra使用的是与Google Bigtable相似的column oriented key-value store. 我们先来看两幅图:

- DynamoDb data model 示例

- Cassandra data model 示例

最明显的区别就是Cassandra的每一行都是semi－structured的，由多个column组成，但是要注意这里no relations，relation db的columns是预先定义好且不能改变的，Cassandra则可以随意增减column。

有了column的支持，数据库的操作也就有所不同了，Dynamo只能对整个row进行操作，而Cassandra却可以精确到row中的某一column! 看一下他们的api，非常明显

• Dynamo api: get(key), put(key, context, object)
• Cassandra api: get(table, key, columnName)，insert(table, key, rowMutation)，delete(table, key, columnName)

总结一下，Cassandra的data model定义如下：
Map<RowKey, SortedMap<ColumnKey, ColumnValue>>

分布式存储系统必然要把数据partition/distribute到不同的节点上存储，大家从昨天DynamoDb的课上学到的一致性哈希consistent hashing也被用作cassandra的partition algorithm，简单复习一下，这是一个hash ranges组成的环式结构，没有master，只有一种类型的节点，每一个节点负责它和上一个节点之间的数据

DynamoDb为了实现load balance，引入了virtual node的概念，期望其在整个集群中分散负载的作用，即一个物理节点承担多个逻辑节点的任务，这样当一个新物理节点加入时，期待所有其他物理节点向其迁移数据。当一个物理节点故障时，同样期待将其数据由所有其他物理节点暂管。原本consisten hashing中希望节点只会影响与它相邻的节点，然而DynamoDb为了balancing和对节点异构能力的支持，对设计原则造成了打破。

在cassandra的早期实现中，是没有virtual nodes的，后来的版本中虽然加入了virtual nodes的实现，但是提出了一些担心，假设一个物理节点非常大（TB），在向其他节点迁移的时候会带来很大的DISK I/O, NETWORK I/O, CPU开销，这种负荷会拖垮整个cluster，一些采用cassandra的网站如Reddit, Digg的事故报告中都提到了这个问题。必须采用合适的stream throttling方法进行控制。

再者，Virtual Nodes的维护是一个大问题，早期cassandra是靠人工配置来维护的。试想每个节点有几百个Virtual Nodes，要保证其不能相邻，否则散列会不均匀，balancing无法实现。这种方法如果做成动态的，节点在ring上的移动将很频繁，而造成控制混乱。如果靠手动维护，维护工作又很繁重。

（2）数据一致性 

Dynamo和Cassandra支持用户总是可写，而解决一致性冲突留给了读操作，从而获得了很好的性能增益。回顾一下N W R模型。N代表N个备份，W代表要写入至少W份才认为成功，R表示至少读取R个备份，要求W+R > N，也就是 R > N-W。Cassandra希望实现可调的一致性（tunable consistency），即通过调整R W的值，实现available和consistency之间的转换。

给W配置一个小值R配置一个大值则"writes never fail"（high availablility）；给R配置一个小值W配置一个大值则"block for all replicas to be readable"（strong consistency）

看上去非常灵活！那我们来思考一下可能会有什么不足之处呢？

a. 由于读操作时做一致性检查，对于write多read少的应用数据，read repair几乎不能保证数据的一致性。结果是数据长期不一致，甚至永远不一致。因为很多数据可能很久或者永远也不会被读到。

【Question】什么是Read Repair呢？

就是写的时候我们不要求strong consistency，read的时候才进行一致性的检查与修复。比如NWR中N=3, W=1, 在写的时候，只需要一份replica被修改就成功了，读的时候如果我们只读一份replica，那么很有可能读到的被修改的这份.所以要实现high availability我们要在w=1的同时设置R=3才能保证所有3份data都被读到，这时候再解决inconsistency的问题，就叫做read repair。

b. DynamoDb离线检查采用的Merkle Tree算法对于静态数据是比较有效的，但由于每次比较必须对现有数据重建Merkle Tree，如果数据总是不一致重建Merkle Tree需花太多系统资源和时间。Dynamo的数据模型是简单的key-value，而且每个key-value都很小。Merkle Tree用于这种数据模型也许是适用的。但对于Cassandra的数据模型Key-Columns，因为操作时的粒度是Column，每个列都要有自己的Merkle树，而且出现同一个Key的不同Column不一致的情况更加普遍，这种同步策略是否适应有待考验

再简单提一下，大家可能还记得昨天的课上有提到在merge多版本数据时DynamoDb使用了Vector Clock的数据结构，即一个<节点编号，计数器>对的列表，这个结构作为context，保存在节点的metadata中。

在Cassandra中，data model不同，熟悉Bigtable的同学可能会知道三元组<row-key, column-key, timestamp>，时间戳记录了列上一次被更新的时间，用于在服务端解决冲突。Cassandra采用最后写入获胜。当有多个版本的数据存在时，以时间戳大的值为准。不过，时间戳并不是一个自动的元数据属性，客户端在写数据的时候必须同时提供时间戳的值。

（3）故障检测

当一台机器挂掉之后，怎么通知所有的结点这个信息呢？Dynamo和Cassandra系统去中心化，没有一个结点是管理员这样的角色，通知大家一起更新。在这种环境下，各个结点之间用peer-to-peer通信方式，基于gossip protocol。简单来说，就是模拟人类社会中流言传播的方式。每个节点随机地把消息发给它的邻居，接到消息的节点，如果之前没收到这个消息，则会继续随机地转发给它的邻居，否则不转发。这样，失败的结点或者成员信息的变化会像流言一样迅速到达Dynamo的所有结点。

“Hinted Handoff”是Dynamo和Cassandra的临时性故障修复方式，简单的说就是让其他节点暂时存储某个掉线机器的data，等该机器上线后再把data“还给”该机器，并从暂时存储的机器中删除备份。为了避免整个机房掉线的问题，配置”Preference List”使数据就尽量写到不同数据中心的节点上。但如果数据中心之间相距很远，如一个机房在北美洲，一个在亚洲，则需要担心一下low latency的问题。

ebay的data infra是relational db和nosql混合，现在主要还是relational db，不过他们正在努力迁徙到nosql上，主要是因为有一些use case关系数据库不合适，比如sparse data,big data,schema optional, real time analytics，而且在scalibility上nosql更灵活。

一张图展示其Cassandra deployment：

三个数据中心构成了一个大型consistent hash ring，dns将用户的请求分配给最近的load balancer，再由其分发给其下的application server，application server与最近的data center通信。我们看到有一个data center专门留给了real time analytics，这一块主要是各种log的记录以及分析（performance monitoring，fraud detect...）

3.2 Use case: Like

再来简单看一个use case，ebay商品界面的like功能就是由cassandra负责的。

- Data Model

Data Model部分，可以看出有多个表和这个功能有关：

可以看到每个表都是典型的colum-oriented key-value store，userid或itemid作为row key, likeCount是其中一个column。用户的一个更改将会影响四张表，并且同时可能有很多用户并发修改，实际上中间有很多非常麻烦的问题。