一、背景：

    闊別個人部落格有大半年了，這大半年來我從一個all in flink的角色轉變到了一個兼顧實時流式處理與實時OLAP處理的角色。

    最近由於工作需要，在關注實時的OLTP+OLAP的HTAP場景的資料處理，優先保證低延遲的OLAP查詢。說到這裡，很容易讓人想到Google的F1、Spanner，開源領域的代表TiDB。TiDB是個分散式的MySQL，對OLTP的支援很好，其有一個子專案叫做TiSpark，依賴Spark與TiKV做些OLAP的請求，但是這些複雜SQL執行的優先順序(DistSQL API)是低於OLTP請求的，且當資料量大時(上億條+多表join)，這些SQL執行的時間不是很理想。

    由於我們的需求是同時對流資料以及歷史資料做OLAP查詢，要求是快速的返回結果。Apache Flink等純流式處理框架處理的是實時的資料，如果融入歷史資料，那麼實現起來也不是很方便。最主要的是如果OLAP查詢的維度非常多，且不固定時，例如可以選擇商圈、城市、省份、使用者、時間等維度做聚合，那麼flink去處理的話， 會發現key的選擇很多，實現起來既麻煩也費時。如果選擇druid或者kylin建立cube，那麼由於我們的資料還會有些OLTP的操作，同時實時性也較差，因此也不太適合。

    因此我們注意到一個完全基於記憶體的分散式資料庫(同步或非同步寫到磁碟)：SnappyData，其是一個行、列混和的記憶體分散式資料庫，內部由GemFire(12306的商業版)+Spark SQL(支援列存可壓縮)實現，既支援OLTP，也支援複雜的OLAP請求，且效率很高。

    上邊說了來龍去脈，下面開始針對SnappyData發表的論文，對其進行簡單的介紹。

二、SnappyData介紹

    在網際網路時代，許多場景同時要求事務型操作、分析型操作以及流處理。企業為了應對這些需求，通常搭建各自用途的平臺來分別處理OLTP類的關係型資料庫，以及OLAP的資料倉庫和Streaming流處理框架。在實現OLAP的過程中，已經有很多SQL On Hadoop的技術方案來實現OLAP的查詢了，例如Hive、Impala、Kudu、Spark SQL等。這些系統的一大特點就是資料不可變，例如Spark RDD以及Hive中的資料，雖然各自在批處理的優化上做了很多的努力，但是還是缺乏對事務的ACID的支援。

    流處理框架倒是可以支援流資料的處理，但是如果要關聯大量的歷史資料進行處理，顯然效率也是較低的，且支援複雜的查詢也比較困難。

    那麼為了支援這種混合負載的業務，通常公司都會進行大量的工作，既費時也費力，且效率較低。這些異構的系統雖然可以實現不同的需求，但是卻有以下一些缺點：

1、複雜度和總成本增加：需要引入Hive、Spark、Flink、KV，Kylin、Druid等，且要求開發人員具備這些能力並進行運維。

2、效能低下：一份資料要在不同的系統中儲存和轉換，例如RDBMS中一套、Hive中一套、Druid中一套，資料還得經過Spark、Flink轉換處理。

3、浪費資源：依賴的系統越多，使用的資源也就越多，這個是顯而易見的。

4、一致性的挑戰：分散式系統事務的一致性實現困難，且當流處理失敗時，雖然檢查點可恢復但是有可能重複寫入外部儲存中(sink的exatcly-once沒法保證)。

    我們的方法是將ApacheSpark作為計算引擎與Apache GemFire無縫整合，作為記憶體事務儲存。 通過利用這兩個開源框架的互補功能，SnappyData同時實現了這3種需求。同時，SnapppyData的商業版還通過概率的預算提供對超大歷史資料查詢時近似精確的結果。

    下面我們看看SnappyData的具體實現。

三、方法與挑戰

    為了支援混合型的工作，SnappyData設計成了Spark+GemFire的組合。

    Spark的RDD很高效，但是其畢竟只是個計算引擎，而並不是儲存引擎；同時其要求資料是不可變的。

    GemFire(也叫Geode)，是一個面向行的分散式儲存，可以實現分散式事務的一致性，同時支援點的更新以及批量更新，資料儲存在記憶體中，也可以根據append-log持續的寫入到磁碟中。

    SnappyData則對兩者進行了完美的結合，用Spark作為程式設計模型，資料的可變性和HA的需求則是使用GemFire的複製技術和細粒度的更新技術實現。

    當然在兩者融合上也有些挑戰，最典型的需求就是擴充套件Spark，要其利用GemFire的鎖服務以及可變性的操作，對資料進行點查詢，點更新以及複雜結構上的insert操作。最後就是SnappyData要完全相容Spark。

四、SnappyData架構

    說到這，相信大家對SnappyData能幹什麼有個大體的瞭解，採用與Spark RDD一樣的列存，同時根據GemFire支援行存，以及兩者互補，在行、列儲存上進行點查、點更新以及批量寫入和更新。那麼其架構到底是什麼樣呢？

    從圖中可以看到，SnappyData的儲存主要在記憶體中，同時支援行存與列存，以及預測型的儲存。列存是來源於Spark的RDD，行存則擴充套件了GemFire的表且支援行存上建索引。而且SnappyData還有個AQP的概念，即以近似準確的結果來快速的響應對超大歷史資料的查詢，不過這個功能只能用於商業版上。

    SnappyData支援2種程式設計模型：SQL和Spark API。因此SnappyData是一個SQL資料庫，其可以使用Spark SQL進行開發。SnappyData的副本一致性和點更新則依賴GemFire的P2P對等網路，事務支援則是依賴GemFire中通過Paxos實現的2階段提交實現。

五、混合儲存模型

    SnappyData中的表是可以分片的，也可以在每個節點的記憶體中複製。

    行表會佔用較大的記憶體空間，但是適合較小的表，且可以建立索引。列表也可以被更新，且做了壓縮，減少了對記憶體的壓力。那麼SnappyData是如何對列存進行更新的呢？其使用了一個delta row buffer的區域，當記錄被寫入列表時，首先會進入delta row buffer中，它是一個在記憶體中與更新的列表有相同分割槽策略的行存，這個行存由混合佇列支援，可以週期性的寫入到列存中並清空自己，當連續的對同一記錄進行更新操作時，只會把最終的狀態寫入列存中，即合併的含義，而不是一次一次的寫入。同時，delta row buffer使用了copy-on-write的語義來保證併發更新時的資料一致性。

    SnappyData中將GemFire的表進行分割槽或複製到不同的節點上，根據分割槽資料被放到不同的buckets中，因此訪問資料是並行進行的。這個與Spark讀取外部parquet或Json資料很像，但是SnappyData內部做了優化，例如每個分割槽本身就是列式儲存，避免了資料的複製和序列化開銷，同時允許表與表之間的儲存本地化，即colocate相關聯的資料，例如一個事實表可以根據分割槽鍵colocate其父表配置或者維度資訊表，這樣在做join時可以避免shuffle和跨節點的資料傳輸，效率非常高。

    這裡的colocate的作用，是為了在分散式join時，避免分散式鎖、shuffle的網路開銷。通過colocate操作，將分散式的join變成了本地join，並剪枝了不必要的分割槽。個人覺得這個技術或者思路，也許會應用到未來很多的分散式join的場景中。除了colocate外，還有一種方法是將一個表的資料進行replicated操作，即複製到每個server的記憶體中一份，也可以避免分散式join，但是其要求是個很小的表，否則記憶體壓力太大了。

六、混合叢集管理

    SnappyData作為一個儲存和計算引擎，必須要靠高併發、高可用以及資料的一致性。

    為了快速的檢測失敗，SnappyData依賴於UDP的neighbor ping和TCP的ack超時機制，而一致性則依賴GemFire內部的機制。這裡的具體實現忽略。

七、總結

    SnappyData的效能有很多benchmark，例如早期的測試如下：

    總結起來，SnappyData = stream寫入+ OLTP + OLAP。由於其結合了GemFire與Spark，使得其可以在列存上進行更新；同時由於其資料既存在於記憶體又可刷到磁碟，因此全記憶體的計算使得其速度很快；最後其colocate的設定使得多表join的效能很高。綜上所屬，如果你的業務對OLAP型別的延時要求很低，同時要能夠查詢實時的資料，那麼SnappyData是個非常不錯的選擇。

    我們在SnappyData中具體的使用以及效果，我會在後續的文章中進行介紹。

相關資料：