郭斯杰：重新思考流計算時代的分布式存儲_HDF

導語：本文根據郭斯杰老師于第十屆中國系統架構師大會(SACC2018)的現場演講《從文件存儲，對象存儲到流存儲-重新思考流計算時代的分布式存儲》內容整理而成。

講師介紹：

郭斯杰，Streamlio的聯合創始人之一。Streamlio是一家專注于構建下一代流計算基礎設施的初創公司。他也是ApacheBookKeeper的PMC主席，ApachePulsar的PMC成員。在創立Streamlio之前，他是Twitter消息團隊的技術負責人。

演講正文：

大家好，我今天主要分享的是在流計算時代，關于分布式存儲的一些思考。自我介紹一下，我是一個開源愛好者，我主要參與的兩個開源社區，Pulsar，一個雅虎開源的下一代流數據平臺;然后BookKeeper，一個分布式的日志存儲系統。在這之前我在Hadoop也做了一些跟Hive、HBase相關的工作。

我目前在硅谷的一家創業公司叫Streamlio，顧名思義，Streamlio其實是由三個單詞構成的，就是Stream、ml加IO。所以意圖很明顯，我們主要是做跟流相關的基礎架構，未來我們可能會在基于流的基礎架構之上再去做一些ml相關的事情。和流相關的基礎機構包括消息中間件、存儲、還有計算。在加入Streamlio之前，我在Twitter待了五年，主要負責整個Twitter的消息中間件平臺以及數據庫復制，包括跨機房復制相關的基礎設施。在Twitter之前的話，我是在雅虎做了三年多的時間，然后是華中科大畢業的，后來在中科院計算所讀的研究生。

一、背景介紹

我的題目其實是關于“實時時代”，或者說“流計算時代”，對于分布式系統我們應該怎么去考量?我們先看一下，八九十年代，我們最開始存儲數據無外乎兩種辦法，一個是用數據庫，另外一種就是以文件的方式進行存儲。當你想去存一些數據時，基本上考量的就是這兩個事情。后來慢慢演化出了很多分布式數據庫、分布式文件系統。隨著互聯網的到來，數據量越來越大，沒辦法用傳統的技術、本地文件系統、或者是網絡文件系統去存大量數據，沒有辦法在原來的Oracle里面去存數據。因此就演化出了所謂的“神一般的存在”，就是Hadoop。其實Hadoop的整個生態的話是起源于Google的GFS和MapReduce這套系統，基本上現在大家說大數據，開源社區就一個東西，Hadoop，基本上所有互聯網公司都會有一套Hadoop的系統。

整個Hadoop的生態，其實是為批量大數據而存在的。隨著云時代的到來，像AWS開始把Hadoop放到云上去，作為產品去賣，然后google也會有各種類似的一系列產品，這其實就是為批量以及大數據產生的一套自己的生態，而這個生態的基礎，基本上是圍繞分布式文件系統去做的存儲。

這是批量時代。批時代的到來，很多其實是Google主導，它主要是搜索引擎的應用，對時延的要求性不是很高，可以15分鐘跑一個任務，去全網抓一些網頁進行索引。

但是隨著一些新興企業的產生，包括比如說像Twitter、Facebook這些社交媒體，會有大量實時產生的數據，就包括用戶發的推特，然后你投遞的廣告以及移動互聯網的興起，會有大量移動端的數據，這個數據的到來就產生了另外一個概念叫做“流”。所謂“流”，它的誕生基本上是因為，雖然我有Hadoop，但是我的時延通常是幾分鐘甚至小時以上的。

美聯儲巴爾金：傾向于支持美聯儲加息25個基點:2月17日消息，美聯儲巴爾金稱他傾向于支持美聯儲加息25個基點；由于季節性調整問題，最近的就業增長、零售銷售數據其實并沒有帶來太多信號；感覺美國在控制通貨膨脹方面進展“緩慢”；過度儲蓄、財政支出、雇主留住工人的想法可能會抵消美聯儲加息的影響；控制通貨膨脹將需要更多的加息，至于要加多少，需要后續評估；曾在上次貨幣政策會議上支持美聯儲加息25個基點；仍然沒準備好宣布抗通脹取得勝利；希望看到通脹回落至2%這一通脹目標。[2023/2/18 12:14:10]

這個是因為在批量時代，它整個計算模型就是一個批處理的模型，所有的數據采集都是通過一個批量的任務，比如說每5分鐘、每15分鐘寫成一堆文件，拿著這個文件提交給計算引擎去處理。隨著應用發生變化，這時候你需要處理的響應時間會變得更長。所以在這個情況下，誕生了Storm+Kafka這種特殊的模式。這種模式最開始變得流行，是由Twitter帶起來的。因為Twitter整個的產品定位，相當于每個人都是往Twitter上發消息，然后每個人都從Twitter上接受消息，整個的消息的投遞、發送都是實時的，整個Twitter的業務是圍繞“實時”構建的。

所以在這種情況下，誕生了一個所謂的——基本上大家在做流的時候，不可避免地會認識到的一個東西，叫Storm。因為Storm是一個流計算的框架，但是你的存儲本質上還是文件，也就是說我的數據的采集還是以5分鐘、15分鐘進來。但它不能真正意義上發揮流計算的潛質，所以在這種情況下，你需要的是一個新型的面向流的存儲系統。

在那個情況下，基本上大家都是用消息中間件，因此也是在這樣的背景下，Kafka與Storm很天然的一個整合，讓整個系統變成了流計算的一個事實規范。所以在這種情況下，就衍生出了后面更多的一些流計算框架，比如說Flink，以及Hadoop演化出來的Spark，Spark再繼續做SparkStreaming，最后整個計算生態就變得百花爭艷。

但是這種情況下，假如我有一個批處理的數據，還有一個流處理的數據，這時候我們會發現在整個公司內部的數據架構，其實變成了兩個部分，兩個相互隔離的數據孤島。其中一個數據孤島是流數據孤島(StreamingDataSilo)，就是說完全是為了StreamingData去做的一套架構，另外一個數據孤島是批量數據孤島(BatchDataSilo)，也就是為了批量計算構建的一套獨立系統。StreamingData的話，你通常可以看到你需要一個消息中間件，類似于Kafka或者ACTIVEMQ，然后在上面可以跑流計算的一個引擎。在Batch的話，基本上還HDFS或者其他的分布式文件系統;在云上的話，可能就是對象存儲如S3，GCS等類似的對象存儲，這上面你可能需要有MapReduce或者是一個類似Hive的數據倉庫。但它其實是相互隔離的一個狀態。

這個隔離會帶來一個問題，你的兩個數據其實是相互分離的。分離帶來的問題是，我需要學兩套API，需要兩套集群去存放數據，然后圍繞這個生態圈，需要兩套工具去管理這些數據。

怎么去解決出現的情況，在Twitter最開始提出的一個架構叫Lambda，意思是說，我有兩層，一層是批處理層，就跑我的全量數據，包括MapReduce、包括Spark，但是因為它是批處理，時延是相對比較大的。這時我為了快，再加一個加速層。加速層的是怎么工作的?我的所有數據是通過消息中心進來，基本上是以流的方式進到計算引擎，然后拿一個流處理的計算引擎去做處理，處理完了以后，產生的是一個增量的結果，然后把這兩個結果做整合，提供給最終的用戶。

數據：Lens Protocol總用戶數超10萬:據官方數據，Web3社交圖譜協議Lens Protocol總用戶數已達107,672個，總發帖數（post）達839,069，總評論數達319,861。[2022/12/28 22:12:04]

所以Lambda的架構，其實只是把這兩個DataSilo在最終的結果端做了一個匯總，讓用戶看起來是一個統一體。但實際上，你最開始的數據進來，會被分流到兩個不同的系統里面去，一個是所謂的消息中間件，一個是文件系統或者對象存儲。你整個業務的代價、機房的開銷、硬件設施的代價、維護人員代價，其實都是雙倍的。所以在這種情況下，我們發現Lambda會變得越來越重，因為隨著數據量變得越來越大，本質上是沒法使用Lambda支撐更大的業務。

所以在這種情況下，就有人提出了Kappa的概念。它的概念其實很簡單，就是所有的數據都以日志的方式存進來，因為日志它其實是一個append-only的存儲。存進來了以后，日志本身上它具有流的一個特性，然后所有的訪問都是以追尾讀的方式去讀，所以它可以做到消息中間間的低時延。那就可以把所有數據存起來，當你需要回放歷史數據時，還可以做相應的批處理。

在計算引擎上，因為你的數據表征變成一份了，你的計算引擎、API就可以統一起來，所以這也是后來為什么那么多計算框架在做批、流一體。比如Flink既可以支持批，也可以支持流。其實都在做一個事情，就是要讓API在處理批跟處理流上是統一的，只需要學一個計算框架、一套API，就能做原來的兩件事情。

Kappa這個架構是很好的一個想法，但要真正做到真正意義上的Kappa，你不能拿一個消息中心間去做一個Kappa。因為消息中間件基本上是為消息，就是你的尾端數據去做設計的，所以它基本上沒有考慮存儲方面的一些特性。所以我們認為在批、流一體的時代，當你需要去做批、流一體化的計算時，其實是需要一個真正意義上的流存儲。

流存儲的意思就是，你的數據可以通過流的一個或者append-onlylog的方式去表征。你最新的數據被append到log里面以后，它其實是作為我們通常意義上說的流計算的尾端數據進行處理，但你的數據被append之后會慢慢累加，就會變成你的歷史數據。當你需要去做全量計算，需要回溯到7天前、30天前這樣一個處理歷史數據的時候，你可以進行相應的回溯。所以我們認為Kappa是你做批、流一體的一個正確基礎，但是我們需要為Kappa去做一個真正意義上的流存儲。所以我們streamlio大部分創始團隊都是來自于Twitter、雅虎原來做消息中間件跟流計算的，所以我們一直在琢磨一個事情，就是說我們怎么去做真正的流存儲。

二、關于ApachePulsar

接下來，主要分享我們正在做的一些工作，然后我會回過頭來怎么看我們做的工作是怎么映射到現在針對于批、流一體的計算方式下的一個考量。

我們先來看一個特點，什么叫流存儲?流存儲，它其實兩個單詞，一個stream，一個是storage。Stream代表了它的訪問模式，就是在數據寫入以后，我的消費者或者我的讀者，是立馬可見的，不需要等經過比如5分鐘的批處理，等文件全部寫完了以后，才能提交到執行引擎去處理。通常stream的一個接口，就是傳統意義上的消費訂閱模式，就是說我的數據生產到這個stream里面之后，我的訂閱者、消費者就立馬可見。

杭州拱墅區成功發放首批數字人民幣補貼76.8萬元:金色財經消息，近日，浙江省杭州市拱墅區市場監管局結合省、市經濟穩進提質一攬子政策，拱墅區成為全市首個爭取到面向消費者發放財政補貼的城區。該局聯合交警拱墅大隊、區金融辦等部門做好前期調研摸底，牽頭制定《拱墅區電動自行車“促消費穩增長”數字人民幣補貼實施方案》，以數字人民幣的形式為參加提前淘汰置換活動的備案非標電動車主發放補貼。自六月啟動補貼政策以來，已通過工、農、中、建、交、郵儲銀行為2560名非標車主發放數幣補貼76.8萬元。（中國市場監管新聞網）[2022/7/22 2:31:56]

Storage的意思是跟傳統意義上storage類似，它本質上是一個分布式存儲，它需要保證我能夠可靠穩定地保存數據，不丟數據，而且無論我什么時候再去訪問這個數據，我的架構上都有能力把這個數據給存下來。因為我們剛才一直在說的就是所謂的大數據，有大量的數據要算，所以這個東西必須是水平可擴展的，并且是高吞吐的。

所以我們圍繞一個開源的項目叫ApachePulsar來做這個事情。Pulsar是什么呢?其實最簡單的、最容易理解的是，我們通常來說這個事情的話，我們會把它放到消息隊列里面，就跟大家知道的Kafka、ACTIVEMQ、RabbitMQ都是在一個space里面，但是它不同的地方是什么?就是說Pulsar在整個設計上，并不只是一個簡簡單單的消息隊列，我們通常用一句話來概括Pulsar是什么，“FlexiblePub/Submessagingbackedbydurablelog/streamstorage”，它其實有兩層意思，就是說你是一個面向消息或者面向流的消息系統，但是這個消息系統可以提供低延時地消費你的數據的能力，但是你的底層是以一個這種持久化的append-only的日志或者流的存儲，作為分布式存儲的一個支撐。

在解釋這個事情之前，我大概介紹一下這個項目。這個項目是在2012年雅虎內部啟動，然后經過了無數的迭代以后，在2016年9月把它貢獻、開源出來。在開源不到一年之后，去年6月份，把它捐獻給Apache軟件基金會，今年的9月它正式成為頂級項目。在雅虎的Github上我們做了22個releases，然后把Pulsar捐獻給軟件基金會之后不到一年的時間，我們做了9個發布，基本上是不到一個半月有一次發布，目前是相對比較活躍的一個項目。

Pulsar與傳統消息系統的不同

那我說了那么多，它跟傳統的消息系統不同的地方是什么?首先是因為是消息系統，那不可避免地要跟傳統的消息系統做對比。它在應用層面上其實是做了一個數據消費模型的統一，既支持傳統的隊列，也支持高性能的流的處理。這是從API的角度來說。但我們其實更關注的是分布式存儲，它跟傳統的消息系統不一樣的地方是，傳統的消息系統其實是面對消息數據去做的，所以它其實沒所謂存儲的概念。但是在Pulsar里面，我們其實是把存儲跟計算分離，但計算更多的是強調messaging的那一塊。

我們把這兩個分離了以后，同時把傳統消息中間件使用的一個物理分區的模型，變成了一個更多分布式系統用的分片模型，那這樣的話可以打造出一個既有實時消息系統的streaming的一個API，然后你又能夠有一層是可提供無限存儲的一個存儲系統。

靈活統一的消息模型：隊列+流

未知用戶在一小時內購買價值17億美元的以太坊期貨合約，但比特幣新手仍處于虧損狀態:7月20日消息，CryptoQuant首席執行官Ki Young Ju的子賬戶Kate Young Ju發推表示，一個未知實體在一小時內購買了價值17億美元的以太坊期貨合約，這是七個月以來最大的每小時交易量記錄。此外Ki Young Ju指出，雖然最近市場有所反彈，但大部分比特幣投資者仍處于虧損狀態，目前占已實現市值的大部分比特幣新來者仍然處于虧損狀態。（The Crypto Basic）[2022/7/20 2:25:34]

我簡單說一下隊列加流的模型，基本上在流計算里面都不可避免要用消息中間件，那消息中間件無外乎就是生產者、消費者，然后中間加一個topic，或者是consumergroup和subscription這樣的概念。

不一樣的地方是，像Kafka這樣的一個消息中間件，它更多的是面向流去做設計的。像傳統的ACTIVEMQ以及阿里的RocketMQ，更多的是面向隊列去做設計，這時候Pulsar其實是在這上面做了一個統一。統一的意思是，我的生產者都往一個地方去發消息，這個消息的載體叫topic，但是我允許在topic上有不同的訂閱模式。所謂的訂閱模式，它處理的場景是不一樣的，可以是順序消費的streaming的模式，可以是共享消費的隊列模式。

舉三個簡單的例子，第一個我們叫獨占式訂閱，就是對于一個topic的一個流，只有一個消費者去消費，那么所有的消費都是順序的，這時候我可以做很高的吞吐。

另外一種災備式訂閱，其實是獨占式的一個延伸，相當于有兩個消費者或多個消費者，都想消費這個topic，但是在某一個時間段只有一個活躍的消費者，這個消費者可以源源不斷的去消費，其他人只是他的一個災備。如果活躍的消費者down掉了以后，其他消費者就可以立馬接管，保證整個的處理是相對特別快的。

另外一個方式，更多的是在線業務、訂單業務、通知系統里面會用的共享式訂閱，就是說對同一個topic，我可能不需要不在乎序，這時候我可以加無限個消費者去擴展消費能力，這些消息以共享的方式分發給不同的消費者。

這是Pulsar在所謂的API模型上做的一個統一，但這只是給大家一個簡單的一個概念，最主要的地方是，它其實是把傳統的消息系統跟傳統的分布式文件系統或者分布式存儲系統做了一個整合。

存儲與計算分離

另外提出的一個概念其實很簡單，就是存儲跟計算分離。分離的意思是說我有兩層，一層是存儲層，可以無限擴容，存很多數據;有一層是所謂的消息層，就是傳統意義上的Broker，我可以提供很多低時延的消息投遞，還有讀尾端的數據。

分層以后帶來的一個好處是，每一層都可以獨立擴展，比如當我需要更大的存儲的時候，我可以只加存儲節點。當我需要更多的消費者，需要更多人去讀這個數據的時候，只需要加我的更多Broker就行了。分層以后還有一個好處是，我的Broker變成了沒有狀態，沒有狀態了以后，做一些容錯、擴容就會變得特別簡單，不需要搬很多的數據。

在存儲層，其實是用了另外一個項目叫ApacheBookKeeper。它本質上一個專門針對分布式日志的存儲系統。它的定位很簡單，就是一個強一致的系統，是一個刷盤落盤的持久化存儲，能保證即使落盤，我也能做到低延時跟高吞吐。它的設計會在可用性上做很多的文章，保證一些讀寫高可用。給大家一個大概的概念，BK可以做什么?BK最開始的一個主要應用場景是HDFSNameNode，它做的是NameNode的HA，就是說NameNode會有一個addedlog，它其實是整個HDFS的靈魂，就是說你所有對HDFS的Metadata的修改，都會落到addedlog里面進去。BookKeeper最開始出現就是為了解決HDFSNameNodeHA的問題，但后來慢慢就衍生成了一個通用的分布式日志存儲系統。

DeFi協議Shell Protocol宣布將發行治理Token SHELL:7月5日消息，DeFi 協議Shell Protocol在其社交網站宣布，協議確認將發行治理Token SHELL。

此前該協議曾于Arbitrum網絡發行Toucan NFT，并將此NFT系列用于社區治理投票。[2022/7/5 1:51:35]

它最主要的兩個應用場景，我們概括出來，一個的是數據庫，另外一個就是消息或者是流。BookKeeper用在數據庫，主要介紹兩個，一個是Twitter內部有KeyValue存儲Manhattan，BookKeeper作為Manhattan的transactionlog去實現ManhattanKeyValue的強一致性。另外一個是，Salesforce拿Salesforce來做一個類似于AmazonAurora的一個NewSQL的數據庫。所以這就是BookKeeper在數據庫的應用場景，我們可以注意到BookKeeper在這里面其實是整個數據庫的transactionlog，基本上就是append-only的工作負載。

另外一個場景是消息，基本上Twitter跟雅虎都是拿BookKeeper作為整個消息平臺的存儲。消息基本上也是append-only的工作負載，我之所以一直強調append-only是因為，BookKeeper的整個設計，就是為了日志或者為流誕生的。

分片存儲

這給大家做一個簡單的介紹，Pulsar的底層用的是一個面向流和面向日志的一個分布式存儲系統，Pulsar圍繞BookKeeper去構建消息中間件，或者是我們叫流存儲的一個概念的話，它在傳統意義上的這種topic分區上做了一個概念，就是說我的分區不再是一個物理的分區，其實是一個邏輯上的分區。

邏輯上的分區的意思是，我可以無窮無盡地往分區上追加數據、寫數據。但是在底層，我會把邏輯上的分區切成不同的分片，分片會均勻打散，存儲到底層的一個存儲系統里面。基本上HDFS或者其他的分布式文件系統都是這樣的一個思路，所以它本質上就是用分布式存儲來做消息系統，或者是做流存儲的一個思路。

我們剛才說了分層跟分片的概念，為什么我們說分層分片特別重要，它其實解決了很多可用性、容錯、包括運維過程中的一些問題。我舉兩個例子，一個是關于容錯，一個是關于擴容。

容錯方面的話，因為它其實是分成兩層了，有一層是Broker，另外一層是存儲節點。Broker失效了以后，其實它會很簡單，因為記得所有的數據其實存在存儲節點上，Broker本身是沒有任何狀態沒有存儲任何數據的，它只擁有一個分區的所有權。所有權的意思是，我可以提供分區的一個寫服務，當它失效的時候，我只需要把所有權從失效的Broker換到其他在線的一些Broker就可以了，你的整個的流量、你的消費者、消費的數據都可以轉到新的Broker上繼續處理，所以基本上，基本在毫秒級別就可以完成。

BookKeeper的容錯上，其實跟傳統的分布式文件系統是類似的，因為是計算跟存儲分離了以后，你的存儲基本上是被Broker層給隱藏了，所有的生產者、消費者都是跟Broker打交道，所以當你一個存儲節點掛掉的時候，你的應用端其實是沒有任何感知的，而整個失效恢復都是由存儲節點在后臺進完成。

舉個最簡單的例子(如下圖)，在Bookie2上，就是第二個存儲節點上，我的Segment4掛了或者整個第二個節點掛掉了以后，我Segment4的副本數據從3變成2，為了保證數據可用性，需要把副本的數量從2調回到3。這時候后臺有一個自動恢復機制，從第三個存儲節點跟第四個存儲節點，把Segment4的數據重新復制到第一個存儲節點上，從而保證數據有副本。所以這個是Bookie的容錯，它整個容錯會使你的應用端變的相對更能容忍失效的發生。

另外一個好處是，擴容會變得特別簡單，相當于只需要追加新的存儲節點。追加新的存儲節點的話，所有的Broker就能發現存儲節點被新加進來了，當我需要繼續寫新的數據的時候，我會開新的Segment，新的Segment就會自然而然地落到新的存儲節點，整個擴容就會自動、平滑、均勻地擴散到整個集群里面。這就是分層架構帶來的一些容錯和運維上的好處。

下面是一個示意圖，傳統的消息中間件基本上都是基于物理分區去做的，因為它其實不是為了所謂的流存儲去做設計的，它的設計理念就是，我只需要保證我的數據存儲在一段時間內，我的消息消費完了就可以了。物理分區帶來的問題是，數據跟存儲節點是強綁定的，對于一個分區，是沒辦法增長到比存儲節點更大容量的，這時候其實不適合用來做流存儲或者是長遠的一個存儲。

另外一個問題是，如果要做容錯恢復，或者說做擴容的時候，會涉及到一個問題，因為我的物理分區其實是跟存儲節點強綁定，強綁定帶來問題是，如果要做負載均衡，要容錯，這時候需要把物理分區重新拷貝到新的存儲節點上，整個拷貝的過程其實是一個特別消耗帶寬的過程。

但是Pulsar的做法不一樣，它把物理分區變成了一個邏輯分區。邏輯分區以后，整個分區的容量不再受限于單臺機器，而是整個集群，這樣的話你能做到真正意義上的流存儲。邏輯分區以后，其實是把存儲跟計算相互分離，這時候你的失效的處理會變得相對快速，然后沒有痛點，然后你在每一層都可以做獨立的擴展。

舉一個最簡單的例子或更形象的一個例子，大家可以看下面這兩張圖，就是說你分區模型的話，所有分區的數據只能落在某一個存儲節點上，你只能在這個節點上無限增長，但是它會有個頂，到了那個頂了以后，沒有辦法繼續保存你的數據了。但是這種邏輯分區和分片的模型，其實是把你的數據切成不同的小的粒度，均勻打散到整個存儲空間里面進去。這時候你其實能做到真正意義上的流存儲，因為可以無窮無盡地存數據流。

為什么要引入流存儲?

存儲其實都是有一定限度的，而且很多情況下的話，已經有了HDFS，我在云上已經有太多的對象存儲，有太多的文件系統存儲，這時候我這些系統已經跑得好好的了，為什么還要引入一個新的系統?這個時候我們其實在想，我們做流存儲的真正目的，并不是要再造一個存儲系統，而是說我們要在整個數據的抽象上能做到統一，通過流的方式既能表征我的實時數據，也能表征我的歷史數據，那這個時候我們完全可以利用已有的云存儲，或者更廉價的存儲去存已經變老的一些數據。

因為我們其實是一個分片模型，分片模型帶來的好處是，能夠很自然地知道我的數據什么時候變老，當數據變老的時候，可以很容易地把整個數據卸載到廉價存儲里面。但是從應用端來看，整個的數據還是以流的接口、流的表征方式去表達。你在編寫計算的時候，不用去區分，你需要訪問的是到消息隊列里面去算實時計算，還是到HDFS里面去算實時計算，只需要通過統一的一個API就可以做批、流一體的計算，所以這就是我們圍繞Pulsar，圍繞流的模型以及分片模型做的一個事情。

因為我們覺得在批、流一體的計算時代，你的數據如果還是以不同的方式去表征的話，其實還是存在這種所謂的冗余、數據拷貝、數據重復，然后兩個數據源之間不一致的問題，通過流的方式可以統一地表達相應的數據。

三、例子-交互式查詢

做流存儲有什么好處呢?舉一個最簡單的例子，就是我們在Pulsar上面做的另外一個事情叫交互式查詢。有時候很多應用場景，不僅僅只是想查7天前或者1天前的數據，可能是想把我7天前的數據，再加上這1小時的數據一起查，這個時候業務系統很難辦，因為你7天前的數據在HDFS里面，你這1小時的數據在你的Kafka或者是各種MQ里面，你沒辦法把兩個數據串在一起查。

但是通過流存儲的方式，我們其實可以很容易做到這一點。因為你整個流表征的是，歷史數據跟實時數據，其實在統一的一個數據抽象里面，你的數據只要進到流里面，就是可以查的。這種交互式查詢，具有傳統意義上，像Hive這種交互式SQL能帶來的一個復雜性，就是我可以寫很復雜的SQL的查詢，然后我同時還具有streamingSQL的實時性，但它不能完全等同streamingcircle，更多的是說我編寫了一條SQL，我讓它不斷的在那跑，然后它吐結果。交互式查詢更多的是說，我一旦有一個需求，需要查我的歷史數據加實時數據的時候，這個時候叫交互式的查詢。

我們在Pulsar里面做的一個事情叫，我們沒有再去造一個SQL的輪子，而是用Facebook的Presto做了一個深度的開發。為什么我們用Presto，因為它本質上是一個純的SQL查詢引擎，它不帶自己的存儲節點，但它有自己的一個runtime，所以你可以運行相應的一個SQL查詢。而且它可以查詢不同的數據源，就是說如果你的業務是在已經在HDFS，你不想動了，但是你新的業務跑在Pulsar上，然后你想把這兩部分的數據進行一個join或者是關聯查詢，這時候Pulsar可以支持不同的數據源進行查詢，這時候不需要你把所有的數據都導到Pulsar里面再進行一個查詢。

它的實現其實相對特別簡單，就像我剛才說的，整個的數據它是一個流，流被切成了不同的分段，不同的分段其實存儲在不同的存儲節點上，即使最尾端的分段，其實只要數據寫進來了，就可以讀。所以當你做SQL查詢的時候，你的訪問不再是受限于分區的數量，就是說，不再是當只有五個分區時，并發度只能是五。因為整個的分區是個邏輯分區，你的分片可能是一百個分片，這時候如果一百個存儲節點，甚至是一百個執行節點，并發度就可以是一百，整個SQL查詢的執行效率就會特別高，因為它是以分片為粒度，而不再是以分區為粒度。

所以就像上圖展示的，它實現的是一個多對多的數據訪問，這個數據訪問是因為Pulsar本質上是一個分布式的存儲系統，它是一個分片存儲，所以你整個的Presto的worker可以并發訪問同一個分區的不同分片。我們其實在整個流里面第一次去加了時間索引，根據你的寫入時間，就PublishTime快速定位分片，我們可以根據PublishTime快速做定位，然后我們就能知道要讀哪些分片，不需要進行全掃描。

四、總結

總結一下我的分享，我們發現，在計算趨于批、流一體化的這種大浪潮下，所有的Spark、Flink、Beam……的API其實都是批、流一體。在這個時候，我們發現你的數據表征其實還是分散的，就是說你的實時數據、流數據還是以消息的方式去表征。你的歷史數據是以文件系統或者是對象存儲去表征，但是你會發現這部分數據它其實是同一部分數據。就是說你的消息跟你最后的文件系統，它們其實表彰的是同一部分數據。這個數據它只是一枚硬幣的兩面，你的流計算，是處理硬幣的一面，然后你的批計算是處理硬幣的另外一面。其實本質上在數據表征態的話，它不應該分開。流是原始數據最自然的一種表征，因為你所有的數據流進來，會不斷的追加到流里面，其實不需要進行任何的一個轉換，所以流是能夠很自然地把所有的實時數據跟歷史數據都保存下來。我們認為在實時時代，你需要有一個流存儲作為數據的分布式存儲，然后去實現真正意義上的批、流的一體化。

以上是我今天的分享，當然可能大家會有不認同流存儲的概念，歡迎提意見。

Tags：HDFSHDFAPIHDFS幣是什么幣HDF價格HDF幣API價格API幣

BNB