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

 

當你的系統(tǒng)每天要實時從MySQL到Kafka發(fā)布幾十億條消息時，你會怎么管理這些數(shù)據(jù)的模式信息呢?當你的系統(tǒng)要接入幾百個服務(wù)時，你就要處理幾千種不同的模式，手工管理是不可行的。必須有自動化的方案來處理從上游數(shù)據(jù)源到所有下游消費者的模式改變問題。Confluent公司的Schema Registry和Kafka Connect都是不錯的選擇，可惜當我們開始構(gòu)建Yelp數(shù)據(jù)管道時它們還沒發(fā)布。因此就有了我們的Schematizer。

 

 

Yelp數(shù)據(jù)管道的一個重要設(shè)計就是將所有數(shù)據(jù)都模式化，也就是說，所有流經(jīng)數(shù)據(jù)管道的數(shù)據(jù)都必須遵守某種預先定義好的模式，而不是格式隨意的。為什么非要強調(diào)這一點呢?因為我們想讓所有的數(shù)據(jù)消費者都可以對他們要獲取的數(shù)據(jù)格式有預期，因此可以在上游數(shù)據(jù)生產(chǎn)者決定改變他們發(fā)布的數(shù)據(jù)模式時，不會對下游造成非常大的影響。統(tǒng)一的模式表現(xiàn)也讓Yelp數(shù)據(jù)管道可以輕松地整合各種使用不同數(shù)據(jù)格式的系統(tǒng)。

 

Schematizer是用于跟蹤和管理所有數(shù)據(jù)管道中用到的模式，并且提供自動化文檔支持等功能的模式存儲服務(wù)。我們使用Apache Avro來表達模式。Avro有許多我們在數(shù)據(jù)管道中需要的功能，尤其是模式演進，它是解耦數(shù)據(jù)生產(chǎn)者和消費者的關(guān)鍵因素之一。每一條流經(jīng)數(shù)據(jù)管道的消息都用Avro模式序列化過。為了減小消息體積，我們沒有把全部模式信息都放在消息里，而只是帶上了模式的ID。數(shù)據(jù)消費者可以用ID來在運行時從Schematizer中獲取模式信息并將消息反序列化。Schematizer是所有預定義的模式信息的唯一可靠來源。

 

 

Schematizer用兩種方法組織和管理模式：從數(shù)據(jù)生產(chǎn)者的角度和數(shù)據(jù)消費者的角度。

 

第一種方法根據(jù)數(shù)據(jù)的產(chǎn)生信息來將模式分組，每個組由名字空間和數(shù)據(jù)源來定義。生產(chǎn)者在向Schematizer注冊模式時必須提供名字空間和數(shù)據(jù)源信息。比如一個準備向數(shù)據(jù)管道發(fā)布數(shù)據(jù)庫數(shù)據(jù)的服務(wù)，它就可以把服務(wù)名作為名字空間，把表名作為數(shù)據(jù)源。

根據(jù)名字空間和數(shù)據(jù)源來將模式分組

 

第二種方法按數(shù)據(jù)的目的方信息來分組。比如Redshift集群或者MySQL數(shù)據(jù)庫都是數(shù)據(jù)目的方，它們會對應(yīng)一個或多個數(shù)據(jù)生產(chǎn)者，每個數(shù)據(jù)生產(chǎn)者又會關(guān)聯(lián)一個或多個模式，這就對應(yīng)著第一種方法中定義的名字空間和數(shù)據(jù)源。

根據(jù)單個數(shù)據(jù)目的方來將模式分組

 

這兩種方法讓我們可以按不同的需要來檢索和相關(guān)的模式。比如，一個程序可能想知道它會向哪些Topic發(fā)布數(shù)據(jù)，另一個服務(wù)又想知道它的Redshift集群中的數(shù)據(jù)都來自哪里。

 

數(shù)據(jù)管道要求所有發(fā)布到其中的數(shù)據(jù)都必須用預定義的Avro模式進行模式化和序列化。因此，當一個數(shù)據(jù)生產(chǎn)者準備向數(shù)據(jù)管道發(fā)布數(shù)據(jù)時，它要做的第一件事就是向Schematizer注冊模式，最通用的辦法就是直接注冊一個Avro模式。

 

對于沒有或者無法創(chuàng)建Avro模式的數(shù)據(jù)生產(chǎn)者，也可以向Schematizer中加入模式轉(zhuǎn)換器來把非Avro模式轉(zhuǎn)換成Avro模式。MySQLStreamer就是一個代表，它是一個把MySQL數(shù)據(jù)庫中的數(shù)據(jù)發(fā)布到數(shù)據(jù)管道的服務(wù)，它只知道MySQL表模式。Schematizer可以把MySQL表模式定義轉(zhuǎn)換成相應(yīng)的Avro模式。但如果數(shù)據(jù)生產(chǎn)者改變了模式定義的話，它必須重新注冊。

 

 

所有數(shù)據(jù)管道服務(wù)都不能回避的共同痛點就是該如何應(yīng)對上游模式改變。通常這都需要許多在上游生產(chǎn)者和下游消費者之間的溝通和協(xié)調(diào)工作。Yelp也不能免俗。我們也有批量任務(wù)和系統(tǒng)，它們要處理別的批量任務(wù)和系統(tǒng)產(chǎn)生的數(shù)據(jù)。每一次上游的模式改變都是非常痛苦的，它可能導致下游服務(wù)崩潰，整個處理過程都是非常耗費人力的。

 

我們通過模式兼容性來解決這個問題。在模式注冊過程中，Schematizer會根據(jù)模式兼容性來決定Topic和新模式之間的對應(yīng)關(guān)系。只有兼容的模式才能延用舊的Topic。如果有不兼容模式注冊上來，Schematizer會用相同的名字空間和數(shù)據(jù)源來為新模式注冊一個新的Topic。那Schematizer又怎么確定兼容性呢?答案就是Avro解釋規(guī)則(Avro resolution rules)。Avro解釋規(guī)則保證在相同的Topic中，用新版模式打包的消息可以按舊版模式解包，反之亦然。

不兼容的模式會分配不同的Topic

 

目前Yelp數(shù)據(jù)管道中大部分數(shù)據(jù)都產(chǎn)生自MySQLStreamer。比如我們想為某業(yè)務(wù)表增加一個字段，MySQLStreamer就會向Schematizer注冊新模式。因為按照Avro解釋規(guī)則這樣的改動是兼容的，所以Schematizer會創(chuàng)建新Avro模式，并把這個名字空間和數(shù)據(jù)源對應(yīng)的舊的Topic分配給它?？扇绻窍氚涯匙侄螐膇nt改成varchar，那這就是一個不兼容的改動了，Schematizer會為新模式創(chuàng)建一個新Topic。

 

保證了在Topic內(nèi)部的模式兼容性，下游數(shù)據(jù)消費者就可以放心的用舊模式去處理這個Topic中的任何數(shù)據(jù)，不必擔心數(shù)據(jù)模式變化會引起自身的崩潰等任何問題。他們也可以根據(jù)自己的需要在合適的時候連上新Topic。這就讓整個系統(tǒng)自動化程度更高，在模式改變時減少人工介入。

 

除了Avro解釋規(guī)則，我們也在Schematizer中定義了一些自己的規(guī)則來支持一些數(shù)據(jù)管道功能。模式的主鍵字段被用于在數(shù)據(jù)管道中做日志壓縮。因為對同一個Topic來說做日志壓縮的主鍵必須保持一致，所以任何對主鍵的改動都被認為是不兼容的，會導致Schematizer為新模式創(chuàng)建一個新Topic。而且，當人工不可讀(non-PII，Personally Identifiable Information)的模式開始包含人工可讀字段時，這樣的改動也被認為是不兼容的。人工不可讀的數(shù)據(jù)和人工可讀的數(shù)據(jù)必然分開存儲，這樣就簡化了人工可讀數(shù)據(jù)的安全實現(xiàn)，避免了下游消費者不小心讀到一些他們本來沒有權(quán)限讀的數(shù)據(jù)。

決定是否需要新Topic的邏輯流程

 

值得一提的是模式注冊過程是冪等的。如果把相同的模式注冊多次，那只有第一次會產(chǎn)生一個新模式，后面的都直接返回已注冊的模式。這就讓應(yīng)用程序和服務(wù)可以非常容易地初始化它們的Avro模式。許多應(yīng)用程序和服務(wù)都是把Avro模式定義在文件中或代碼中的，但它們沒辦法寫死模式ID，因為模式ID是由Schematizer管控的。所以應(yīng)用程序可以調(diào)用模式注冊接口來直接注冊模式，如果已經(jīng)存在就把模式信息取回來了，如果不存在就直接注冊，一舉兩得。

 

 

為了讓數(shù)據(jù)管道可以完全以流水線的方式處理模式改變事件，Schematizer會根據(jù)當前模式和新模式的信息來為下游系統(tǒng)生成模式遷移計劃。目前Schematizer只能為Redshift表生成模式遷移計劃。對于把數(shù)據(jù)從數(shù)據(jù)管道中應(yīng)用到Redshift集群的下游系統(tǒng)來說，在模式發(fā)生改變時它可以直接獲取模式遷移計劃并且執(zhí)行，而且自動獲取新的模式信息，不需要任何人工介入。這個功能是很容易擴展的，而且模式遷移計劃生成器也是很容易替換的，所以將來我們會增加更多的模式遷移計劃生成器來支持更多的模式類型，或者改用更好的算法來生成遷移計劃。

 

 

除了管理注冊的模式，Schematizer還會跟進所有數(shù)據(jù)生產(chǎn)者和消費者的信息，包括哪個團隊哪個服務(wù)負責生產(chǎn)或消費什么數(shù)據(jù)，發(fā)布數(shù)據(jù)的頻率如何，等等。在需要人工介入時我們就可以用這些信息來有效地找到相應(yīng)團隊并與他們溝通協(xié)商。而且這些信息也可以幫助我們監(jiān)控和找出那些過期了的模式和Topic，從而可以將它們做廢或刪除。這樣，就可以在新模式注冊上來時簡化兼容性驗證工作。Schematizer可以跳過那些廢棄的模式，只檢查新模式與Topic內(nèi)剩下的有效的模式的兼容性就可以了。

 

所有數(shù)據(jù)生產(chǎn)者和消費者在啟動時都必須提供這些信息。最初我們只想著把它們保存在Schematizer里就好了，但事實上這些信息對探索性的分析和預警都是非常有用的，最終我們決定把它們寫到數(shù)據(jù)管道系統(tǒng)之外的單獨的Kafka Topic中。這樣數(shù)據(jù)就可以被Redshift和Splunk處理，也可以導入Schematizer和通過前端Web界面展示出來。我們用的是Yelp自行研發(fā)的通過Clog寫入數(shù)據(jù)的異步、非阻塞式Kafka生產(chǎn)者，這樣就不會影響生產(chǎn)者正常地發(fā)布數(shù)據(jù)。另外，這樣也可以避免環(huán)形依賴，有時候正常的生產(chǎn)者要用相同的信息去注冊多次。

 

 

與一般意義上的Kafka生產(chǎn)者不同，數(shù)據(jù)管道的數(shù)據(jù)生產(chǎn)者不需要事先知道它們應(yīng)該把數(shù)據(jù)發(fā)送到哪個Kafka Topic中。因為Schematizer規(guī)定了注冊上來的模式和Topic之間的對應(yīng)關(guān)系，所以數(shù)據(jù)生產(chǎn)者只要提供自己序列化數(shù)據(jù)所使用的模式信息，就可以從Schematizer那里得到正確的Topic信息并發(fā)布數(shù)據(jù)了。將Topic信息抽象出去可以讓接口更簡單易用。

 

對數(shù)據(jù)消費者也是類似的機制。盡管也可以給它們定下一些具體的Topic去消費，但更常見的用例是讓Schematizer根據(jù)數(shù)據(jù)消費者感興趣的組的信息來提供正確的Topic。在本文前面章節(jié)介紹了各種不同的分組機制。數(shù)據(jù)消費都可以或者指定名字空間和數(shù)據(jù)源，或者指定數(shù)據(jù)目的方，Schematizer就會找出那個組內(nèi)的相應(yīng)Topic。這種機制對于數(shù)據(jù)消費者感興趣的一組Topic可能由于模式的不兼容改變而變來變?nèi)サ膱鼍坝绕溆行АＫ寯?shù)據(jù)消費者不必再跟蹤組內(nèi)的每一個Topic。

 

 

模式把數(shù)據(jù)格式化了，但對于想了解數(shù)據(jù)確切意義的人來說提供的信息可能又不夠。我們注意到使用數(shù)據(jù)的人通常不是生產(chǎn)數(shù)據(jù)的人，因此他們不知道去哪里找到有用的信息來讓他們理解他們要用的數(shù)據(jù)。因為Schematizer負責管理數(shù)據(jù)管道中的所有模式，所以把數(shù)據(jù)的描述信息也保存在它這里就很合適。

 

 

Schematizer要求模式的注冊方隨著模式一起提供文檔，然后Schematizer會提取文檔信息并保存起來。為了讓Yelp公司內(nèi)的各個團隊可以獲得模式和數(shù)據(jù)文檔，我們開發(fā)了Watson，一個全公司員工都可以用來挖掘數(shù)據(jù)內(nèi)容的Webapp。Watson實際上是Schematizer的一個可視化前端，它通過Schematizer的幾個RESTful API來獲取數(shù)據(jù)。

 

Watson提供了關(guān)于數(shù)據(jù)管道狀態(tài)的有價值信息：現(xiàn)有的名字空間、數(shù)據(jù)源及相關(guān)的Avro模式信息。最重要的是，Watson為查看Schematizer管理的所有數(shù)據(jù)源和模式信息提供了簡單的方法。

 

 

目前流經(jīng)我們數(shù)據(jù)管道的數(shù)據(jù)主要都來自于數(shù)據(jù)庫。我們用SQLAlchemy模型來為這些數(shù)據(jù)的數(shù)據(jù)源和模式整理文檔。在Yelp，SQLAlchemy用來描述我們數(shù)據(jù)庫中的所有模型。除了docstring之外，SQLAlchemy還允許用戶為模型的字段增加額外信息。因此，它自然成了我們保存文檔的首選之處，記錄各個數(shù)據(jù)模型和字段的目的和意義。

 

SQLAlchemy還引入了一個屬主字段來記錄每個模型的維護者和專家。我們認為生成數(shù)據(jù)的人是提供文檔的最佳人選。另外，這種方法也會鼓勵大家時刻保持真實數(shù)據(jù)模型與描述的同步。

 

class BizModel(Base): 
__yelp_owner__ = Ownership( 
teams=[TEAM_OWNERS['biz_team'], 
members=[], 
contacts=[] 
) 
__table_name__ = 'my_biz_table' 
__doc__ = 'Business information.' 
id = Column(Integer, primary_key=True, doc=r"""ID of the business.""") 
name = Column(String(64), doc=r"""Name of the business.""") 

 

可是開發(fā)者在做SQLAlchemy模型的時候并不總是會記得提供文檔信息。為了防止這樣的事情發(fā)生，我們開發(fā)了自動校驗功能來強制要求所有模型都必須完整地提供了屬性描述和文檔，這是絕不會退讓的硬性標準。每當有新模型要加入時，如果要求的文檔信息不完備，或者沒有屬主信息，校驗就會失敗。這些自動校驗功能幫助我們朝著100%文檔覆蓋率的目標邁進了一大步。

 

 

當數(shù)據(jù)模型有了文檔之后，我們就可以把它導入Schematizer并最終通過Watson展現(xiàn)出去。在深入具體提取流程之前，我們先介紹一下這個過程中的另一個重要模塊：特定應(yīng)用轉(zhuǎn)換器(Application Specific Transformer)，簡寫為AST。與名字含義一樣，AST從一個或多個數(shù)據(jù)管道Topic中輸入消息流，用轉(zhuǎn)換邏輯處理消息模式和數(shù)據(jù)包，再把轉(zhuǎn)換后的消息輸出到另外的數(shù)據(jù)管道Topic中。提供具體轉(zhuǎn)換處理的轉(zhuǎn)換模塊是可以串連起來的，因此可以組合多個模塊來做非常細致的轉(zhuǎn)換工作。

 

我們用AST中的許多個轉(zhuǎn)換模塊來依據(jù)SQLAlchemy模型生成更易理解的數(shù)據(jù)。因為模塊是可以串連的，現(xiàn)在我們只是簡單的創(chuàng)建一個從SQLAlchemy模型中提取文檔和屬主信息的轉(zhuǎn)換模塊，并把它加入 到已有的轉(zhuǎn)換鏈中。這樣，所有模型的文檔和屬主信息就通過現(xiàn)有管道自動提取并導入Schematizer了。實現(xiàn)過程相當簡單，并無縫接入管道，所以可以非常有效地生成高質(zhì)量文檔。

AST中的轉(zhuǎn)換模塊

 

如上所述，AST中現(xiàn)在已經(jīng)有了一些為用戶生成更有意義的信息的轉(zhuǎn)換模塊。位標志轉(zhuǎn)換模塊會解釋一個整型字段的不同數(shù)據(jù)位的具體含義。相似地，Enum字段轉(zhuǎn)換模塊也會把Enum值轉(zhuǎn)換成可讀的文字表述。這些轉(zhuǎn)換模塊帶來的另一個好處是它們同時也產(chǎn)生了自解釋和自生成文檔的模式，因此也產(chǎn)生了更好的文檔。

 

 

開發(fā)者的文檔并不是我們要講述的最后一項內(nèi)容。Watson也提供了功能讓終端用戶可以一起努力，為使Yelp的數(shù)據(jù)更具可讀性而貢獻自己的力量。

 

第一個功能就是打標簽。Watson允許用戶為任意數(shù)據(jù)源打標簽分類。一個數(shù)據(jù)源可能是個MySQL數(shù)據(jù)庫表，也可能是個數(shù)據(jù)模型。比如，一個業(yè)務(wù)數(shù)據(jù)源可以打上“Business Information”的標簽，而一個用戶信息數(shù)據(jù)源可以打上“User Information”標簽。終端用戶可以把相關(guān)的數(shù)據(jù)源都打上相同的標簽，這樣以對自己最有意義的方式把它們組織在一起。打標簽可以讓我們更深入的理解我們的各個數(shù)據(jù)源之間是如何彼此關(guān)聯(lián)的。

打上了“Business Info”標簽的業(yè)務(wù)數(shù)據(jù)源

 

Watson提供的另一個功能是添加注釋。終端用戶，尤其是非技術(shù)人員，可以通過這種方法來為一個數(shù)據(jù)源或字段提供他們自己的文檔。比如業(yè)務(wù)分析師常常就會對使用數(shù)據(jù)有非常寶貴的見解，他們可以通過注釋來分享各種疑難雜癥、邊界用例和時效性很強的信息。

 

終端用戶對于Watson的最大的需求就是檢索。我們在Watson中實現(xiàn)了簡單的檢索引擎，讓用戶可以檢索數(shù)據(jù)的模式、Topic、數(shù)據(jù)模型描述等各方面信息。在檢索后臺我們沒有用Elasticsearch，而是選擇了Whoosh Python包，因為它可以幫助我們快速完成開發(fā)。就我們目前的檢索量來說Whoosh的性能足以應(yīng)付。隨著數(shù)據(jù)規(guī)模增大，我們將來會考慮換用其它更易擴展的引擎。

 

 

Schematizer是Yelp數(shù)據(jù)管道的一個重要組成部分。它的模式注冊操作是數(shù)據(jù)管道的許多重要功能的基礎(chǔ)，包括在上游數(shù)據(jù)生產(chǎn)者更改模式時減輕對下游消費者程序和服務(wù)的影響等。Schematizer也管理了數(shù)據(jù)發(fā)布的Topic分配，讓用戶不必再關(guān)心具體使用哪個Topic等這樣的細節(jié)。最后，它要求所有寫入數(shù)據(jù)管道的數(shù)據(jù)都必須有文檔，這促進了全公司內(nèi)的知識分享。Watson的加入更是使得Yelp公司內(nèi)的所有員工都可以方便地得到最及時的信息。