????????今天要講的文章是SOSP 2017年的一篇文章，Wukong+S?：Sub-millisecond Stateful Stream Querying over Fast-evolving Linked Data。本文主要解決的問(wèn)題是：隨著流數(shù)據(jù)和存儲(chǔ)數(shù)據(jù)量的不斷增加，及時(shí)查詢有用的信息十分重要。對(duì)于公共數(shù)據(jù)集合數(shù)據(jù)流，可能有大量的用戶不同的數(shù)據(jù)流查詢請(qǐng)求，因此需要支持高并發(fā)的查詢。而且流數(shù)據(jù)通常包含巨大的有用信息， 這樣的數(shù)據(jù)應(yīng)該被一致地和立即地整合到存儲(chǔ)系統(tǒng)，以用于將來(lái)的連續(xù)查詢。

????????然而目前現(xiàn)有的系統(tǒng)對(duì)于正在變化的數(shù)據(jù)集的側(cè)重點(diǎn)在于流計(jì)算。流計(jì)算和流查詢不同的是 前者通常傾向于對(duì)大部分流數(shù)據(jù)進(jìn)行序列化計(jì)算，而后者側(cè)重于對(duì)流和存儲(chǔ)數(shù)據(jù)的特定集合的并發(fā)查詢。大多數(shù)先前的系統(tǒng)也沒(méi)有集成流數(shù)據(jù)為了并發(fā)的查詢中，或者不查詢持久化存儲(chǔ)的歷史數(shù)據(jù)來(lái)獲得基礎(chǔ)知識(shí)，因此是無(wú)狀態(tài)的。 盡管大多數(shù)流處理數(shù)據(jù)庫(kù)都明確支持語(yǔ)義和SQL接口，但是他們?cè)诳焖傺莼腖inked 數(shù)據(jù)下，當(dāng)面臨大量并發(fā)的查詢請(qǐng)求下，由于高昂的Join操作開(kāi)銷和一些ACID的語(yǔ)義，他們的查詢性能是低效的。

????????Wukong+S為了尊重?cái)?shù)據(jù)本地化并最大限度地減少數(shù)據(jù)傳輸，它使用由基于時(shí)間的臨時(shí)存儲(chǔ)和連續(xù)持久化存儲(chǔ)組成的混合存儲(chǔ)，為正在到來(lái)的數(shù)據(jù)和持久化的數(shù)據(jù)提供不同的存儲(chǔ)管理。Wukong + S提供了流數(shù)據(jù)的快速訪問(wèn)流索引。流數(shù)據(jù)通過(guò)局部感知分區(qū)進(jìn)行分片，其中一些流索引在節(jié)點(diǎn)間動(dòng)態(tài)復(fù)制。這節(jié)省了查詢成本并提供高效的負(fù)載平衡。為了在多個(gè)不同規(guī)模的流數(shù)據(jù)上提供一致的流查詢，Wukong + S使用分散的矢量時(shí)間戳來(lái)推導(dǎo)出最近一致性狀態(tài)的流式數(shù)據(jù)插入。Wukong + S使用有限的標(biāo)量化方案將矢量時(shí)間戳投影到標(biāo)量快照數(shù)量中，通過(guò)協(xié)調(diào)多個(gè)流的更新到底層持久存儲(chǔ)區(qū)。這樣的設(shè)計(jì)在有效的內(nèi)存使用情況下擴(kuò)展了Wukong + S節(jié)點(diǎn)和大量查詢。

1. RDF and SPARQL?

????????那么什么是RDF數(shù)據(jù)呢？RDF全稱就是資源描述框架。他將我們生活中的描述成每一個(gè)實(shí)體和實(shí)體之間的關(guān)系。每一個(gè)實(shí)體就是圖中的一個(gè)頂點(diǎn)，相鄰頂點(diǎn)的邊就是兩個(gè)實(shí)體之間的關(guān)系。他通過(guò)一組基于主題、謂詞、對(duì)象的三元組集合 來(lái)描述現(xiàn)實(shí)生活中的資源。

2. BackGround

????????在我們生活中，每時(shí)每刻都在產(chǎn)生著大量的數(shù)據(jù)。不同的數(shù)據(jù)源高速不斷地生成實(shí)時(shí)流數(shù)據(jù)。比如說(shuō)社交網(wǎng)絡(luò)數(shù)據(jù)、城市監(jiān)控?cái)?shù)據(jù)、視頻圖像數(shù)據(jù)。

????????在各種各樣的應(yīng)用場(chǎng)景下，產(chǎn)生的大數(shù)據(jù)主要可以分為以下兩類。一類是實(shí)時(shí)流數(shù)據(jù)：這樣的數(shù)據(jù)是非常快速的產(chǎn)生。一類是存儲(chǔ)數(shù)據(jù)：也就是歷史數(shù)據(jù)（存在的有價(jià)值的歷史數(shù)據(jù)），這些數(shù)據(jù)是非常巨大的并且它由于stremdata，它還會(huì)不斷演化。????????

????????然而對(duì)于社交網(wǎng)絡(luò)，城市監(jiān)控和市場(chǎng)反饋等應(yīng)用需要有狀態(tài)的流式查詢，狀態(tài)流查詢不僅要查詢存儲(chǔ)歷史數(shù)據(jù)，也要從實(shí)時(shí)流數(shù)據(jù)從提出有用的信息，查詢實(shí)時(shí)流數(shù)據(jù)。實(shí)時(shí)流數(shù)據(jù)提取的有用信息，也需要持續(xù)不斷地整合到存儲(chǔ)的數(shù)據(jù)中，以便為上述和未來(lái)提供查詢服務(wù)。 然而目前現(xiàn)有的系統(tǒng)對(duì)于正在變化的數(shù)據(jù)集的側(cè)重點(diǎn)在于流計(jì)算。流計(jì)算和流查詢不同的是 前者通常傾向于對(duì)大部分流數(shù)據(jù)進(jìn)行序列化計(jì)算，而后者側(cè)重于對(duì)流和存儲(chǔ)數(shù)據(jù)的特定集合的并發(fā)查詢。大多數(shù)先前的系統(tǒng)也沒(méi)有集成流數(shù)據(jù)為了并發(fā)的查詢中，比如關(guān)系型數(shù)據(jù)庫(kù)，或者不查詢持久化存儲(chǔ)的歷史數(shù)據(jù)來(lái)獲得基礎(chǔ)知識(shí)，因此是無(wú)狀態(tài)的。

3.Example Dataset for Stateful Stream Query

我們來(lái)舉一個(gè)簡(jiǎn)單的狀態(tài)流查詢的實(shí)例。

????????LinkeData 表示就是實(shí)體和實(shí)體之間的關(guān)系信息，存儲(chǔ)的靜態(tài)數(shù)據(jù)。假設(shè)目前存在這樣一個(gè)查詢需求：在過(guò)去30分鐘內(nèi)，哪些IPADS 成員發(fā)布了一個(gè)推文。這個(gè)推文被其他IPADS的成員點(diǎn)贊了。當(dāng)這個(gè)查詢請(qǐng)求被用戶提交后，這個(gè)查詢請(qǐng)求將不斷在后臺(tái)服務(wù)器中計(jì)算，并且不斷地產(chǎn)生查詢結(jié)果。直到這個(gè)查詢請(qǐng)求被用戶取消。

????????我們仔細(xì)分析上面的Example，可以得出這樣的一個(gè)抽象的系統(tǒng)。在這樣的一個(gè)系統(tǒng)中，存在著連接屬性圖數(shù)據(jù)。狀態(tài)流查詢不僅讀取Store數(shù)據(jù)還讀取實(shí)時(shí)流數(shù)據(jù)。并且這個(gè)存儲(chǔ)的數(shù)據(jù)還是隨著時(shí)間不斷演化的，它會(huì)從實(shí)時(shí)流數(shù)據(jù)中吸收永恒的數(shù)據(jù)。

4.?Conventional Approach

????????傳統(tǒng)的解決方案是將實(shí)時(shí)流處理系統(tǒng)和圖儲(chǔ)存系統(tǒng)相結(jié)合。一次性查詢?nèi)蝿?wù)直接被發(fā)送到Graph Store System 中，直接查詢歷史存儲(chǔ)數(shù)據(jù)。而連續(xù)查詢將被拆分成分別在流處理系統(tǒng)中和面向查詢?yōu)橹鞯膱D處理系統(tǒng)兩部分分別執(zhí)行。這兩部分的結(jié)果將被join連接起來(lái)以得到最終結(jié)果。然而這樣的混合設(shè)計(jì)仍然不全是狀態(tài)查詢。因?yàn)橐淮涡圆樵冎粫?huì)運(yùn)行在靜態(tài)的存儲(chǔ)數(shù)據(jù)中而沒(méi)有從實(shí)時(shí)流數(shù)據(jù)得到更新后的實(shí)時(shí)數(shù)據(jù)。除此之外，這里還存在一些關(guān)鍵性的缺陷。

????????由于目前存在的CSPARQL-engine運(yùn)行在單機(jī)節(jié)點(diǎn)中。作者將Wukong（分布式RDF圖存儲(chǔ)數(shù)據(jù)庫(kù)）與Apache Storm（高吞吐量低延遲的實(shí)時(shí)流處理系統(tǒng)相結(jié)合）。

????????作者通過(guò)實(shí)驗(yàn)發(fā)現(xiàn)通過(guò)將實(shí)時(shí)流處理系統(tǒng)和圖處理系統(tǒng)簡(jiǎn)單組合存在高延遲、低吞吐量的問(wèn)題。主要是由于以下的這幾個(gè)原因：
1. 首先，它存在跨系統(tǒng)的開(kāi)銷。因?yàn)樗菍?shí)時(shí)流處理系統(tǒng)和圖儲(chǔ)存系統(tǒng)簡(jiǎn)單結(jié)合，在二個(gè)不同系統(tǒng)之間存在數(shù)據(jù)轉(zhuǎn)換和數(shù)據(jù)傳輸?shù)拈_(kāi)銷。

2. 其次，為了降低跨系統(tǒng)執(zhí)行的次數(shù)，組合設(shè)計(jì)會(huì)改變查詢計(jì)劃。通過(guò)改變查詢條件的執(zhí)行順序，Strom首先將兩個(gè)實(shí)時(shí)流數(shù)據(jù)進(jìn)行join，得到一部分中間結(jié)果。然后將這個(gè)中間結(jié)果發(fā)送到Wukong中去，Wukong進(jìn)行檢索，最終得到查詢結(jié)果。然而，對(duì)于兩個(gè)實(shí)時(shí)流數(shù)據(jù)進(jìn)行join操作，這通常會(huì)造成巨大的冗余數(shù)據(jù)和時(shí)間開(kāi)銷。

5. System? Desgin

????????Wukong + S使用一種新穎的集成設(shè)計(jì)，用于快速演化的鏈接數(shù)據(jù)的狀態(tài)流查詢。Wukong+S在一個(gè)系統(tǒng)中管理實(shí)時(shí)流數(shù)據(jù)和靜態(tài)歷史數(shù)據(jù)。如何正常整合實(shí)時(shí)流數(shù)據(jù)和存儲(chǔ)的數(shù)據(jù)？永恒持久化的數(shù)據(jù)和時(shí)效性很強(qiáng)的數(shù)據(jù)。

????????為了保證一次性查詢，它既查詢存儲(chǔ)數(shù)據(jù)，也可以對(duì)實(shí)數(shù)流數(shù)據(jù)吸收其中永恒數(shù)據(jù)。Wukong+S提出了一個(gè)集中化的設(shè)計(jì)，在這個(gè)設(shè)計(jì)中數(shù)據(jù)被分成兩個(gè)類型。一種是永恒持久化的數(shù)據(jù)，他被存儲(chǔ)在Continuos Persisten Store這樣的一個(gè)結(jié)構(gòu)中它還會(huì)不斷地從實(shí)時(shí)流數(shù)據(jù)中提取一些永恒可以用來(lái)持久化的數(shù)據(jù)（比如Like流和Tweet流）。還有一種數(shù)據(jù)是Timing data 瞬時(shí)實(shí)時(shí)的數(shù)據(jù)。這種數(shù)據(jù)和時(shí)間關(guān)聯(lián)性很大，比如一些GPS 地理位置數(shù)據(jù)。一次性查詢直接查詢永恒的數(shù)據(jù)，因?yàn)檫@種查詢不關(guān)心實(shí)時(shí)性的數(shù)據(jù)。持久化的查詢會(huì)不斷地從Timing Data以及Timeless Data中查詢結(jié)果。

????????流處理系統(tǒng)首先由用戶定義的謂語(yǔ)。將實(shí)時(shí)流數(shù)據(jù)分成兩類分別是Timing數(shù)據(jù)和Timeless數(shù)據(jù)。Timeless數(shù)據(jù)被底層數(shù)據(jù)存儲(chǔ)吸收寫(xiě)入到服務(wù)器引擎的持久化存儲(chǔ)中。Timing數(shù)據(jù)被存儲(chǔ)在Time-based transient store存儲(chǔ)中。

5.1 Hybrid Store

????? ? Wukong+S 對(duì)于這樣的永恒持久化的數(shù)據(jù)和瞬時(shí)實(shí)時(shí)的數(shù)據(jù)采用一種混合存儲(chǔ)的策略。對(duì)于這種永恒持久化的數(shù)據(jù)，Wukong+S采用一種Continuous Persistent Store的結(jié)構(gòu)。持續(xù)持久化存儲(chǔ)不斷從實(shí)時(shí)流數(shù)據(jù)中吸收永恒持久化的數(shù)據(jù)。他的設(shè)計(jì)目標(biāo)就是支持持續(xù)的狀態(tài)流查詢和最新的一次性查詢。

????????對(duì)于這種瞬時(shí)實(shí)時(shí)的數(shù)據(jù)，Wukong+S采用一種Time-based Transient 存儲(chǔ)（基于時(shí)間的瞬時(shí)存儲(chǔ)）。瞬時(shí)數(shù)據(jù)只會(huì)在一個(gè)時(shí)間間隔內(nèi)被相關(guān)的連續(xù)查詢請(qǐng)求訪問(wèn)。它的設(shè)計(jì)目標(biāo)是：對(duì)于一部分瞬時(shí)的數(shù)據(jù)流，它能夠支持快速垃圾收集(GC)。定期掃除過(guò)期的瞬時(shí)數(shù)據(jù)。

具體的細(xì)節(jié)設(shè)計(jì)如下圖所示：

使用混合存儲(chǔ)來(lái)存儲(chǔ)實(shí)時(shí)流數(shù)據(jù)和永恒持久化數(shù)據(jù)，存在以下兩點(diǎn)優(yōu)點(diǎn)：
1.在實(shí)時(shí)數(shù)據(jù)和永恒數(shù)據(jù)之間不存在互相干擾。
2.由于設(shè)計(jì)數(shù)據(jù)存儲(chǔ)分開(kāi)來(lái)存儲(chǔ)，我們可以針對(duì)不同的操作模式進(jìn)行優(yōu)化。

對(duì)于永恒持久化數(shù)據(jù)采用continuous persistent store 對(duì)于實(shí)時(shí)Timed數(shù)據(jù)采用time-based transient store。

6. Wukong+S Architecture

????????數(shù)據(jù)是被分區(qū)存儲(chǔ)在多個(gè)服務(wù)器上的。每個(gè)服務(wù)器引擎都可以提供狀態(tài)流查詢功能和支持?jǐn)?shù)據(jù)被分區(qū)存儲(chǔ)功能。

6.1 Stream Index? ??????

????????那么如何在一定的時(shí)間間隔內(nèi)快速訪問(wèn)持續(xù)持久化存儲(chǔ)？Wukong+S提供了Strem Index機(jī)制，方便查詢根據(jù)時(shí)間戳快速定位到相應(yīng)的KV-Strore相應(yīng)的存儲(chǔ)中。

????? ? Wukong+S采用Stream Index機(jī)制，方便查詢根據(jù)時(shí)間戳快速定為到相應(yīng)的KV-Store相應(yīng)的存儲(chǔ)中。

6.2 Locality-aware partitioning

????????分區(qū)存儲(chǔ)流索引的一般方法是將流索引和數(shù)據(jù)一起放置存儲(chǔ)，這可以為連續(xù)查詢提供數(shù)據(jù)局部性。然而，這種分區(qū)策略會(huì)造成連續(xù)查詢的執(zhí)行被拆分成多個(gè)節(jié)點(diǎn)，即遷移執(zhí)行。然而同實(shí)驗(yàn)表明：這種遷移執(zhí)行的方式，會(huì)造成嚴(yán)重額的網(wǎng)絡(luò)傳輸?shù)拈_(kāi)銷以及額外的調(diào)度延遲。Wukong+S考慮到索引大小一般都比較小，連續(xù)流查詢可以從遠(yuǎn)程的節(jié)點(diǎn)中抓取數(shù)據(jù)，并且在一個(gè)機(jī)器上執(zhí)行查詢?nèi)蝿?wù)。這種策略叫做遷移數(shù)據(jù)。連續(xù)流查詢還需要將Stream Index復(fù)制到執(zhí)行持續(xù)性查詢?nèi)蝿?wù)的機(jī)器上去，這樣執(zhí)行查詢?nèi)蝿?wù)的機(jī)器可以通過(guò)Stream Index快速訪問(wèn)到流數(shù)據(jù)。

7 Consistent Data Snapshot

????????Wukong+S設(shè)計(jì)用于處理大量的查詢請(qǐng)求。在分布式系統(tǒng)的實(shí)現(xiàn)中，面對(duì)大量的并發(fā)查詢請(qǐng)求的時(shí)候，以及不斷演化的Liked數(shù)據(jù)的時(shí)候，如何去提供一個(gè)一致性查詢視圖成為一個(gè)重要的挑戰(zhàn)。

7.1 Decentralized Vector Timestamp （VTS）

Wukong+S 針對(duì)持續(xù)的查詢工作，提出了一個(gè)向量時(shí)間戳的解決方案。

????? ? 因?yàn)椴煌姆?wù)器執(zhí)行實(shí)時(shí)流數(shù)據(jù)插入的速度不一樣，導(dǎo)致系統(tǒng)在執(zhí)行RDF查詢的時(shí)候。系統(tǒng)要給出一個(gè)整個(gè)集群插入數(shù)據(jù)的一致性視圖。VTS：就是用來(lái)做這個(gè)工作的，首先每個(gè)服務(wù)器都有一個(gè)本地Local_VTS,用來(lái)標(biāo)識(shí)本地服務(wù)器已經(jīng)插入批次的位置。然后每個(gè)服務(wù)器將自己本地的Local_VTS發(fā)送給Master協(xié)調(diào)器Coordinator，協(xié)調(diào)器Coordinator維持一個(gè)Stable_VTS服務(wù)器用來(lái)保存全局信息。這個(gè)全局信息可以用來(lái)進(jìn)行Continous Query操作。

7.2 Bounded Snapshot Scalarization

????????VTS的這種解決方案要求與矢量時(shí)間戳相關(guān)聯(lián)的值中的所有流數(shù)據(jù)，由于大量的Key/Value鍵值對(duì)造成大量?jī)?nèi)存的開(kāi)銷，并且無(wú)邊界的數(shù)據(jù)流還會(huì)造成大量的資源浪費(fèi)。因此Wukong+S提出了一種Bounded Snapshot Scalarization的解決方案。

????????Wukong + S通過(guò)將矢量時(shí)間戳（VTS）轉(zhuǎn)換為標(biāo)量快照編號(hào)（SN）來(lái)解決此問(wèn)題。 首先，協(xié)調(diào)員將預(yù)先公布快照號(hào)（SN）與矢量時(shí)間戳（VTS）范圍之間的映射計(jì)劃（即，SN-VTS計(jì)劃），其次，每個(gè)節(jié)點(diǎn)上的工作進(jìn)程確保所有流批次與相同的快照號(hào)碼（SN）連續(xù)存儲(chǔ)在鍵/值對(duì)中; 以這種方式，每個(gè)鍵的快照僅與一個(gè)存儲(chǔ)間隔相關(guān)聯(lián)。 最后，每個(gè)查詢將獲得穩(wěn)定的快照編號(hào)（Sta ble_SN）而不是Stable_VTS，并使用它從持久性存儲(chǔ)中讀取一致的快照。

8.Evaluation

8. Conclusion

????? ? Wukong+S提出了一種分布式流查詢引擎采用新型集成設(shè)計(jì)，實(shí)現(xiàn)快速演進(jìn)鏈接數(shù)據(jù)的狀態(tài)流查詢。實(shí)現(xiàn)超過(guò)每秒100萬(wàn)次查詢的亞毫秒延遲和吞吐量。

總結(jié)

以上是生活随笔為你收集整理的[SOSP 17] Wukong+S : 不断演化的RDF数据的亚毫秒级别的状态流查询的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。