Percolator 中的分布式事務
下一代大規模增量索引平台 – Percolator
簡介
繼google的3大基石GFS， MapReduce，BigTables之後，Google在10月份osdi會議上公布了論文《Large-scale Incremental Processing Using Distributed Transactions and Notification》，介紹了他們最新的内容索引技術。這項技術是Google下一代内容索引系統Caffeine的核心。該架構在抓取網頁的同時進行對文檔的處理，将平均延遲降低為原來的百分之一，平均文檔壽命（document age）降低50%。
motivation
傳統索引系統通常采用了多級索引的方式，按照網頁的重要性将網頁進行分級，分别按照小時，天，周對網頁庫進行更新。其特點是将網頁收錄到各級網頁庫時需要對全庫進行處理。這種模式最大的缺點是新産生的網頁或者資訊不能被及時的收錄到網頁庫中，而且定期做全庫掃描也會造成計算資源的浪費。根源在于現有MapReduce架構對于網頁庫的處理粒度過粗，一次全庫的更新需要幾天的時間才能完成。針對這個問題Percolator細化了更新粒度，提供了對文檔的随機通路，實作了對單個文檔的處理，避免了MapReduce對全庫的周遊。下面我們介紹下Percolator的設計及實作。
設計
在最初的Percolator設計中曾經考慮過直接使用BigTable存儲網頁庫，但是對于建構網頁庫而言，需要比BigTable行原子性更強的一緻性語義。而BigTable的可擴充性，容錯以及負載均衡等設計是非常優秀的，為了避免重新發明“輪子”，Percolator在BigTable基礎上，通過兩階段送出實作了跨行，跨表事務以及notifiication架構。
Percolator提供了對PB級網頁庫的随機通路功能。我們可以單獨的處理每一個頁面，進而避免使用mapreduce架構重建索引庫時對全庫的scan操作。此外為了實作高吞吐以及高并發下的同步，Percolator支援ACID相容的事務語義。
對增量索引系統而言除了使用者觸發的操作外，很多處理流程是資料觸發的。 Percolator中資料的變化，并且根據資料的變化進行觸發後續的一系列操作（比如：頁面解析，内容抽取等）。為了滿足此需求，Percolator提供了notification語義。notification語義類似DBMS中的觸發器，當Percolator中的某個cell資料發生變化，就觸發應用開發者指定的Observer程式。一系列Observer程式通過“責任鍊”的方式級聯，并完成各自邏輯。
作為一個基礎架構，Percolator繼承了BigTable的一緻性和容錯模型，并且可以通過增加機器實作叢集的線性擴充。此外，它提供了友好的開發接口，使索引系統的開發者專注于頁面解析，抽取，分詞等算法的開發，而不必被分布式系統中常見的一緻性，容錯等問題所困擾。
圖1 Percolcator主要元件
如圖1所示，Percolcator是建構在GFS和BigTable之上，主要包含3個元件。 timestamp server，Percolcator worker，Chubby。
Timestamp server提供了統一時間的服務，它保證每次擷取的時間戳單調遞增，timestamp server會持久化時間戳，以保證服務重新開機時時間戳的順序性。Chubby 服務提供了分布式鎖服務，保證Percolcator在處理全局臨界資源時，可以互斥通路。Percolcator worker是Percolator中的主要部分，實作了跨行/跨表事務和notification機制。為獲得更好的吞吐量，Percolcator worker啟動了多個BigTable client對BigTable進行并發通路。Percolcatorghm,ghdm, worker不持久化任何資料，換言之每個Percolcator worker都是無狀态的，如果Percolator worker節點失效，Percolator client可以通過重試切換到另一個Percolator worker，進而不影響服務。
Percolator的設計主要針對大規模資料存儲，以及對網頁索引具有實時性更新需求的應用。相對于傳統的OLTP，Percolator沒有集中的事務管理機制，如果有機器在進行事務的過程中失效，失效事務中的鎖釋放是一件比較大的挑戰。由于Percolator用于網頁庫建索引這樣的線下服務，放松了對實時性的要求，采用一種延遲的方式清理鎖。
實作
Percolator提供了類似BigTable的使用者接口， Percolator中的cell通過BigTable中5列來表示。其中lock， write，data 3列用于實作Percolator事務的功能。Notify和ack_O這2列是為了實作Percolator的notification機制。
下表為Percolator在BigTable中的schema。
下面我們通過Percolator中的兩個主要的feature：事務和notification架構來介紹下Percolator的實作機制。
事務
Percolaor在BigTable的基礎上，提供了跨行，跨表的ACID語義。Percolator中的事務是通過傳統的兩階段送出實作。
在實作分布式事務的時候, Percolator 使用了兩個基本的服務, 一個提供精确時間用來産生timestamp, 另外一個提供一種簡單的分布式鎖, 用來檢查一個程序是否還活着. 這兩個服務這裡不做介紹了. 我們集中精力來看 Percolator 是如何充分利用 bigtable 的單行原子性以及多版本這兩個特性來實作自己的分布式事務的. 簡單來說, 就是把多列, 多表的事務的送出和復原變成一個個單行事務.
我們看論文中的具體例子.這個例子是從Bob的賬戶中轉7美金到joe的賬戶. 這是一個涉及到兩個 row 的事務.
下面的表示中, 冒号前面的部分是一個版本号, 可以通過時間服務來解決其取值問題, lock 這個列用來放鎖的情況. write 這個列放最終的資料寫入情況.
下面的叙述中請大家注意, 單行的操作是有原子性的( bigtable 的保證).
key bal:data bal:lock bal:write
Bob 6: 6: 6: data @ 5
5: $10 5: 5:
Joe 6: 6: 6: data @ 5
5: $2 5: 5: 1 這個是初始狀态, bob 賬戶有10美金, joe 有2個美金. write 列中的 6:data @ 5 表示 目前的資料是 version 為 5 的(感謝 bigtable 的多版本支援)
Bob 7:$3 7: I am primary 7:
6: 6: 6: data @ 5
5: $10 5: 5:
Joe 6: 6: 6: data @ 5
5: $2 5: 5: 2 事務的第一個階段, bob的賬戶變成3美金了. 注意 lock 列被加鎖, 并且标明自己是 primary. 每個事務中, 隻有一個primary, 也正是這個primary的存在, 使得我們能夠用行原子性來實作分布式事務.
Bob 7: $3 7: I am primary 7:
6: 6: 6: data @ 5
5: $10 5: 5:
Joe 7: $9 7: [email protected] 7:
6: 6: 6: data @ 5
5: $2 5: 5: 3 現在給joe加上7美金, 是以joe是9美元了, 注意 joe 這一行的 lock 是指向 primary 的一個指針.
Bob 8: 8: 8: [email protected]
7: $3 7: 7:
6: 6: 6: data @ 5
5: $10 5: 5:
Joe 7: $9 7: primary @ Bob.bal 7:
6: 6: 6:data @ 5
5: $2 5: 5: 4 事務送出的第一階段, 送出 primary, 移除lock 列的内容 在 write 列寫入最新資料的 version
Bob 8: 8: 8: data @ 7
7: $3 7: 7:
6: 6: 6: data @ 5
5: $10 5: 5:
Joe 8: 8: 8: [email protected]
7: $9 7: 7:
6: 6: 6: data @ 5
5:$2 5: 5: 5 事務送出的第二階段, 送出除 primary 之外其它部分. 送出的方式也是移除 lock, 同時在 write 列寫入新資料的 version
從這個講述我們看到是怎麼把兩行的事務用單行原子性搞定的了. 事務是否送出完全取決于 primary. 如果 primary 送出了, 則lock列被清空, write列辨別了正确的版本号. 反之就是未送出. 這種方式可以用來處理多列, 多表, 的事務, 而且可以并發執行的事務數量幾乎是無限的. 因為沒有任何全局鎖.
Percolator是通過快照隔離(Snapshot isolation)實作事務的，多版本資料是快照隔離的必要條件，幸運的是bigtable可以通過時間戳來支援多版本的資料。。Snapshot isonation要求存放資料的多個版本，每個版本的資料是一個一緻的快照。它可以防止寫-寫沖突，如果兩個事務的更新 基于同一個版本的資料，那麼這兩個更新隻會有一次成功。設計的算法必須保證這一點。
Snapshot isolation并不提供串行化的保證，一般使用這種isolation的事務都對更新順序不敏感。這種isolation的優點是更新事務不會阻塞讀，每次讀取都可以讀取 一個一緻的快照，這樣讀請求就不用加鎖。
下面看下事務處理的算法僞代碼：
class Transaction
{
struct Write {Row row; Column col; string value;};
// 所有的寫都會cache在這個數組中，直到commit的時候才會進行真正的寫入
vector<Write> writes_;
// 事務開始的時間，事務基于該時間之前最後送出的版本為基礎進行修改
int start_ts_;
// 事務建立的時候擷取基準timestamp
Transaction() : start_ts_(oracle.GetTimeStamp()){}
// 所有更新都被緩存起來
void set(Write w) {writes_.push_back(w);}
// 這裡的Get主要是在更新事務中實作read-modify-write，普通讀取是不用
// 在事務中進行的，不需要判斷目前是否有人加鎖，直接擷取最新的write中記錄的資料版本，
// 然後直接讀取相應版本的資料
bool Get(Row row, Column c, string *value)
{
while(true)
{
bigtable::Txn T = bigtable::StartRowTransaction(row);
/// 檢查是否有更新事務正在進行
if (T.Read(row, c+"lock", [0,start_ts_]))
{
/// 有未決的事務; 執行clean或者等待
BackoffAndMaybeCleanupLock(row,c);
continue;
}
/// 查找小于start_ts_最後commit的事務，由于每次讀取都是通過c.write，是以在一個事務送出之前其他事務是看不到它修改的資料的
last_write = T.Read(row, c+"write", [0,start_ts_]);
if(!last_write.found()) return false; /// no data
/// 擷取事務commit的資料的timestamp
int data_ts = last_write.start_timestamp();
*value = T.Read(row, c+"data", [data_ts,data_ts]);
return true;
}
}
/// prewrite, tries to lock cell w, return false in case of conflict
/// 兩階段送出的第一階段，prewrite，對cell w進行加鎖，如果存在沖突，則傳回false
/// 沖突有兩種情況：
/// -# 事務開始之後别的程序進行了commit操作
/// -# 如果有任何别的線程已經加鎖
/// 由于行級事務的存在，保證并發的程序隻有一個能夠對cell加鎖，進而保證
/// snapshot isolation
/// 這裡值得關注行級事務如何實作read-modify-write
/// 由于prewrite沒有執行取消别人的鎖的操作，是以加鎖的時候必須保證加鎖的順序，以防止死鎖
bool Prewrite(Write w, Write primary)
{
Column c = w.col;
bigtable::Txn T = bigtable::StartRowTransaction(w.row);
/// Abort on writes after our start timestamp ...
if (T.Read(w.row, c+"write", [start_ts,MAX])) return false;
/// ... or locks at any timestamp
if (T.Read(w.row, c+"lock", [0,MAX])) return false;
/// 由于行級事務的存在，保證并發的程序隻有一個能夠對cell加鎖，進而保證
T.write(w.row, c+"data", start_ts_, w.value);
T.write(w.row, c+"lock", start_ts_, {primary.row, primary.col};
return T.Commit();
}
bool Commit()
{
Write primary = write_[0];
vector<Write> secondaries(writes_.begin() + 1, writes_.end());
/// 首先對primary加鎖
if (!Prewrite(primary, primary)) return false;
/// 然後對所有需要修改的cell加鎖
for(vector<Write>::iterator beg = secondaries.begin();
beg != secondaries.end();
beg ++)
{
if(!Prewrite(w, primary)) return false;
}
/// 這個地方為什麼不直接使用start_ts_？這樣不能夠保證snapshot isolation
/// 設想如下執行場景
/// -# T A start, start_ts_ = 8
/// -# T B start, start_ts_ = 9
/// -# T A commit, if use start_ts_, then write = 8
/// -# T B prewrite, then B will not discover that there is a commit transaction after its start
/// 是以重新擷取commit_ts是為了防止并發更新
int commit_ts = oracle_.GetTimestamp();
//首先送出primary
Write p = primary;
bigtable::Txn T = bigtable::StartRowTransaction(p.row);
if(!T.Read(p.row, p.col+"lock", [start_ts_, start_ts_]))
{
/// 由于是樂觀鎖，是以在加鎖(Prewrite)之後執行commit之前，鎖可能被其他程序
/// 取消，這種情況下事務應該失敗，由于行級事務的存在，鎖的取消和送出是串行的
return false;
}
/// start_ts_是cell的結果
T.Write(p.row, p.col+"write", commit_ts_, start_ts_);
/// 解鎖
T.Erase(p.row, p.col+"lock", commit_ts);
/// 如果primary更新失敗，則事務失敗
if(!T.Commit()) return false;
/// Sencond phase: write out write records for secondary cells
/// 下面的更新沒有在行事務中，并且更新失敗也不會讓事務abort，是以primary
/// commit成功，事務一定最終被commit
for(vector<Write>::iterator beg = secondaries.begin();
beg != secondaries.end();
beg ++)
{
/// 在沒有行事務的情況下，這裡必須保證這種寫入順序，
/// 必須先寫write再erase鎖，否則異常情況下沒有辦法恢複
bigtable::Write(w.row, w.col+"write", commit_ts, start_ts_);
bigtable::Erase(w.row, w.col+"lock", commit_ts);
}
return true;
}
};
在事務開始時，取得一個稱為start_ts_的時間戳。見line7。
寫
在commit之前，通過調用Set()函數，新記錄全部在記憶體中緩存，見line13。commit的代碼見line66-78，commit分為兩個操作。
第一，給所有要寫的記錄上鎖，見line66-78。在上鎖的過程中與其它事務可能産生兩種沖突，其一，write-write conflict，即本事務看到了待寫記錄的時間戳(write列)大于本事務的開始時間(start_ts_)，見line55；其二，在待加鎖的記錄上發現其它的事務加鎖,見line57。在發生沖突情況下，Percolator不會對事務進行排隊(not provide serializability)，是以在這種情況下本事務會被abort(需要復原，但是此時不會復原，需要其它事務的觸發)。
配置設定一個記錄上的鎖為主鎖。在這裡我們指定第一個記錄上的鎖是主鎖，其它鎖均是對這個鎖的引用。
如果沒有産生沖突，新記錄将被寫入表中，見line60。但是沒有更新時間戳(write列)，是以使用者看不到新寫入的資料。
第二，首先從timestamp服務取得一個稱為commit_ts的時間戳，見line87。從主鎖開始，釋放各個鎖并使得在第一部中的資料更新對其它事務是可見的(把start_ts_寫入至記錄的write列)。
一旦主鎖對應的記錄對其它事務是可見(即更新其write列為start_ts_)，則此事務在任何情況下都不會失敗，即事務一定會commit成功，反之，此事務可能會被abort。這個時間點稱為commit point。
讀
見line18-38。首先會檢測目的記錄上是否存在[0,start_ts_]的鎖，若有，則等待至解鎖，若沒有則讀取最新版本的資料。
處理client 異常
如何發現事務所屬client程序退出(dead)或死鎖(stuck)
利用chubby鎖服務檢測client程序的死活，并在其中寫入一個時間戳，client程序會定期的向鎖檔案寫入新的時間戳。若一個client是活的(alive)，但是其時間戳太小，client程序被認為是死鎖(stuck)。
client發生異常時，如何處理事務？
當其它client上的程序上的事務(事務A)與異常client的事務(事務B)發生沖突時(沖突點可能在line24、line57)，事務A會檢視事務B是否到達了commit point(B的主鎖是否解鎖)，若事務B還沒有到達commit point，則事務A會roll back(復原)事務B，删除事務B所有的鎖和記錄；若事務B已經到達commit point，則事務A會roll forward(使得B中的更新記錄對其它事務可見，并删除B對應的所有鎖)事務B。
從僞代碼可以看出，算法就是标準的2PC。commit的時候的執行過程如下：
- 對所有participants加鎖并執行預處理（預更新）
- 對primary執行commit
- 對所有其他participants執行commit
對primary執行commit成功與否是整個事務成功或者失敗的标志。
Percolator就是通過以上5步(從Bob的賬戶中轉7美金到joe的賬戶例子可以看出)完成事務的兩階段送出。對于分布式事務來講，由于沒有集中事務管理機制，其較大的困難就是在處理事務的過程中，Percolator client如果出現異常crash，如何清除已有的鎖并重新使用被加鎖的行。
Percolator采用的是比較消極的鎖釋放機制。如果事務A在執行的過程中，Percolator client crash掉了，Percolator 并不會主動釋放事務A所占有的鎖。如果事務B 在進行的過程中，使用到了事務A占有的資料，則由事務B負責鎖的釋放。
此時，有件比較棘手的事情就是：我們應當如何區分事務A是由于Percolator client crash導緻不能完成，還是事務A長時間處于commit 的狀态。為了避免此類競态條件，我們需要一種鎖機制來判斷這兩種情況。由于BigTable能夠保證行的原子性，Percolator 很自然的使用BigTable的一個cell作為鎖（稱為primary lock），先搶到primary lock的則可以執行，否則退出。
Notification機制
粗略的講， Notification機制類似于DBMS中的觸發器，即使用者對cell的修改都會觸發觀察該cell的程式（我們把這個程式稱為”Observer”）。多個Observer組成一個”責任鍊”，當被觀察的cell被修改的時候，Observer鍊上的全部Observer将被觸發。
在實作上，為了能夠及時發現被修改的cell，Percolator在其每一行中增加了一個BigTable列notify，用于辨別被修改且沒有觸發執行Observer的cell。為了發現被辨別為notify的cell，每個Percolator worker都會随機選取tablet，并進行scan。對于被辨別為notify的dirty cell，則觸發observer鍊。為了避免多個worker同時scan相同的tablet，worker使用分布式鎖服務(Chubby)在開始scan一個tablet的時候對tablet進行加鎖。對于網頁庫而言，這種周期性的對全表上P級資料進行scan是一個非常低效的操作，為了對其優化減少每次scan的資料量，Percolator利用了BigTable按列存儲的優勢，将notify列指定為單獨的locality group，保證worker隻對notify列進行scan，避免了全表掃描造成的對多餘資料的加載。
此外，為了區分已被觸發的Observer，Percolator增加了列ack_O,（表示名字為O的Observer對應的ack），該列的内容為對應Observer的最後啟動時間。當被觀察的列被修改，Percolator啟動一個事務來處理notification。該事務讀取被觀察的列以及其相關聯的ack列，如果該column最後寫時間大于ack_O中的時間戳，則認為該column對應的Observer需要被觸發。否則，認為Observer已經被觸發。
在Notification機制的實作上需要考慮以下幾個重要問題：
1. 如何防止多個notification事務的無限循環？
2. 如果多個notification事務觸發了對同一個cell的修改，如何避免并發修改造成的正确性問題？
對于問題1目前Percolator實作中并未對notification事務的無限循環進行檢測和防止，是以應用開發者必須在Observer開發中警惕此種情況，避免notification循環的出現。
對于問題2，Percolator的解決方法是使每一個被觸發的Observer為一個事務，如果啟動了多個Observer，同時修改一個cell，則事務鎖機制可以避免重複修改引入的正确性問題。
總結
本文簡單介紹了下一代大規模增量索引平台-Percolator所要解決的問題，實作的方法。該平台建構在Big Table和GFS上，利用了GFS的資料安全，BigTable的行原子性，負載均衡以及服務容錯等優越特性，提供了穩定，可擴充的增量索引建構平台，并且相對于傳統索引技術而言，帶來了搜尋結果的實時性和叢集資源使用率的全面提升。在一窺Google基礎架構的同時，我們可喜的看到，作為Percolator中最主要的2大功能的身影已經出現在HBase社群的開發計劃中（Transaction和Coprocessor），下一代大規模增量索引平台離我們并不遙遠。
percolator相關補充：
随着需要收集和處理的資料規模以驚人的速率增長，曾經隻有Google級别的系統才會遇到的可伸縮性需求變得更普遍，并常常需要專門的解決方案。Daniel Peng和Frank Dabek最近發表了一篇論文，介紹Google索引系統Percolator的技術細節。Percolator目前運作在數千台伺服器上，存儲了數十PB的資料，并且每天要處理數十億次的更新。
在抓取網頁的同時進行索引更新，意味着在新文檔不斷加入時，需要對已有的總文檔庫進行持續地更新。這是通過小規模、獨立的變換實作海量資料轉換任務的一個典型範例。現有的技術基礎平台恰恰不能勝任這樣的任務：傳統DBMS無法滿足存儲量和吞吐率的需求，而MapReduce和其它批處理系統無法逐個處理小規模更新，因為它們必須依賴于建立大量的批處理任務才能獲得高效率。
Daniel和Frank解釋說，盡管索引的過程是一項批處理任務，可以通過一系列的MapReduce操作來表現。但每次重新爬完一些頁面後要更新索引的時候，由于新增文檔和已有文檔之間存在連結引用的關系，隻對增量部分運作MapReduce操作是遠遠不夠的，實際上必須基于整個文檔庫進行MacReduce操作。事實上在Percolator出現之前，索引就是以上述的方式更新的。這樣帶來的主要問題就是由于要對整個文檔庫重新處理而産生的延遲。
解決此問題的關鍵是優化增量資料的處理方式。Percolator的一個關鍵設計理念是：提供對庫中文檔的随機通路，以實作對單個文檔的處理，進而避免了像MapReduce那樣對文檔全集進行處理。Percolator通過“快照隔離”實作了遵從ACID的跨行及跨表事務，進而滿足多線程在多台伺服器上對文檔庫進行轉換操作的需求。Percolator還提供了“觀察者（observer）”機制，在使用者指定的列發生更新之後，這些觀察者會被系統觸發，以幫助開發者追蹤計算過程所處的狀态。
論文作者補充到：
Percolator是專門針對處理增量更新而設計，但不是用于取代大多現有的資料處了解決方案。那些不能被拆分為單個微小更新的計算任務（比如對一個檔案排序）仍然最好由MapReduce承擔。
Percolator更适合于在高一緻性及在資料量和CPU等方面有很高需求的計算任務。對于Google來說，它的主要用途是将網頁實時地添加到Web索引中。運用Percolator，Google可以在抓取網頁文檔的同時來對文檔進行處理，進而将平均延遲降低為原來的百分之一，平均文檔壽命（document age）降低50%。
Percolator建立于分布式存儲系統BigTable之上。叢集裡的每台伺服器上運作着三個可執行檔案：worker，BigTable tablet伺服器和Google File System chunkserver伺服器
所有觀察者都被關聯到Percolator worker上，後者會對BigTable進行掃描，一旦發現更新過的列就會在worker程序中以函數調用的方式觸發（"notification"）相應的觀察者。觀察者通過向BigTable tablet伺服器發送讀、寫RPC請求來運作事務，繼而觸發後者向GFS chunkserver伺服器發送讀、寫RPC請求。
Percolator沒有提供用于事務管理的中心伺服器，也沒有全局鎖偵測器。因為Percolator不需要像運作OLTP任務的傳統DBMS一樣，對低延遲有很高要求，是以它采取了一種延遲的方式來清理鎖，也是以在事務送出時造成了數十秒的延遲。
這種方法增加了事務沖突時的延遲，但保證了系統可以擴充到幾千台伺服器的規模……盡管增量資料處理在沒有強事務的情況下也能進行，但事務使得開發者更容易地去分析系統的狀态，并避免将錯誤引入到長時間運作的文檔庫中。
Percolator的架構可以在普通廉價伺服器叢集上線性擴充多個數量級。在性能方面，Percolator處于MapReduce和DBMS之間。和DBMS相比，在處理同樣數量的資料情況下，Percolator由于其分布式架構，資源消耗遠大于DBMS，同時它還引入了約30倍的額外性能開支。和MapReduce相比，Percolator可以以低很多的延遲來處理資料，同時需要額外的資源來支援随機查找。Percolator自2010年4月開始為Google web搜尋提供索引，它利用合理的額外資源消耗，獲得了更低的延遲。