在《Designing Data-Intensive Applications》的第一部分（參考上文），介紹了資料系統的基礎理論與知識，都是基于single node。而在DDIA的第二部分（Distributed Data），則是将視野擴充到了分布式資料系統。資料的分布式主要有以下三個原因：
Scalability
Fault tolerance/high availability
Reduce latency
當負載增加的時候，有兩種應對方式，scale up vs scale out，前者指使用更強大但昂貴的裝置：更快更多核的CPU、更大的RAM、更大容量更快讀寫速度的磁盤，這中shared-memory的形式，不僅造價昂貴，而且容錯性較差。而後者，是分布式資料系統采用的shared-nothing 架構，通過增加普通機器節點（node）來應對負載的增加，這也是目前主流的應對大容量資料的方式。
如何将資料分布在多個節點上，有兩種方式，replication and partition。《Distributed systems for fun and profit 》中這個圖形象說明了這兩種方式:
當然，分布式系統并不是銀彈，分布式在帶來可擴充性、高可用性的同時，也帶來了諸多挑戰，如分布式事務，共識。
本文位址: https://www.cnblogs.com/xybaby/p/9503743.html
如上圖所示，replication（複制集）就是将一份資料（副本）儲存在多個節點上，資料的備援有以下好處
reduce latency：To keep data geographically close to your users
increase availability：To allow the system to continue working even if some of its parts have failed
increase read throughput : To scale out the number of machines that can serve read queries
複制集的最大挑戰在于資料的一緻性：如何在一定的限制條件下保證複制集中所有副本的資料是一緻的。按照在複制集中的不同角色（Leader、Follower），有三種算法 single leader, multi leader, no leader。其中，關于中心化的複制集協定(single leader）我在帶着問題學習分布式之中心化複制集一文中已經有較為詳細的介紹.
中心化複制集協定需要考慮以下問題：
（1）資料在多節節點間的寫入是同步還是異步
（2）新增的Follower（secondary）如何快速同步資料
（3）如何處理節點的故障：對于Follower（Secondary）故障，需要catch up； 對于Leader（Primary）故障，需要選舉出新的Leader，如何判斷Leader故障，如何保證在Leader Failover的過程中不丢失資料，以及避免腦裂（同時存在多個Leader）都是挑戰。
很多情況下，資料的異步寫入是更好的方式，因為有更好的可用性，并發量更高。但異步寫入，需要處理replication lag問題，即Leader與Follower之間的資料延遲，這樣使用者通過複制集中不同節點讀取到的資料可能是不一緻的。下面針對幾種具體的情況下，來看看如何保證一定程度上的一緻。
Reading Your Own Writes
使用者能夠查詢到自己成功更新的内容，但并不關心别的使用者能否立即查詢。這就需要read-after-write consistency
實作方法：
（1）當讀取的是可能被使用者修改的内容是，從leader讀取，否則可以從follower
（2）記錄更新時間，超過一定時間則從follower讀
Monotonic reads: （單調讀）
only means that if one user makes several reads in sequence, they will not see time go backward
即一個使用者如果讀到了新版本的資料，那麼重複讀取的時候，不能讀到舊版本的資料
實作辦法：
（1）每個使用者從固定一個副本讀取
Consistent prefix reads
因果關系：比如”問一個問題“與”回答該問題“，一定是前者先發生。但在複制集多個節點間異步通信的時候，第三者（Observer）可能先看到答案，後看到問題，這就違背的因果性。如圖所示：
This guarantee says that if a sequence of writes happens in a certain order, then anyone reading those writes will see them appear in the same order. One solution is to make sure that any writes that are causally related to each other are written to the same partition
如圖所示，這個問題在單個複制集（單個partition）中是不會出現的，隻有partitioned (sharded)的環境下才會出現。
解決辦法：
（1）有因果關系的操作路由到同一個partition
Leaderless Replication，去中心化的副本協定，就是說副本集中沒有中心節點，所有節點的地位是平等的，大家都可以接受更新請求，互相通過協商達成資料的一緻。在Amazon的Dynamo: Amazon’s Highly Available Key-value Store及其開源實作（Riak, Cassandra, and Voldemort）中就使用了Leaderless replication。
Leaderless的最大優點在于高可用性，不會因為單個（少數）節點的故障導緻系統的不可用，高可用性的核心在于Quorum協定：複制集中節點數目為N，當一份資料成功寫入到W個節點，每次讀取的時候得到R個節點的傳回，隻要W + R > N，那麼R中就一定包含最新的資料。如下圖所示：
事實上，每次寫入或者讀取的時候都是發給所有的節點，但是隻用等到W（R）個節點的成功傳回即可通知用戶端結果。
如上圖所示，Node 3（replica 3）由于資料寫入時故障，傳回了過時的資料，資料系統需要使複制集的資料趨于一緻，達到最終一緻性。有兩種方法
read repair： 讀取的時候多讀幾個replica的資料，修複過時的資料。
Anti-entropy process ： 背景程序檢查差異
Quorum并不是萬能的，在Leaderless中，即使使用了Quorum，還有以下潛在的問題
在不同節點的并發寫導緻的沖突，這是Leaderless最大的挑戰
在讀寫并發的情況下，缺乏隔離性，可能讀取到舊的資料
寫失敗時（少于w個節點寫入成功），不會復原
leader less下并發寫會可能沖突，在 read－repair 或者 hinted handoff 的時候也可能産生沖突。下面是一個沖突的示例：
在并發的情況下，如果每個節點收到請求就寫資料，那麼複制集就無法達成一緻，如上圖所示，不同節點資料是不一緻的。如何解決并發沖突，其中一種方式是 Last write win，cassandra就是這麼解決沖突的，作用前提：準确無誤判斷recent；每一個寫操作拷貝到所有的副本。缺點是存在資料丢失的情況：在寫入W份告知用戶端寫入成功的前提下，某些寫入會被silently discard
如何判斷兩個操作是不是并發：有沒有happened before關系
An operation A happens before another operation B if B knows about A, or depends on A, or builds upon A in some way. In fact, we can simply say that two operations are concurrent if neither happens before the other
如果存在happened before：那麼後者覆寫前者是可行的；隻有concurrent才會有沖突。
使用version vector來判斷多個寫操作的依賴關系。
關于Partitioning（Sharding），我在帶着問題學習分布式系統之資料分片一文中也有詳細介紹，可供參考。是以在本章節，隻補充新知識。
Partitioning的主要原因是伸縮性（scalability）。如何對資料進行劃分，如何rebalance資料是Partition需要解決的兩個基礎問題。
如果某個Partition上的資料或查詢比其他Partition多，那麼稱這個現象為skewed，高負載的Partition為hot spot
partitioning是按照primary index來分片的，那麼secondary indexes是如何解決的呢
two main approaches to partitioning a database with secondary indexes: document-based partitioning and term-based partitioning.
each partition maintains its own secondary indexes, covering only the documents in that partition.
每個分片維護自己的輔助索引，隻包含了在該分片上的資料的輔助索引資訊。
a document-partitioned index is also known as a local index
是以寫資料的時候隻用修改本地的輔助索引檔案。
用輔助索引查詢時，查詢語句需要在所有分片上執行，并彙總（scatter-gather）.如下圖所示，color就是一個輔助索引。
local index非常使用廣泛：MongoDB, Riak, Cassandra, Elasticsearch SolrCloud and VoltDB
也稱之為global index，輔助索引資料也分片。
相比Local index，優點是使用輔助索引讀取資料時更高效（無需scatter gather) reads more efficient. 缺點是使得寫入操作變慢而且複雜（需要分布式事務來保證）
Rebalance的目标是：
rebalance之後 各節點間負載均衡
rebalance不影響（不中斷）讀寫服務
節點間遷移資料不多不少（不要多）
分片環境下，用戶端如何得知改與哪個節點通信。
This is an instance of a more general problem called service discovery
方式：
（1）用戶端連接配接任一節點，如果該節點不能處理請求，那麼轉發到正确的節點
（2）用戶端發送請求到路由（routing tier）
This routing tier does not itself handle any requests; it only acts as a partition-aware load balancer.
（3）用戶端知道分片資訊與節點的映射關系
事務是在軟硬體出現各種異常（fault)的情況下，提升系統可靠性（reliable）的重要手段。
A transaction is a way for an application to group several reads and writes together into a logical unit. Conceptually, all the reads and writes in a transaction are executed as one operation: either the entire transaction succeeds (commit) or it fails (abort, rollback).
組成一個事務的多個操作要麼都成功（commit），要麼都不執行（rollback、abort），而不會存在部分執行成功的情況，即 all-or-nothing。
事務簡化了應用層對異常的處理，系統是否需要事務，取決于事務帶來的安全性保障，以及對應的代價。傳統的關系型資料庫都會選擇支援事務，而在分布式資料庫，如Nosql中，則（部分）放棄了對事務的支援，原因在于事務是可伸縮性的對立面，會影響大型系統的性能與可靠性。
當我們談論事務的時候，一般都是指事務的ACID特性。
一個資料庫對ACID的實作（甚至是了解）是不一定等同于其他資料庫的，其中，最複雜的是Isolation（隔離性）。
隔離性是指并發的兩個事務的執行互不幹擾，一個事務不能看到其他事務運作過程的中間狀态。當然，并發的讀是不會互相幹擾的，隻有并發的讀寫、或者并發的寫，才會帶來race condition。實作隔離性最好的方式是可串行化serializable，達到和順序執行一樣的效果，但這樣的方法存在性能問題。是以，資料庫提供不同的隔離性級别來兼顧隔離線與并發性能。
關于隔離型這一部分，筆者打算另外寫一篇筆記。
分布式系統帶來了更多的挑戰，更多意向不到的錯誤和異常，除了單點系統的問題，分布式系統還需應對的兩個難題是：
problems with networks
clocks and timing issues
與單點系統不同的是，分布式系統容易出現partial failure：即部分工作、部分異常。partial failure的最大問題是nondeterministic，不确定性。分布式系統需要在軟體層面實作容錯（fault tolerance），以應對partial failure。
分布式系統使用的網絡是不可靠的，資料包可能丢失，可能延遲。而且丢失或者延遲既可能發生在request的路上，也可能發生在response的路上，這都是不确定的。
網絡消息其中一個重要的應用就是心跳。
系統需要檢測到異常的節點，如load balancer需要監測到不工作的節點，如中心化複制集協定對leader的監測。
在節點crash的時候，如果能準确判斷且通知到系統中的其他節點，那最好不過。但是很多時候，無法判斷一個節點是否crash，而且，一個節點雖然沒有crash但也無法繼續工作，這個時候還是得靠心跳逾時，之前寫過這麼一篇文章《Hey，man，are you ok? －－ 關于心跳、故障監測、lease機制》來介紹相關問題。
當在網絡資訊中使用逾時時，逾時時長是個問題：逾時時間太長，那麼需要等很長時間；太短，又很容易容易誤判。
If the system is already struggling with high load, declaring nodes dead prematurely can make the problem worse. cascading failure
而網絡延時在各種環境下變化又很大，擁塞控制導緻發送方排隊、網絡交換機排隊、虛拟機管理排隊、CPU忙時排隊、多租戶環境下（超賣）受其他服務影響都有可能影響到網絡延時。比較厲害的就是根據網絡延時自動調節的逾時時間，如Phi Accrual failure detector ， TCP的逾時重傳就使用了類似的思想。
時間很重要，因為時間意味着：order，duration，points in time。
我們常用的時間，即time-of-day(wall-clock time).：是指根據某種月曆傳回的時間。在程式中，wall-clock time存在一些問題
NTP可能導緻時間回退
通常會忽略閏秒
是以wall-clock time不适合衡量時間差（measuring elapsed time）
是以，作業系統提供了另一種時間Monotonic clocks，如Linux上的clock_gettime(CLOCK_MONOTONIC)，Monotonic clocks保證了時間不會jump back。
當分布式系統中各個節點的時鐘不一緻時，會出現各種問題，如一個常用但容易出問題的場景：用時間（timestamp）來判斷多個節點上事件發生的順序
在LeaderLess複制集中，last write win（lww）是解決并發沖突的一個方法，如果這個時候不同節點資料不一緻，可能導緻資料被 悄無聲息 地丢失。
即使使用了NTP，也無法完全保障各節點間資料的一緻。一種有意思的想法是使用置信區間：
Clock readings have a confidence interval： it doesn’t make sense to think of a clock reading as a point in time—it is more like a range of times, within a confidence interval:
很多算法與協定，依賴對本地時間的判斷，如Lease，即使各節點的資料一緻，在某些情況下也會出問題，那就是Process Pause。
比如，某段代碼執行前會去check lease，check的時候滿足lease，然後發生了Process Pause，恢複的時候可能已經不再滿足lease了。因為不知道哪裡可能會pause，也就無從再次檢查
什麼會導緻Process Pause呢，很多：
gc
virtual machine can be suspended and resumed
多線程
磁盤IO： 非預期的disk access，如python import
swap
Unix SIGSTOP（Ctrl z）
特點是gc這種stop the world的行為，在有記憶體管理的程式設計語言Java、Python中時有發生。
gc導緻process Pause，在Hbase中就發生過，如圖所示：
分布式鎖的實作中，使用了lease，即使在stop-the-world-gc pause，client 1任然認為自己持有lease，而事實上client 1持有的lease已經過期。是以在分布式系統中：
The Truth Is Defined by the Majority。 A node cannot necessarily trust its own judgment of a situation.
解決辦法很簡單：fencing token
當我們提及算法和協定，總是基于一定的系統模型，系統模型是算法工作環境的前提或者假設
system model, which is an abstraction that describes what things an algorithm may assume。
對于時間的假設：
Synchronous model
Partially synchronous model： 絕大多數是同步的，bounded；偶爾超出bound問題不大 ，依靠imeout機制
Asynchronous model
對于Node failure的假設
Crash-stop faults
Crash-recovery faults（nodes are assumed to have stable storage）
Byzantine (arbitrary) faults
如何衡量一個算法設計與實作是否正确呢：在系統模型下，所承諾的屬性（ properties）都得以滿足。比如unique屬性，比如事務中的Atomic屬性。
屬性可以分為兩類：
Safety：nothing bad happens, liveness： something good eventually happens.
分布式算法，在任何系統模型下，都需要滿足safety屬性
For distributed algorithms, it is common to require that safety properties always hold, in all possible situations of a system model，However, with liveness properties we are allowed to make caveats:
本章讨論在分布式系統中的容錯算法、協定。
建構容錯性系統的最好方式是：找出并實作通用的抽象模型（這些抽象解決一類問題），這樣應用層代碼就無需考慮、處理這些問題，即使發生各種異常。如資料庫提供的事務。在分布式系統中：重要的抽象就是共識 consensus: that is, getting all of the nodes to agree on something。
在CAP理論與MongoDB一緻性、可用性的一些思考 一文中介紹過CAP理論，CAP理論是說對于分布式資料存儲，最多隻能同時滿足一緻性（C，Consistency）、可用性（A， Availability）、分區容錯性（P，Partition Tolerance）中的兩者。強一緻性能保證對于每一次讀操作，要麼都能夠讀到最新寫入的資料，要麼錯誤。
linearizability 能實作強一緻性，因為
make a system appear as if there were only one copy of the data, and all operations on it are atomic
線性一緻性是一個很有用的特性：比如通過lock的形式來選舉leader，那麼鎖必須是線性的linearizable:；比如unqueness限制。
不同的複制集協定能否保持線性呢？對于single leader：如果隻從leader讀資料，那麼基本上是線性的，也有例外，如資料被復原，這個時候就不能保證線性。對于leaderless，理論上使用Quorum來保證線性，但實際中，也會出現非線性，如下圖所示
這個圖說明了在滿足quorum的情況下，也不能保證線性，上圖是dirty read的情況，另外如果出現部分節點寫失敗，讀取的時候也不能保證線性。
linearizability其實就是強一緻性，雖然linearizability容易了解，易于使用，但分布式系統大多選擇不支援linearizability，原因在于線性一緻性容錯性差，性能也不好。
在分布式一緻性語義下，線性就是隻有一份資料，且每個操作在某個時間點原子性執行，這就意味着某種順序
Linearizability是total order，隻有一份拷貝，且操作原子性發生，所有操作都有相對順序。但事實上，很多操作是可以并發執行的，隻要互相不影響。
Causality consistenc（因果一緻性）是partial order，某些操作間是有順序的，其他操作則是可以并發的。
In fact, causal consistency is the strongest possible consistency model that does not slow down due to network delays, and remains available in the face of network failures 線性一緻性代價大，且很多時候沒有必要
要在多個節點間記錄因果順序是比較複雜的，具體參考lamport timestamp
上述的因果一緻性并不能解決所有問題，比如當兩個使用者并發登記同一個username，是沒有因果的，但并不滿足username的uniqueness限制，是以需要共識算法。共識就是幾個節點對某件事情達成一緻，顯然共識能解決uniqueness constraint問題。初次之後，比如single leader的選舉，比如 分布式事務的atomic commit，都需要共識。
Two-Phase Commit (2PC)是實作分布式事務的經典手段，通過2PC，也能實作共識。但是2PC的問題在于容錯性差，節點故障和網絡逾時都會導緻重試，直到節點或者網絡恢複
consensus算法定義：
one or more nodes may propose values, and the consensus algorithm decides on one of those values
公式算法要滿足的屬性：
Uniform agreement No two nodes decide differently. Integrity No node decides twice. Validity If a node decides value v, then v was proposed by some node. Termination Every node that does not crash eventually decides some value.
前三是safety屬性，最後一個是liveness屬性，最後一個也要求了系統要有容錯性（2pc就不能滿足這個屬性）
single leader能保證共識，但single leader的選舉依賴于共識算法，常見的容錯的共識算法包括（Viewstamped Replication (VSR) , Paxos , Zab）
共識算法依賴leader，但leader不是固定的：the protocols define an epoch number (called the ballot number in Paxos, view number in Viewstamped Replication, and term number in Raft) and guarantee that within each epoch, the leader is unique
是以，single leader隻是緩兵之計，不是不需要共識，而是不需要頻繁的共識。
不同的資料系統選擇不同的形式來滿足leader 選舉等共識需求，如mongodb，在replica node間使用類似raft的算法來選舉leader。而其他系統，如hbase，使用outsourced的服務（如zookeeper）來達成共識、故障檢測，把專業的事交給專業的人，大大簡化了資料系統的複雜度。
DDIA的第二部分資訊量很大，設計到大量的算法和理論，僅僅看這本書是很難搞明白的。于我而言，對LeaderLess replication與consensus這些部分還不是很清楚，比如LeaderLess因果性，vector clock、lamport clock、Paxos & Raft算法，還需要花點時間研究一下。
Designing Data-Intensive Applications
Distributed systems for fun and profit
本文版權歸作者xybaby（博文位址：http://www.cnblogs.com/xybaby/）所有，歡迎轉載和商用，請在文章頁面明顯位置給出原文連結并保留此段聲明，否則保留追究法律責任的權利，其他事項，可留言咨詢。