Uber 大數據平台的演進(2014~2019)

大數據技術2019-07-11 08:00:57

來自公眾號:過往記憶大數據

https://www.iteblog.com/archives/2557.html

原文:https://eng.uber.com/uber-big-data-platform/


Uber 致力於在全球市場上提供更安全,更可靠的運輸服務。為了實現這一目標,Uber 在很大程度上依賴於數據驅動的決策,從預測高流量事件期間騎手的需求到識別和解決我們的駕駛員-合作伙伴註冊流程中的瓶頸。自2014年以來,Uber 一直致力於開發大數據解決方案,確保數據可靠性,可擴展性和易用性;現在 Uber 正專注於提高他們平台的速度和效率。

本文將介紹 Uber 的大數據平台的演進。

第一代:Uber 大數據平台的開端

在2014年之前,Uber 的數據量非常有限,可以直接存儲在在線事務處理(OLTP)數據庫(比如,MySQL、PostgreSQL)。為了利用這些數據,Uber 的工程師必須單獨訪問每個數據庫或表,如果他們需要組合來自不同數據庫的數據,則用户需要自己編寫一些代碼。那時,Uber 沒有全局訪問或所有存儲數據的全局視圖。實際上,那時候 Uber 的數據分散在不同的 OLTP 數據庫中,總數據大小大約為幾TB,訪問這些數據的延遲非常快(通常是亞分鐘)。下圖概述了2014年之前的數據架構:


如果想及時瞭解SparkHadoop或者HBase相關的文章,歡迎關注微信公眾號:iteblog_hadoop

隨着 Uber 的業務呈指數級增長(無論是運營的城市/國家數量以及每個城市使用該服務的乘客/司機數量),傳入的數據量也增加。為了能夠在一個地方訪問和分析所有的數據,Uber 建立第一代分析數據倉庫。為了使 Uber 儘可能地受數據驅動,需要確保分析師可以在一個地方訪問分析數據。為實現這一目標,Uber 首先將數據用户分為三大類:

  • 城市運營團隊(數千名用户):這些現場工作人員負責管理和擴展 Uber 在每個市場的運輸網絡。隨着我們的業務擴展到新的城市,有成千上萬的城市運營團隊需要定期訪問這些數據,以解決駕駛員和騎手的問題。

  • 數據科學家和分析師(數百名用户):這些分析師和科學家分佈在不同的團隊,這些團隊需要數據為用户提供最佳的運輸和交付體驗。

  • 工程團隊(數百名用户):整個公司的工程師專注於構建自動數據應用程序,例如我們的欺詐檢測(Fraud Detection)和司機入職(Driver Onboarding)平台。

Uber 的第一代分析數據倉庫專注於將所有數據聚集在一個地方並且簡化數據的訪問。對於前者,Uber 使用 Vertica 作為數據倉庫軟件,因為它具有快速,可擴展和麪向列的設計。Vertica 是一款基於列存儲的 MPP 架構的數據庫。它可以支持存放多至 PB 級別的結構化數據。Vertica 是由關係數據庫大師 Michael Stonebraker(2014 年圖靈獎獲得者)所創建,於2011年被惠普收購併成為其核心大數據平台軟件。同時,Uber 還開發了多個臨時 ETL(提取,轉換和加載)作業,這些作業將來自不同數據源(比如 AWS S3,OLTP 數據庫,服務日誌等)的數據複製到 Vertica 中。為了實現後者,Uber 將 SQL 標準化並作為其解決方案接口,然後構建了一個在線查詢服務來接受用户查詢,最後這些查詢將提交給底層查詢引擎。下圖描述了這個分析數據倉庫的架構:


如果想及時瞭解SparkHadoop或者HBase相關的文章,歡迎關注微信公眾號:iteblog_hadoop

第一代數據倉庫服務的發佈對整個公司的工程師來説是一個巨大的成功。用户第一次擁有全局視圖,可以在一個地方訪問所有數據。這導致大量新團隊使用這個服務作為其技術和產品決策的基礎。在幾個月內,這個服務的分析數據量增長到數十TB,用户數量增加到數百。

使用 SQL 作為簡單的標準接口,使城市運營者能夠輕鬆地與數據進行交互,而無需瞭解底層技術。此外,不同的工程團隊開始構建針對用户需求量身定製的服務和產品,這些服務和產品由這些數據提供信息,並組建了新團隊(比如機器學習團隊)以更好地使用和提供這些數據。

不足

第一代數據倉庫的廣泛使用以及傳入數據的增長使得這個服務的缺點很快暴露出來。由於數據是通過臨時 ETL 作業攝取的,而且缺乏正式的通信機制,因此數據可靠性成為一個問題。另外,大多數數據都是 JSON 格式,可擴展性不強。

另外,隨着公司的發展,擴展數據倉庫變得越來越昂貴。為了降低成本,不得不開始刪除舊的過時數據,以釋放空間。除此之外,Uber 的大部分大數據平台都不能橫向擴展,因為之前的主要目標是解決集中數據訪問的關鍵業務需求,並且沒有足夠的時間來確保所有部件都是水平可擴展的。這個數據倉庫實際上被用作成數據湖,裏面堆積了所有原始數據,並在其之上進行了數據建模。

最後,由於生成數據的服務與下游數據消費者之間缺乏通信,將數據寫入到數據倉庫的 ETL 作業也非常脆弱。如果不同的用户在攝取期間執行了不同的轉換,則可能導致相同的數據被攝取多次。這對我們的上游數據源造成了額外的壓力,並影響了他們的服務質量。此外,這也導致數據倉庫存儲了幾個相同數據的副本,進一步增加了存儲成本。數據的攝取工作缺乏標準化,因此很難攝取任何新的數據集和類型。

第二代:Hadoop 的引入

為了解決這些限制,Uber 開始圍繞 Hadoop 生態系統重新構建了新的大數據平台。更具體地説,Uber 引入了一個 Hadoop 數據湖,其中所有原始數據僅從不同的在線數據存儲中提取一次,並且在攝取期間不進行轉換。這種設計轉變明顯降低了在線數據存儲的壓力,使 Uber 能夠從臨時攝取作業過渡到可擴展的攝取平台。為了讓用户能夠訪問 Hadoop 中的數據,使用 Presto 來實現交互式查詢,使用 ApacheSpark 對原始數據進行編程訪問,使用 Apache Hive 進行非常大的離線查詢。這些不同的查詢引擎允許用户針對其需求進行選擇,這使得第二代大數據平台更加靈活。

為了保持平台的可擴展性,我們確保所有數據建模和轉換僅在 Hadoop 中進行,從而在出現問題時實現快速恢復。只有最關鍵的建模表才可以遷移到這個數據倉庫。這大大降低了運行龐大數據倉庫的運營成本。

我們還利用了 Apache Parquet 的標準列式文件格式,這提高了數據壓縮率,從而節省了存儲空間。此外,Parquet 與 Apache Spark 的無縫集成使該解決方案成為訪問 Hadoop 數據的流行選擇。下圖總結了我們的第二代大數據平台的架構:


如果想及時瞭解Spark、Hadoop或者HBase相關的文章,歡迎關注微信公眾號:iteblog_hadoop

除了整合 Hadoop 之外,我們還使該生態系統中的所有數據服務都可以橫向擴展,從而提高了我們的大數據平台的效率和穩定性。特別是,具有這種通用的水平可擴展性可以滿足業務需求,使我們能夠專注於構建下一代數據平台。

與第一代平台不同,第二次大數據平台允許對所有數據進行模式化,從 JSON 轉換為 Parquet 以將模式和數據存儲在一起。為實現這一目標,我們構建了一個集中模式服務來收集,存儲和相關客户端庫,以便將不同服務數據模式集成到這個中央模式服務中。

隨着 Uber 的業務繼續以快速的速度擴展,我們很快就擁有了數百PB的數據。每天都有數十TB的新數據被添加到數據湖中,大數據平台增長到超過10,000個 vcores,每天都有超過100,000個批處理作業。這導致這個Hadoop數據湖成為所有分析 Uber 數據的集中真實來源。

不足

隨着公司不斷擴展,並且這個數據平台存儲了數百PB的數據,我們開始面臨着一系列新的挑戰。

首先,HDFS 中存儲了大量的小文件開始對 NameNode 產生額外的壓力。最重要的是,數據延遲仍然遠遠超出了我們的業務需求。用户只能每隔24小時訪問一次新數據,這對於需要做實時決策的需求來説太慢了。將 ETL 和建模遷移到 Hadoop 使得這個過程更具可擴展性,這些步驟仍然是瓶頸,因為這些 ETL 作業必須在每次運行重新創建整個建模表。除此之外,新數據的攝取和相關派生表的建模都需要創建整個數據集的新快照,並將新表替換舊錶以向用户提供對新數據的訪問。攝取作業必須返回源數據存儲區,創建新快照,並在每次運行期間將整個數據集轉換為列式存儲的 Parquet 文件。隨着我們的數據存儲量的增長,這些工作可能需要超過20個小時才能運行超過1,000個Spark執行程序。

每項工作的很大一部分涉及將歷史數據和新數據進行轉換並生成新快照。雖然每張表每天只添加100MB左右的新數據,但每次運行的攝取作業都必須轉換該表的整個數據集。對於在每次運行中重新創建新派生表的ETL和建模作業也是如此。從本質上講,我們的數據包含許多更新操作。由於 HDFS 和 Parquet 不支持數據更新,因此所有的提取作業需要從更新的源數據創建新快照,將新快照寫入到 Hadoop 中,並將其轉換為 Parquet 格式的文件,然後替換表的數據以查看新數據。下圖總結了這些基於快照的數據攝取如何在大數據平台進行的:


如果想及時瞭解Spark、Hadoop或者HBase相關的文章,歡迎關注微信公眾號:iteblog_hadoop

第三代:從長遠角度重新構建大數據平台

到2017年初,我們的大數據平台被整個公司的工程和運營團隊使用,使他們能夠在一個地方訪問新的和歷史的數據。用户可以通過單一的 UI 接口輕鬆訪問 Hive,Presto,Spark,Vertica,Notebook 中的數據。我們的計算集羣中有超過 100PB 的數據,集羣共有 100,000 vcores。每天有超過 100,000 個 Presto 查詢、 10,000個Spark 作業、20,000個 Hive 查詢。我們的 Hadoop 分析平台遇到了可擴展性問題,許多服務都受到高數據延遲的影響。

幸運的是,由於我們的底層基礎架構可以橫向擴展以滿足當前的業務需求,因此我們有足夠的時間研究數據內容,數據訪問模式和用户特定需求,以便在構建下一代大數據處理平台之前確定最緊迫的問題。我們總結了四個主要的痛點:

  • HDFS 可擴展性限制:許多依賴 HDFS 擴展其大數據基礎架構的公司都面臨着這個問題。根據設計,HDFS 受 NameNode 內存容量的限制,因此存儲大量小文件會顯着影響性能。當數據大小超過10PB這個問題開始出現,如果數據量達到50-100 PB就會成為問題。幸運的是,有一些相對簡單的解決方案可以將 HDFS 從幾十PB擴展到幾百PB,例如利用 ViewFS 和 HDFS NameNode Federation;或通過控制小文件的數量並將不同的數據移到單獨的集羣,這樣我們能夠減輕 HDFS 的瓶頸。

  • Hadoop中的數據需要快速更新:Uber 的業務是實時計算的,因此,我們的服務需要儘可能的訪問到新數據。因此,對於許多用例而言,T+1的數據更新延遲太長了,這些場景對數據的實時更新需求很高。我們的第二代大數據平台基於快照的攝取方法效率低下,使得我們難以以較低的延遲攝取數據。為了加速數據交付,我們不得不重新構建我們的數據流管道,並僅僅增量攝取更新的和新插入的數據。

  • 支持在 Hadoop 和 Parquet 中進行更新和刪除操作:Uber 的數據包含大量更新,更新範圍包括過去幾天到幾周甚至幾個月。通過基於快照的數據提取,我們每24小時提取一次源數據的新副本。換句話説,我們一次攝取所有更新,每天一次。但是,由於需要更新的數據和增量攝取,我們的解決方案必須能夠支持現有數據的更新和刪除操作。但是,由於我們的大數據存儲在 HDFS 和 Parquet 中,因此無法直接支持對現有數據的更新操作。另一方面,我們的表通常包含 1,000 個列,並且有五個或更多嵌套級別,而用户的查詢通常只觸及其中一些列,從而阻止我們使用更加高效的行式存儲。為了讓我們的大數據平台能夠實現長期增長,我們必須找到一種方法來解決 HDFS 文件系統中的這種限制,以便我們也可以支持更新/刪除操作。

  • 支持更快的 ETL 和建模:與原始數據攝取類似,ETL 和建模作業也是基於快照方式的,這要求我們的平台在每次運行時重建派生表。為減少建模表的數據延遲,ETL 作業也需要以增量方式獲取數據。這要求 ETL 作業逐步從原始源表中提取已更改的數據並更新先前派生的輸出表,而不是每隔幾小時重建整個輸出表。

Hudi 介紹

考慮到上述要求,我們構建了 Hudi(Hadoop Upserts anD Incremental),這是一個開源的 Spark 庫,在 HDFS 和 Parquet 之上提供抽象層,以支持所需的更新和刪除操作。Hudi 可以在任何 Spark 作業中使用,可以橫向擴展,並且運行時只依賴於 HDFS。因此,任何需要支持歷史數據更新/刪除操作的大數據平台都可以利用 Hudi。

Hudi 使我們能夠在 Hadoop 中更新,插入和刪除現有的 Parquet 數據。此外,Hudi 允許用户獲取僅更改的數據,顯着提高查詢效率並允許派生建模表的增量更新。

我們的 Hadoop 生態系統中的原始數據是根據時間分區的,任何舊分區都可能在以後被更新。因此,對於依賴於這些原始數據表的用户或 ETL 作業,瞭解哪個分區被更新的唯一方法就是掃描整個源表並根據一些條件過濾不要的數據。這導致查詢的計算成本非常高,需要對源表進行全量掃描。

通過引入 Hudi,用户可以傳遞最後一次 checkpoint 時間戳並檢索所有已更新的記錄,無論這些更新是添加到新分區還是舊分區,無需掃描整個表。

通過 Hudi 庫,我們能夠從基於快照的原始數據提取轉移到增量提取模型,使我們能夠將數據延遲從24小時減少到不到一小時。下圖描述了引入 Hudi 後我們的大數據平台架構:


如果想及時瞭解Spark、Hadoop或者HBase相關的文章,歡迎關注微信公眾號:iteblog_hadoop

備註:其實 Uber 的 Hudi 和前段時間 databricks 開源的 Delta 功能很類似,這個系統目前也開源了,具體參見:https://hudi.apache.org

第四代:展望

自2017年推出第三代大數據平台以來,整個公司的用户可以快速可靠地訪問 Hadoop 中的數據,但是這個平台還是有很大的跟進空間。下面總結了我們增強 Uber 的大數據平台的努力,以提高數據質量,數據延遲,效率,可擴展性和可靠性。

數據質量

為了加強數據質量,我們確定了兩個改進方向。首先,當某些上游數據存儲在存儲之前沒有強制執行或檢查數據模式時,我們希望避免非符合模式的數據。因為這個會導致不規範的數據進入我們的 Hadoop 生態系統,從而影響所有依賴此數據的下游用户。為了防止髒數據流入,我們正在對所有上游數據存儲的數據模式進行強制檢查,並在數據存在任何問題時拒絕這些數據的寫入。

我們發現的第二個改進點是數據內容的質量。雖然使用模式檢查能夠確保數據包含正確的數據類型,但它們不檢查實際數據值。為了提高數據質量,我們正在擴展架構服務以支持語義檢查。這些語義檢查允許我們在基本結構類型檢查之外添加對實際數據內容的額外約束。

數據延遲

我們的目標是將 Hadoop 中的原始數據延遲減少到五分鐘,將建模表的數據延遲減少到十分鐘。這將允許更多用例從流處理轉向使用 Hudi 增量數據拉取的更有效的小批量處理。

我們還在擴展我們的 Hudi 項目以支持視圖模式,其中包括現有的讀取優化視圖,以及顯示延遲僅幾分鐘的數據實時視圖。這個實時視圖依賴於 Merge-On-Read 或 Hudi 2.0。

數據效率

為了提高數據效率,我們打算不再依賴專用硬件來實現任何服務和服務容器化。此外,我們統一了 Hadoop 生態系統內部和跨 Hadoop 生態系統的所有資源調度程序,以彌合整個公司的 Hadoop 和非數據服務之間的差距。這允許所有作業和服務以統一的方式進行調度。隨着 Uber 的發展,數據位置將成為 Hadoop 應用程序的一大關注點,成功的統一資源管理器可以將所有現有的調度程序集中在一起。

擴展性和可靠性

為了確保無論數據從哪裏來的都能統一進行數據攝取,我們與 Uber 數據存儲團隊合作啟動了一個項目,以統一所有上游數據源的更改日誌的內容,格式和元數據。



●編號848,輸入編號直達本文

●輸入m獲取到文章目錄

推薦↓↓↓

程序員數學之美

https://hk.wxwenku.com/d/201109248