這是關于Yelp的實時流數(shù)據(jù)基礎設施系列文章的第四篇。這個系列會深度講解我們?nèi)绾斡?ldquo;確保只有一次”的方式把MySQL數(shù)據(jù)庫中的改動實時地以流的方式傳輸出去，我們?nèi)绾巫詣痈櫛砟Ｊ阶兓?、如何處理和轉換流，以及最終如何把這些數(shù)據(jù)存儲到Redshift或Salesforce之類的數(shù)據(jù)倉庫中去。

在2010年時，Yelp開源了一個名叫MRJob的框架，是用來在AWS基礎設施上運行大MapReduce Job的。Yelp的工程師們用MRJob實現(xiàn)了很多功能，從廣告推送到翻譯，比比皆是。事實證明，MRJob是一個非常強大的工具，可以在我們當時豐富的數(shù)據(jù)集合上完成計算和聚集操作。

不幸的是，隨著使用MRJob的服務數(shù)量巨增，運行和調(diào)度任務開始變得越來越復雜。由于很多任務都是要依賴上游任務的，所以就要好好地安排整個系統(tǒng)的拓撲。MapReduce任務并不是用于實時處理的，所以任務的拓撲要每天調(diào)度一次。更糟的是，萬一上游的任務失敗了，下游的也會失敗，最終會輸出錯誤的結果。因此就要有非常專業(yè)的能力來判斷應該從哪個任務開始、以什么順序重新運行，最終輸出正確的結果。

愛思考的人就會問了：我們有沒有什么辦法來更高效地完成計算和轉換任務呢?我們還想支持一個復雜的數(shù)據(jù)流中不同數(shù)據(jù)轉換操作之間的依賴關系，尤其是要能優(yōu)雅地處理模式改變及上游的故障。我們還希望系統(tǒng)能實時或者近實時地運行。這樣，系統(tǒng)就可以用于業(yè)務分析及指標監(jiān)控。換句話說，我們需要的是一個流處理器。

Storm之類現(xiàn)成的計算系統(tǒng)本來也是非常不錯的。但由于許多主流的流處理框架對Python的支持都不太好，因此要把我們的其他后臺程序與Storm或者其他現(xiàn)有流處理系統(tǒng)結合起來就會非常痛苦。

我們***用的是Pyleus，這是一個讓開發(fā)者可以用Python處理和轉換數(shù)據(jù)的開源框架。Pyleus的底層仍然是使用Storm的，構建耗時比較久，運行得也慢。Twitter Heron宣布開源后，我們發(fā)現(xiàn)我們也碰上了許多他們碰到過的問題。Yelp自己有功能非常強大的用于部署服務的Platform-as-a-Service平臺PaasTA，相比之下我們更喜歡使用PaaSTA，而不是運行專用的Storm集群。

從2015年7月開始，有一幫工程師們開始研發(fā)一種新型的數(shù)據(jù)倉庫，也碰上了典型的擴展和性能問題。最開始時他們想用Pyleus來先清洗數(shù)據(jù)，再拷貝到Redshift上。后來他們意識到部署一整套Storm集群來運行些簡單的Python邏輯實在太沒必要了：用Yelp自己的運行服務的平臺去部署一套基于Python的流處理器就足夠了。我們的流處理器是基于Samza設計的，目的是提供一些簡單的接口，用一種“處理消息”的方法來做數(shù)據(jù)轉換。

工程師們在Hackathon 17上構建了運行在PyPy上的流處理器的原型，這樣PassStorm就誕生了。

這名字中有什么含義?

PaaStorm的名字其實是PaaSTA和Storm的組合。那PaaStorm到底是干什么的呢?要回答這個問題，咱們先看看數(shù)據(jù)管道的基本架構：

主要看看“Transformer”那一步，就會知道大多數(shù)存儲在Kafka中的消息都并不能直接被導入目標系統(tǒng)。設想有一套Redshift集群是用來存儲廣告推送數(shù)據(jù)的。廣告推送集群想存儲的只是上游系統(tǒng)的某一個字段(比如某個業(yè)務的平均權重)，否則它就要保存原始數(shù)據(jù)并對其進行聚合計算。如果Redhift廣告推送集群要存儲所有上游數(shù)據(jù)的話，就會浪費存儲空間，導致系統(tǒng)性能降低。

在過去，各個服務都會寫復雜的MapReduce任務，在把數(shù)據(jù)寫到目標數(shù)據(jù)存儲之前先進行數(shù)據(jù)處理?？墒?，這些MapReduce任務都碰到了上文所述的性能和擴展問題。數(shù)據(jù)管道給大家提供的好處之一是消費者程序可以拿到它所需要的數(shù)據(jù)的形式，不管上游數(shù)據(jù)本來是什么樣。

減少示例代碼

本來我們是可以讓每個消費者程序自己按自己需要的方式做數(shù)據(jù)轉換的。比如，廣告推送系統(tǒng)可以自己寫一個轉換服務，從Kafka中的業(yè)務數(shù)據(jù)中提取出查看統(tǒng)計量，并自己維護這個轉換服務的。這種辦法最初工作得很好，但最終系統(tǒng)上規(guī)模時我們就碰上問題了。

我們想提供一個轉換框架是基于以下考慮：

很多轉換邏輯是通用的，可以在多個團隊之間共享。比如把標志位轉換成有意義的字段。

這樣的轉換邏輯通常會需要很多示例代碼。比如連接數(shù)據(jù)源或數(shù)據(jù)目的、保存狀態(tài)、監(jiān)控吞吐量、故障恢復等。這樣的代碼本來并不需要在各種服務之間拷來拷去。

要保證能對數(shù)據(jù)進行實時處理的話，數(shù)據(jù)轉換操作要盡可能地快，要基于流。

減少示例代碼最自然的方式就是提供一個轉換接口。大家的服務實現(xiàn)接口中完成一次轉換操作的具體邏輯，然后，剩下的工作就由我們的流處理框架完成。

把Kafka作為消息總線

最初PaaStorm是一個Kafka-to-Kafka的轉換框架，慢慢地才演進成也支持了其他類型的終端節(jié)點。把Kafka做為PaaStorm的終端節(jié)點簡化了很多東西：每個對數(shù)據(jù)感興趣的服務都可以注冊到Topic上，關注任意轉換過的數(shù)據(jù)或者原始數(shù)據(jù)，有新消息到來就處理就好了，完全不必在意是誰創(chuàng)建了這個Topic。轉換過的數(shù)據(jù)按Kafka的保留策略持久化。因為Kafka是一個發(fā)布-訂閱系統(tǒng)，下游系統(tǒng)也可以在任何它想的時候消費數(shù)據(jù)。

用Storm處理一切

當采用了PaaStorm之后，我們該怎樣把我們的Kafka Topic之間的關系可視化呢?因為有些Topic中的數(shù)據(jù)會按照源到端的方式流向別的Topic，我們可以把我們的拓撲結構當成一個有向無環(huán)圖：

每個節(jié)點都是一個Kafka Topic，箭頭表示PaaStorm提供的轉換操作。這時候“PaaStorm”這個名字就變得更有意義了：象Storm一樣，PaaStorm通過轉換模塊(象Bolt一樣)提供對數(shù)據(jù)流的源(象Spout一樣)的實時轉換。

PaaStorm內(nèi)部機制

PaaStorm的核心抽象叫做Spolt(Spout和Bolt的結合物)。象名字表示的一樣，Spolt接口也定義了兩個重要的東西：一個輸入數(shù)據(jù)源，一種對那個源的消息數(shù)據(jù)進行的某種處理。

下面例子定義了一個最簡單的Spolt：

每個PaaStorm實例都用一個Spolt初始化。比如，下面的命令就用上文中定義的UppercaseNameSpolt開啟了一次處理：

PaaStorm(UppercaseNameSpolt()).start() 

這就意味著所有有意寫一個新轉換器的人都可以簡單地定義一個新的Spolt子類，壓根不用修改任何PaaStorm運行體相關的東西。

從內(nèi)部來看，PaaStorm運行體的主方法也是驚人的簡單，偽碼如下：

這個運行體先做了一些設置：初始化了生產(chǎn)者和消費者，以及消息計數(shù)器。然后，它一直等待上游Topic中的新數(shù)據(jù)。如果有新數(shù)據(jù)到來，就用Spolt處理它。Spolt處理之后會輸出一條或多條消息，生產(chǎn)者再把它發(fā)布到下游的Topic。

另外簡單提一下，PaaStorm運行體也提供了比如消費者注冊、心跳機制(名叫“tick”)等。比如某個Spolt要經(jīng)常性地清空它的內(nèi)容，那就可以用tick來觸發(fā)。

關于狀態(tài)保存

PaaStorm保證可以可靠地從故障中恢復。萬一發(fā)生了崩潰，我們就該從正確的偏移位置開始重新消費。但不幸的是，這個正確的偏移量一般情況下都并不是我們從上游的Topic中消費的***那一條消息。原因是雖然我們已經(jīng)消費了它，但事實上我們還沒來得及把轉換后的版本發(fā)布出去。

所以重新啟動時正確的位置應該是上游Topic與已經(jīng)成功發(fā)布到下游的***一條消息對應的位置。在知道發(fā)到下游的***一條消息的情況之后，我們需要知道它對應的上游的消息是哪一條，這樣就可以從那里恢復了。

為了方便實現(xiàn)這個功能，PaaStorm的Spolt在處理一條原始消息時，會把與這條原始消息相對應的在上游Topic中的Kafka偏移量也加到轉換后的包里。轉換后的消息隨后會在生產(chǎn)者的回調(diào)函數(shù)中把這個偏移量傳回來。這樣，我們就可以知道與下游Topic中***一條消息對應的上游Topic的偏移量了。因為回調(diào)函數(shù)只有在生產(chǎn)者成功地把轉換后的消息發(fā)布出去之后才會調(diào)用，也就意味著原始消息已經(jīng)被成功處理了，在這種情況下，消費者就可以很放心的在那個回調(diào)函數(shù)中提交這個偏移量了。萬一發(fā)生崩潰，我們可以直接從還沒有被完全處理的上游消息那里開始繼續(xù)處理。

從上面的偽碼中可以看到，PaaStorm也會統(tǒng)計消費掉的消息數(shù)和發(fā)布的消息數(shù)。這樣，感興趣的用戶可以檢查上游和下游Topic中的吞吐量。這讓我們很輕松地有了對任意轉換操作的監(jiān)控和性能檢查功能。在Yelp，我們是把我們的統(tǒng)計信息發(fā)給SignalFX的：

SignalFX圖可以顯示出在一個PaaStorm實例中生產(chǎn)者和消費者的吞吐量。在這個例子中，輸入輸出消息量并不匹配。

在PaaStorm中對生產(chǎn)者和消費者分開做統(tǒng)計的好處之一是我們可以把這兩個吞吐量放在一起，看看瓶頸是在哪里。如果到不了這個粒度，是很難發(fā)現(xiàn)管道中的性能問題的。

PaaStorm的未來

PaaStorm提供了兩個東西：一個接口，并實現(xiàn)了一套框架來支持這個接口。盡管我們并不希望PaaStorm的接口很快就被改動，但已經(jīng)有一些孵化項目在計劃解決“轉換并連接”的問題了。在將來，我們希望能把PaaStorm的內(nèi)部換成Kafka Stream或者Apache Beam，主要的障礙是對Python的支持程度如何，我們尤其看重的是對終端節(jié)點的支持?？傊谟虚_源的Python流處理項目成熟之前，我們會一直把PaaStorm用下去。