三年前，Uber采用Apache Hadoop作為數(shù)據(jù)平臺，從而可以跨計(jì)算機(jī)集群管理數(shù)PB的數(shù)據(jù)。但是，因?yàn)槲覀冇泻芏鄨F(tuán)隊(duì)、工具和數(shù)據(jù)源，所以需要一種可靠的方式來攝取和分散數(shù)據(jù)。Marmaray是Uber開源的Apache Hadoop數(shù)據(jù)提取和分散框架。Marmaray由我們的Hadoop平臺團(tuán)隊(duì)設(shè)計(jì)和開發(fā)，是一個(gè)建立在Hadoop生態(tài)系統(tǒng)之上的基于插件的框架。用戶可以新增插件以便從任何來源攝取數(shù)據(jù)，并利用Apache Spark將數(shù)據(jù)分散到接收器上。

Marmaray這個(gè)名字源于土耳其的一條連接歐洲和亞洲的隧道，在Uber內(nèi)部，我們將Marmaray設(shè)想為根據(jù)客戶偏好將數(shù)據(jù)從任何來源連接到任何接收器的管道。數(shù)據(jù)湖的數(shù)據(jù)通常在質(zhì)量方面存在很大差異。Marmaray可以確保所有攝取的原始數(shù)據(jù)都符合適當(dāng)?shù)脑茨Ｊ�，保持高質(zhì)量水平，從而帶來可靠的分析結(jié)果。數(shù)據(jù)科學(xué)家可以將時(shí)間花在從這些數(shù)據(jù)中提取有用的見解上，而不是用來處理數(shù)據(jù)質(zhì)量問題。

在Uber，Marmaray以聚合的方式連接各種系統(tǒng)和服務(wù)：

通過我們的模式管理庫和服務(wù)生成高質(zhì)量的模式化數(shù)據(jù)。

通過Marmaray的攝取組件將多個(gè)數(shù)據(jù)存儲(chǔ)中的數(shù)據(jù)攝取到我們的Hadoop數(shù)據(jù)湖中。

使用Uber的內(nèi)部工作流程編排服務(wù)來構(gòu)建管道，用以處理攝取的數(shù)據(jù)，以及根據(jù)這些數(shù)據(jù)保存和計(jì)算業(yè)務(wù)指標(biāo)。

將處理的結(jié)果提供給在線數(shù)據(jù)存儲(chǔ)，內(nèi)部客戶可以通過Marmaray查詢數(shù)據(jù)并獲得近乎實(shí)時(shí)的結(jié)果。

 

 

圖1：Marmaray將數(shù)據(jù)攝取到Hadoop數(shù)據(jù)湖中，并分散到數(shù)據(jù)存儲(chǔ)中。

雖然Marmaray實(shí)現(xiàn)了任意數(shù)據(jù)源到任意數(shù)據(jù)槽的連接，但我們還需要構(gòu)建一個(gè)自助服務(wù)平臺，為來自不同背景、團(tuán)隊(duì)和技術(shù)專業(yè)知識的用戶提供無縫的使用體驗(yàn)。

大規(guī)模數(shù)據(jù)攝入的挑戰(zhàn)

Uber的業(yè)務(wù)生成大量的原始數(shù)據(jù)，并將它們存儲(chǔ)在各種數(shù)據(jù)源中，例如Kafka、Schemaless和MySQL。我們需要將這些數(shù)據(jù)攝取到Hadoop數(shù)據(jù)湖中，以進(jìn)行業(yè)務(wù)分析。隨著Uber業(yè)務(wù)的垂直增長，數(shù)據(jù)攝取的規(guī)模呈現(xiàn)出指數(shù)級的增長。面對大規(guī)模數(shù)據(jù)可靠性的需求，我們不得不重新構(gòu)建我們的攝取平臺，以確保能夠跟上公司增長的步伐。

 

 

圖2：隨著Uber繼續(xù)擴(kuò)展全球業(yè)務(wù)，存儲(chǔ)在Hadoop數(shù)據(jù)湖中的原始數(shù)據(jù)呈指數(shù)級增長。

之前的數(shù)據(jù)架構(gòu)需要運(yùn)行和維護(hù)多個(gè)數(shù)據(jù)管道，每個(gè)管道對應(yīng)不同的生產(chǎn)代碼庫，隨著數(shù)據(jù)量的增加，這些管道變得越來越繁瑣。MySQL、Kafka和Schemaless等數(shù)據(jù)源包含需要被攝取到Hive中的原始數(shù)據(jù)，以支持整個(gè)公司團(tuán)隊(duì)的各種分析需求。每個(gè)數(shù)據(jù)源都有自己的代碼庫和相關(guān)的復(fù)雜性，以及一組獨(dú)有的配置、圖表和警報(bào)。添加新的攝取源迫在眉睫，維護(hù)的開銷要求我們的大數(shù)據(jù)生態(tài)系統(tǒng)支持所有這些系統(tǒng)。輪班待命的負(fù)擔(dān)十分繁重，有時(shí)每周會(huì)收到200多個(gè)警報(bào)。

 

 

圖3：Hadoop平臺團(tuán)隊(duì)的輪班待命警報(bào)圖表，可見維護(hù)系統(tǒng)需要多大的開銷。

隨著Marmaray的推出，我們將攝取管道整合到一個(gè)與源無關(guān)的管道和代碼庫中，以此來提高可維護(hù)性和資源利用率。

無論源數(shù)據(jù)存儲(chǔ)是什么，單一攝取管道都將執(zhí)行相同的有向非循環(huán)圖作業(yè)(DAG)。在運(yùn)行時(shí)，攝取行為將根據(jù)特定源(類似于策略設(shè)計(jì)模式)而有所變化，以協(xié)調(diào)攝取過程，并使用一種通用的靈活配置來適應(yīng)未來不同的需求和用例。

Uber的數(shù)據(jù)分散需求

我們的很多內(nèi)部數(shù)據(jù)用戶，例如Uber Eats和Michelangelo機(jī)器學(xué)習(xí)平臺團(tuán)隊(duì)，他們使用Hadoop與其他工具來構(gòu)建和訓(xùn)練機(jī)器學(xué)習(xí)模型，以生成有價(jià)值的衍生數(shù)據(jù)集，從而提高效率并改善用戶體驗(yàn)。為了最大化這些衍生數(shù)據(jù)集的價(jià)值，需要將這些數(shù)據(jù)分散到在線數(shù)據(jù)存儲(chǔ)中，這些數(shù)據(jù)存儲(chǔ)通常要求比Hadoop生態(tài)系統(tǒng)低得多的延遲。

在引入Marmaray之前，每個(gè)團(tuán)隊(duì)都需要構(gòu)建自己的臨時(shí)擴(kuò)散系統(tǒng)。這些重復(fù)性工作和不具有通用性的功能通常導(dǎo)致工程資源的極度浪費(fèi)。Marmaray于2017年底發(fā)布，以滿足對靈活的通用分散平臺的需求，它將Hadoop數(shù)據(jù)傳輸?shù)饺魏我粋�(gè)在線數(shù)據(jù)存儲(chǔ)，以此來完善Hadoop生態(tài)系統(tǒng)。

跟蹤端到端數(shù)據(jù)傳輸

我們的很多內(nèi)部用戶需要保證將數(shù)據(jù)源的數(shù)據(jù)傳輸?shù)侥繕?biāo)接收器，他們還需要完整性指標(biāo)，包括數(shù)據(jù)傳輸?shù)阶罱K接收器的可靠性。從理論上講，這意味著數(shù)據(jù)被100%傳輸?shù)侥康牡�，但�?shí)際上我們的目標(biāo)是提供99.99%到99.999%的可靠性。當(dāng)記錄數(shù)量非常少時(shí)，可以直接對源系統(tǒng)和接收器系統(tǒng)運(yùn)行查詢來驗(yàn)證數(shù)據(jù)是否已送達(dá)。

在Uber，我們每天攝取數(shù)PB的數(shù)據(jù)和超過1000億條消息，因此無法通過查詢的方式進(jìn)行數(shù)據(jù)驗(yàn)證。面對這么大規(guī)模的數(shù)據(jù)，我們需要一個(gè)能夠跟蹤數(shù)據(jù)傳輸而不會(huì)顯著增加延遲的系統(tǒng)。Marmaray通過自定義Spark累加器來桶分化記錄，讓用戶能夠以最小的開銷監(jiān)控?cái)?shù)據(jù)傳輸。

Marmaray的架構(gòu)

下面的架構(gòu)圖說明了Marmaray的基本構(gòu)建塊和抽象。我們可以通過這些通用組件向Marmaray添加擴(kuò)展，以便支持新的數(shù)據(jù)源和接收器。

 

 

圖4：Marmaray的架構(gòu)，包含各種組件。

DataConverters

攝取和分散作業(yè)主要對來自數(shù)據(jù)源的記錄執(zhí)行轉(zhuǎn)換，以確保在將數(shù)據(jù)寫入目標(biāo)接收器之前具備所需格式。Marmaray將多個(gè)轉(zhuǎn)換器鏈接在一起執(zhí)行多次轉(zhuǎn)換，并且可以寫入多個(gè)接收器。

DataConverters的一個(gè)關(guān)鍵的作用是在轉(zhuǎn)換時(shí)生成錯(cuò)誤記錄。為了保證分析結(jié)果更準(zhǔn)確，所有原始數(shù)據(jù)在被攝入到Hadoop數(shù)據(jù)湖之前必須符合一定的模式。任何包含錯(cuò)誤格式、缺少必需字段或被視為有問題的數(shù)據(jù)都將被過濾掉并寫入錯(cuò)誤表。

WorkUnitCalculator

Marmaray按照批次來移動(dòng)數(shù)據(jù)，批次大小是可配的。為了計(jì)算需要處理的數(shù)據(jù)量，我們引入了WorkUnitCalculator的概念。WorkUnitCalculator將檢查輸入源的類型和之前的檢查點(diǎn)，然后計(jì)算下一個(gè)工作單元或批次。工作單元可以是Kafka的偏移范圍或Hive/HDFS的HDFS文件集合。

在計(jì)算下一批數(shù)據(jù)時(shí)，WorkUnitCalculator還可以應(yīng)用限流信息，例如，要讀取的最大數(shù)據(jù)量或要從Kafka讀取的消息數(shù)。這些限流信息可根據(jù)具體用例進(jìn)行靈活配置，從而確保工作單元的大小不會(huì)壓垮源系統(tǒng)或接收系統(tǒng)。

Metadata Manager

 

 

圖5：Marmaray的Metadata Manager用于存儲(chǔ)運(yùn)行中作業(yè)的相關(guān)元數(shù)據(jù)。

所有Marmaray作業(yè)都需要一個(gè)持久存儲(chǔ)(我們稱之為Metadata Manager)來緩存作業(yè)的元數(shù)據(jù)信息。作業(yè)可以在執(zhí)行期間更新狀態(tài)，并在作業(yè)執(zhí)行成功后替換舊狀態(tài)，否則將無法修改狀態(tài)。在Uber，我們使用Metadata Manager存儲(chǔ)檢查點(diǎn)信息(或Kafka的分區(qū)偏移量)、平均記錄大小和平均消息數(shù)等元數(shù)據(jù)。元數(shù)據(jù)存儲(chǔ)是通用的，可以存儲(chǔ)任何相關(guān)的度量指標(biāo)，這些度量指標(biāo)可用于根據(jù)用例和用戶需求跟蹤、描述或收集作業(yè)的狀態(tài)。

ForkOperator和ForkFunction

ForkOperator使用ForkFunction將輸入的記錄流拆分為多個(gè)輸出流，并驗(yàn)證符合模式的記錄和錯(cuò)誤的記錄，然后可以單獨(dú)處理這些記錄。

 

 

圖6：ForkOperator和ForkFunction用于將原始數(shù)據(jù)記錄拆分為符合模式的記錄和錯(cuò)誤的記錄。

ISource和ISink

ISource包含工作單元所需的源數(shù)據(jù)信息，ISink包含有關(guān)如何寫入接收器的必要信息。例如，Cassandra接收器可能需要包含集群、表、分區(qū)鍵和集群鍵的信息。Kafka源需要包含主題名稱、要讀取的最大消息數(shù)、群集信息和偏移量初始化策略以及其他元數(shù)據(jù)信息。

數(shù)據(jù)模型和工作流

AvroPayload是Marmaray架構(gòu)的核心組件，它是Avro GenericRecord二進(jìn)制編碼格式的包裝器，包含了需要處理的數(shù)據(jù)的元數(shù)據(jù)信息。

 

 

圖7：AvroPayload使用元數(shù)據(jù)包裝GenericRecord。

Avro數(shù)據(jù)(GenericRecord)的主要優(yōu)點(diǎn)是它在內(nèi)存存儲(chǔ)和網(wǎng)絡(luò)傳輸方面都很高效，與JSON相比，二進(jìn)制編碼數(shù)據(jù)通過線路發(fā)送需要更小的模式開銷。在Spark架構(gòu)之上使用Avro數(shù)據(jù)意味著我們還可以利用Spark的數(shù)據(jù)壓縮和加密功能。這些優(yōu)勢有助于我們的Spark作業(yè)更有效地處理大規(guī)模數(shù)據(jù)。

為了支持連接任何數(shù)據(jù)源和任意接收器的架構(gòu)，我們要求所有攝取源將轉(zhuǎn)換器定義為從自己的格式到Avro，所有分散接收器將轉(zhuǎn)換器定義為從Avro到本地?cái)?shù)據(jù)模型(對于Cassandra來說就是ByteBuffers)。

要求所有轉(zhuǎn)換器將數(shù)據(jù)轉(zhuǎn)換為AvroPayload格式或從AvroPayload格式轉(zhuǎn)換為數(shù)據(jù)為形成了松散耦合的數(shù)據(jù)模型。在定義了數(shù)據(jù)源及其相關(guān)的轉(zhuǎn)換器后，就可以將數(shù)據(jù)分散到任何受支持的接收器，因?yàn)樗�有接收器都是源不可知的，它們只關(guān)心AvroPayload格式的數(shù)據(jù)。下圖描繪了Marmaray的數(shù)據(jù)模型：

 

 

圖8：對于攝取和分散，Marmaray要求將數(shù)據(jù)轉(zhuǎn)換為AvroPayload，這是一個(gè)基于Avro的GenericRecord格式的包裝器。

下圖描繪了Marmaray作業(yè)是如何獨(dú)立于特定的數(shù)據(jù)源或接收器進(jìn)行編排的。

 

 

圖9：Marmaray獨(dú)立于數(shù)據(jù)源或接收器運(yùn)行攝取和分散作業(yè)。

在這個(gè)過程中，每個(gè)數(shù)據(jù)源和接收器的屬性配置將指向作業(yè)的下一個(gè)步驟，包括計(jì)算需要處理的數(shù)據(jù)量(即其工作單元)、應(yīng)用ForkFunction將原始數(shù)據(jù)拆分為“有效”和“錯(cuò)誤”記錄以確保數(shù)據(jù)的質(zhì)量、將數(shù)據(jù)轉(zhuǎn)換為適當(dāng)?shù)哪繕?biāo)格式、更新元數(shù)據(jù)并報(bào)告指標(biāo)以便跟蹤進(jìn)度。在Uber，所有Marmaray作業(yè)都運(yùn)行在Apache Spark上，并使用YARN作為資源管理器。

自助服務(wù)平臺

由于我們的很多數(shù)據(jù)平臺用戶不熟悉我們的技術(shù)棧中所使用的語言(如Python和Java)，因此我們的團(tuán)隊(duì)必須構(gòu)建一個(gè)自助服務(wù)平臺，用戶只需通過鼠標(biāo)點(diǎn)擊就可以建立端到端的管道，確保來自所需源的數(shù)據(jù)最終能夠到達(dá)目標(biāo)接收器，以便進(jìn)行后續(xù)的分析工作和查詢。

 

 

圖10：我們的自助服務(wù)UI讓數(shù)據(jù)科學(xué)家和其他用戶能夠?qū)��?shù)據(jù)從任何源移動(dòng)到任何接收器，而無需了解特定的數(shù)據(jù)格式。

在上線之后的七個(gè)月時(shí)間里，已經(jīng)有超過3300個(gè)作業(yè)通過我們的自助服務(wù)平臺加入到我們的系統(tǒng)中。

數(shù)據(jù)刪除

在Uber，所有的Kafka數(shù)據(jù)都是以追加的方式進(jìn)行存儲(chǔ)，并使用了日期分區(qū)。用戶數(shù)據(jù)可以跨越多個(gè)日期分區(qū)，并且每個(gè)分區(qū)通常會(huì)包含很多Kafka記錄。如果底層存儲(chǔ)沒有內(nèi)置的索引和更新支持，那么掃描和更新這些分區(qū)以便進(jìn)行用戶數(shù)據(jù)的更正、更新或刪可能會(huì)非常耗費(fèi)資源。Hadoop使用的Parquet數(shù)據(jù)存儲(chǔ)不支持索引，所以我們根本無法直接更新Parquet文件。為了便于索引和更新，Marmaray使用了Hadoop Updates和Incremental(Hudi)，Uber開發(fā)的另一個(gè)開源庫，用于管理大型分析數(shù)據(jù)集的存儲(chǔ)，將原始數(shù)據(jù)存儲(chǔ)在Hive中。

數(shù)據(jù)生產(chǎn)者使用Hive來掃描數(shù)據(jù)表，識別要?jiǎng)h除的記錄，并將它們發(fā)布到Kafka集群。Marmaray的Kafka攝取管道依次從Kafka集群中讀取它們。然后，Marmaray使用Hudi的批量插入功能攝取新記錄，保持較低的攝取延遲，并使用Hudi的upsert功能處理更新的記錄，將Kafka的舊記錄替換為更新過的數(shù)據(jù)。

 

 

圖11：Marmaray利用Hudi存儲(chǔ)格式來支持?jǐn)?shù)據(jù)刪除。

Marmaray的下一個(gè)篇章

Marmaray對任意源到任意接收器數(shù)據(jù)管道的支持適用于Hadoop生態(tài)系統(tǒng)(主要針對使用了Hive的場景)以及數(shù)據(jù)遷移的各種用例。我們已經(jīng)向開源社區(qū)發(fā)布了Marmary，并期待收到更多反饋，然后不斷改進(jìn)Marmaray平臺。與此同時(shí)，我們正在棄用傳統(tǒng)管道，并將我們所有的工作流程遷移到Marmaray平臺上，以簡化我們的整體數(shù)據(jù)架構(gòu)，并確保隨著數(shù)據(jù)需求的增長，我們能夠輕松地?cái)U(kuò)展。

Marmaray GitHub地址：https://github.com/uber/marmaray

英文原文：https://eng.uber.com/marmaray-hadoop-ingestion-open-source/

標(biāo)簽： Mysql 大數(shù)據(jù) 代碼 網(wǎng)絡(luò)

版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點(diǎn)！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請與原作者聯(lián)系。