隨著 Uber 業(yè)務的擴張，為其提供支持的基礎(chǔ)數(shù)據(jù)呈指數(shù)級增長，因此處理成本也越來越高。當大數(shù)據(jù)成為我們最大的運營開支之一時，我們開始了一項降低數(shù)據(jù)平臺成本的舉措，該計劃將挑戰(zhàn)分為三部分：平臺效率、供應和需求。本文將討論我們?yōu)樘岣邤?shù)據(jù)平臺效率和降低成本所做的努力。

大數(shù)據(jù)文件格式的優(yōu)化

我們大部分 Apache Hadoop 文件系統(tǒng) （HDFS） 空間都被 Apache Hive 表占用。這些表以 Apache Parquet 文件格式或 Apache ORC 文件格式存儲。盡管我們計劃在未來的某個時候整合到 Parquet，但由于許多特定要求，包括場景的兼容性和性能，我們還未能實現(xiàn)這一目標。

Parquet 和 ORC 文件格式都是基于塊的列格式（block-based columnar formats），這意味著文件包含許多塊，每個塊包含大量的行（假設(shè)為 10，000），這些行數(shù)據(jù)是拆分成列的數(shù)據(jù)存儲的。

我們花了很多時間研究 HDFS 上的文件，并決定進行以下優(yōu)化，主要集中在 Parquet 格式：

? 列刪除：我們的許多 Hive 表，尤其是從 Apache Kafka 日志中提取的表，都包含許多列，其中一些是嵌套的。當我們查看這些列時，發(fā)現(xiàn)其中一些列沒有必要長期保留。這些例子包括調(diào)試每個 Kafka 消息的元數(shù)據(jù)以及由于合規(guī)性原因需要在一段時間后刪除的任何字段。對于列格式文件來說，在技術(shù)上可以刪除文件內(nèi)的列，而無需解壓和重新壓縮其他列。這使得列刪除成為一種非常節(jié)省 CPU 的操作。我們在 Uber 實現(xiàn)了這樣一個功能，在我們的 Hive 表上廣泛使用它，并將代碼貢獻回 Apache Parquet，具體可以參見 #755。

? 壓縮算法：默認情況下，我們使用 GZIP Level 6 作為 Parquet 內(nèi)部的壓縮算法。最近社區(qū)關(guān)于 Parquet 支持 Facebook 的 ZSTD 壓縮算法的 ISSUE 引起了我們的注意。在我們的實驗中，與基于 GZIP 的 Parquet 文件相比，ZSTD Level 9 和 Level 19 能夠?qū)⑽覀兊?Parquet 文件大小分別減少 8% 和 12%。此外，ZSTD 9 級和 19 級的解壓速度都比 GZIP 6 級快。我們決定在 1 個月后采用 ZSTD 9 級重新壓縮我們的數(shù)據(jù)，并在 3 個月后采用 ZSTD 級 19 級進行壓縮。這是因為在我們的實驗中，ZSTD 9 級壓縮比 19 級快 3 倍。請注意，重新壓縮作業(yè)是后臺維護作業(yè)，可以使用無保證的計算資源運行。鑒于此類資源的豐富性，我們基本上可以將這些再壓縮作業(yè)視為無開銷的。

?重新排序行數(shù)據(jù)：行順序可以顯著影響 Parquet 文件的壓縮大小。這是由于 Parquet 格式使用 Run-Length Encoding 功能，以及壓縮算法利用局部重復的能力。我們檢查了 Uber 最大的那些 Hive 表，并手動執(zhí)行了排序，這使表大小減少了 50% 以上。我們發(fā)現(xiàn)的一個常見模式是簡單地按用戶 ID 對行進行排序，然后是日志表的時間戳。大多數(shù)日志表都有用戶 ID 和時間戳這些列。這使我們能夠非常好地壓縮與用戶 ID 關(guān)聯(lián)的許多非規(guī)范化列。

?Delta 編碼（Delta Encoding）：一旦我們開始按時間戳對行進行排序，我們很快注意到 Delta 編碼可以幫助進一步減少數(shù)據(jù)大小，因為與時間戳值本身相比，相鄰時間戳之間的差異非常小。在某些情況下，日志具有穩(wěn)定的節(jié)奏，就像心跳一樣，因此差異是恒定的。但是，我們廣泛使用的 Apache Hive、Presto 和 Apache Spark 的環(huán)境中，在 Parquet 中啟用 Delta 編碼并不容易，正如 StackOverflow 這個問題中所述的。不過我們還在探索這個方向。

HDFS EC

糾刪碼（Erasure Coding）可以顯著降低 HDFS 文件的復制因子。由于潛在增加的 IOPS 工作負載，在 Uber，我們主要研究 3+2 和 6+3 方案，復制因子分別為 1.67 倍和 1.5 倍。鑒于默認的 HDFS 復制因子是 3x，我們可以將 HDD 空間需求減少近一半！

不過，擦除代碼有多種選擇：

?Apache Hadoop 3.0 HDFS Erasure Code：這是在 Apache Hadoop 3.0 中官方實現(xiàn)的糾刪碼。這個實現(xiàn)的好處是它既適用于大文件，又適用于小文件。缺點是 IO 效率不高，因為 Erasure Code 的塊非常碎片化。

?Client-side Erasure Code：這個 是 Facebook 在 HDFS-RAID 項目中首次實現(xiàn)的。這種方法的好處是它的 IO 效率非常高。當所有塊都可用時，讀取 IO 效率與三副本的方式基本相同。缺點是它不適用于小文件，因為每個塊都是糾刪碼計算的一個單位。

在咨詢了行業(yè)專家后，我們決定采用 Apache Hadoop 3.0 HDFS Erasure Code，因為這是社區(qū)的方向。我們?nèi)蕴幱?Apache Hadoop 3.0 HDFS 糾刪碼的評估階段，但我們相信這將對降低我們的 HDFS 成本產(chǎn)生巨大影響。

YARN 調(diào)度策略改進

在 Uber，我們使用 Apache YARN 來運行我們大部分的大數(shù)據(jù)計算工作負載（Presto 除外，Presto 是直接在專用服務器上運行）。就像許多其他公司一樣，我們從 YARN 中的標準容量調(diào)度（Capacity Scheduler）開始。Capacity Scheduler 允許我們?yōu)槊總€隊列配置具有 MIN 和 MAX 設(shè)置的分層隊列結(jié)構(gòu)。我們創(chuàng)建了一個以組織為第一級的 2 級隊列結(jié)構(gòu)，允許用戶根據(jù)子團隊、優(yōu)先級或工作類型創(chuàng)建第二級隊列。

雖然容量調(diào)度為我們管理 YARN 隊列容量提供了一個良好的開端，但我們很快就開始面臨管理 YARN 集群容量的困境：

?高利用率：我們希望 YARN 集群的平均利用率（以分配的 CPU 和 Mem GB /集群的總 CPU 和 MemGB 容量衡量）盡可能高；

?滿足用戶期望：我們希望讓用戶清楚地知道他們希望從集群獲得多少資源 我們的許多用戶對 YARN 集群的資源需求有一些尖銳的但可以預測的需求。例如，一個隊列可能有一組日常作業(yè)，每個作業(yè)都在一天的特定時間開始，并在相同的時間內(nèi)消耗相同數(shù)量的CPU/MemGB。

如果我們將隊列的 MIN 設(shè)置為白天的峰值使用量，那么集群利用率將非常低，因為隊列的平均資源需求遠低于 MIN。

如果我們將隊列的 MAX 設(shè)置為白天的高峰使用，那么隨著時間的推移，隊列可能會被濫用，不斷將資源接近 MAX，進而可能影響其他隊列中其他人的正常工作 。 我們?nèi)绾尾蹲接脩舻馁Y源需求并正確設(shè)定他們的期望？我們提出了以下想法，稱為 Dynamic MAX。Dynamic MAX 算法使用以下設(shè)置：

?設(shè)置隊列的 MIN 為隊列的平均使用率；

?將隊列的 MAX 的設(shè)置使用以下公式：Dynamic_MAX = max（MIN， MIN * 24 – Average_Usage_In_last_23_hours * 23）

Dynamic_MAX 在每小時開始時計算，并應用于該小時的隊列 MAX。

這里的 Dynamic MAX 算法背后的直覺是：

?如果隊列在過去 23 小時內(nèi)根本沒有使用，我們允許隊列峰值最多達到其 MIN 的 24 倍。這通常足 以處理我們絕大多數(shù)的峰值工作負載；

?如果隊列在過去 23 小時內(nèi)平均使用的資源為 MIN ，那么我們只允許隊列在下一個小時最多使用 MIN 的資源。有了這個規(guī)則，隊列在 24 小時內(nèi)的平均使用量不會超過 MIN，從而避免了上面提到的濫用情況。

上面的 Dynamic MAX 算法很容易向用戶解釋：基本上他們的使用量最多可以飆升到他們隊列 MIN 的 24 倍，但是，為了集群的公平性，它們在24小時內(nèi)的累計使用量不能超過 MIN 水平上的恒定使用量。

實際上，我們將 MIN 設(shè)置為隊列平均使用量的 125%，以解決高達 25% 的每日使用差異。這反過來意味著我們的 YARN 集群的平均利用率（以 CPU/MemGB 分配衡量）將在 80% 左右，這對于成本效率來說是一個相當不錯的利用率水平。

避開高峰時間

YARN 資源利用率的另一個問題是整個集群級別仍然存在定時任務。許多團隊決定在 0000 UTC 之間運行他們的 ETL 管道，因為據(jù)說那是日志準備好的時刻。這些管道可能會運行 1-2 個小時，這使得 YARN 集群在那些高峰時段非常忙碌。

我們計劃實施基于時間的速率（time-based rates），而不是向 YARN 集群添加更多機器，這會降低平均利用率并損害成本效率。基本上，當我們計算過去 23 小時的平均使用量時，我們會應用一個根據(jù)一天中的小時而不同的比例因子。例如，0-4 UTC 高峰時段的比例因子為 2 倍，其余時間為 0.8 倍。

集群聯(lián)邦

隨著 YARN 和 HDFS 集群不斷變大，我們開始注意到性能瓶頸。由于集群大小不斷增加，HDFS NameNode 和 YARN ResourceManager 都開始變慢。雖然這主要是一個可擴展性挑戰(zhàn)，但它也極大地影響了我們的成本效率目標。

為了解決這個問題，擺在我們面前的戰(zhàn)略選擇有兩個：

?A. 繼續(xù)提升單節(jié)點性能：比如我們可以使用更多 CPU vCores 和 Memory 的機器。我們還可以運行堆棧跟蹤和火焰圖來找出性能瓶頸并逐個進行優(yōu)化。

?B. 集群聯(lián)邦：我們可以創(chuàng)建一個由許多集群組成的虛擬集群。每個底層集群的大小都適合 HDFS 和 YARN 的最佳性能。上面的虛擬集群將處理所有工作負載路由邏輯。我們選擇 B 方案的原因如下：

?世界上大多數(shù) HDFS 和 YARN 集群都比我們在 Uber 需要的要小。如果我們運行超大規(guī)模的集群，我們很可能會遇到在較小規(guī)模的集群中不會出現(xiàn)的未知bugs。

?為了使 HDFS 和 YARN 能夠擴展到 Uber 的集群規(guī)模，我們可能需要更改源代碼以在性能和復雜功能之間做出不同的權(quán)衡。例如，我們發(fā)現(xiàn)容量調(diào)度器有一些復雜的邏輯會減慢任務分配的速度。但是，為擺脫這些而進行的代碼更改將無法合并到 Apache Hadoop 主干中，因為其他公司可能需要這些復雜的功能。

為了讓我們能夠在不分叉的情況下利用開源 Hadoop 生態(tài)系統(tǒng)，我們決定構(gòu)建我們的聯(lián)邦集群。特別是，我們使用基于路由器的 HDFS 聯(lián)邦和 YARN 聯(lián)邦。它們都來自開源 Apache Hadoop。截至目前，我們已經(jīng)建立了數(shù)十個 HDFS 集群和少數(shù) YARN 集群。HDFS Router-based Federation 一直是我們大數(shù)據(jù)可擴展性工作的基石，它也提高了成本效率。

通用的負載均衡

在本節(jié)中，我們將討論通用的負載均衡解決方案，它適用于 HDFS 和 YARN，方法如下：

?HDFS DataNode 磁盤空間利用率平衡：每個 DataNode 可能有不同的磁盤空間利用率比率。在每個 DataNode 中，每個 HDD 可能具有不同的磁盤空間利用率。所有這些都需要平衡，以實現(xiàn)較高的平均磁盤空間利用率。

?YARN NodeManager 利用率平衡：在任何時間點，YARN 中的每臺機器都可以有不同級別的 CPU 和 MemGB 分配和利用率。同樣，我們需要平衡分配和利用率，以允許較高的平均利用率。

上述解決方案之間的相似性導致了廣義負載平衡（generalized load balancing）思想，它適用于我們大數(shù)據(jù)平臺內(nèi)外的更多用例，例如 微服務負載均衡和主存儲負載均衡。所有這些之間的共同聯(lián)系是，目標始終是縮小 P99 與平均值之間的差距。

查詢引擎

我們在 Uber 的大數(shù)據(jù)生態(tài)系統(tǒng)中使用了幾個查詢引擎：Hive-on-Spark、Spark 和 Presto。這些查詢引擎與文件格式（Parquet 和 ORC）相結(jié)合，為我們的成本效率工作創(chuàng)建了一個有趣的權(quán)衡矩陣。包括 SparkSQL 和 Hive-on-Tez 在內(nèi)的其他選項使決策變得更加復雜。

以下是我們查詢引擎提高成本效率的主要工作點：

?專注于 Parquet 文件格式：Parquet 和 ORC 文件格式共享一些共同的設(shè)計原則，如行組、列存儲、塊級和文件級統(tǒng)計信息。但是，它們的實現(xiàn)是完全獨立的，并且與我們在 Uber 使用的其他專有系統(tǒng)具有不同的兼容性。隨著時間的推移，我們發(fā)現(xiàn) Spark 中對 Parquet 文件有更好的支持，在 Presto 中對 ORC 文件有更好的支持。鑒于對文件格式添加功能的需求不斷增長，我們必須決定一種主要的文件格式，我們選擇了 Parquet。單一的主要文件格式使我們能夠?qū)⒕性谝粋€單一的代碼庫中，并隨著時間的推移積累專業(yè)知識。

?嵌套列裁剪（Nested Column Pruning）：Uber 的大數(shù)據(jù)表具有驚人的高度嵌套數(shù)據(jù)。這部分是因為我們的許多上游數(shù)據(jù)集都以 JSON 格式存儲，并且我們在這些數(shù)據(jù)集上強制使用 Avro 模式。因此，對嵌套列裁剪的支持是 Uber 查詢引擎的一個關(guān)鍵特性，否則深度嵌套的數(shù)據(jù)將需要從 Parquet 文件中完全讀出，即使我們只需要嵌套結(jié)構(gòu)中的單個字段 。 我們?yōu)?Spark 和 Presto 添加了嵌套列裁剪功能。這些顯著提高了我們的整體查詢性能，這些功能已經(jīng)回饋了開源社區(qū)。

?常見查詢模式優(yōu)化：在我們的工作負載中看到接近一千行的 SQL 查詢并不少見。雖然我們使用的查詢引擎都有一個查詢優(yōu)化器，但它們并沒有專門處理 Uber 常見的查詢場景。其中一個例子是使用 SQL constructs，如“RANK（） OVER PARTITION”和“WHERE rank = 1”，目的是提取另一列值處于最大值的行中某一列的值，或用數(shù)學術(shù)語來說是 “ARGMAX” 。當查詢被重寫為使用內(nèi)置函數(shù)“MAX_BY”時，像 Presto 這樣的引擎可以運行得更快。

根據(jù)我們的經(jīng)驗，很難預測哪個引擎最適合特定的 SQL 查詢。Hive-on-Spark 通常對于大量 shuffle 數(shù)據(jù)具有很高的可擴展性。反過來，對于涉及少量數(shù)據(jù)的查詢，Presto 通常非?？臁Ｎ覀冋诜e極關(guān)注開源大數(shù)據(jù)查詢引擎的改進，并將繼續(xù)在這些查詢引擎之間切換我們的工作負載以優(yōu)化成本效率。

Apache Hudi

我們在大數(shù)據(jù)平臺中最大的成本效益之一是高效的增量處理。我們的許多事實數(shù)據(jù)集可能會延遲到達或被更改。例如，在許多情況下，乘客直到他或她準備要求下一次行程時才會對上次行程的司機進行評分。信用卡對旅行的退款有時可能需要一個月的時間來處理。

如果沒有高效的增量處理框架，我們的大數(shù)據(jù)用戶必須每天掃描許多天的舊數(shù)據(jù)，才能使他們的查詢結(jié)果保持新鮮。一種更有效的方法是每天只處理增量更改。這就是 Hudi 項目的意義所在。

我們在 2016 年啟動了 Hudi 項目，并于 2019 年將其提交給 Apache Incubator Project。Apache Hudi 現(xiàn)在是一個 Apache 頂級項目（Top-Level Project），我們在 HDFS 上的大部分大數(shù)據(jù)都是 Hudi 格式。這大大降低了 Uber 的計算能力需求。

下一步和挑戰(zhàn)

大數(shù)據(jù)和在線服務同主機托管

雖然我們決定讓大數(shù)據(jù)工作負載在在線服務不需要對應主機時使用在線服務的主機，但讓兩個工作負載在同一主機上運行會帶來許多額外的挑戰(zhàn)。

關(guān)于托管對性能的影響這一領(lǐng)域有很多研究論文。我們方法的主要區(qū)別在于，我們計劃為大數(shù)據(jù)工作負載提供非常低的優(yōu)先級，以盡量減少其對在線服務的影響。

在線和分析存儲的融合

我們的很多數(shù)據(jù)集都存儲在在線存儲系統(tǒng)（比如 MySQL 數(shù)據(jù)庫中）和分析存儲系統(tǒng)（比如 Hive 表）中。此外，為了實現(xiàn)即時查詢速度，我們還使用了 Pinot 等存儲引擎。所有這些都導致了相同邏輯數(shù)據(jù)的許多副本，盡管以不同的格式存儲。

是否可能有一個統(tǒng)一的存儲系統(tǒng)，既能處理在線查詢，又能處理分析查詢？這將大大降低存儲成本。

Project HydroElectricity：利用維護工作來“存儲”額外的計算能力

集群中的計算能力與電力供應非常相似。它通常在供應方面是固定的，并且在需求激增或不一致的情況下會受到影響。

抽水蓄能水力發(fā)電（Pumped-Storage hydroelectricity）可以將多余的電力以水的重力勢能的形式儲存起來，然后在需求高峰時將其轉(zhuǎn)換回電力。

我們可以將相同的想法應用于計算能力嗎？我們可以！這里要介紹的關(guān)鍵思想是維護作業(yè)（maintenance jobs），它們是可以在第二天甚至一周內(nèi)隨時發(fā)生的后臺任務。典型的維護工作包括 LSM 壓縮、compression、二級索引構(gòu)建、數(shù)據(jù)清理、糾刪碼修復和快照維護。幾乎所有沒有嚴格 SLA 的低優(yōu)先級作業(yè)都可以視為維護作業(yè)。

在我們的大多數(shù)系統(tǒng)中，我們沒有明確拆分維護和前臺工作。例如，我們的 Big Data Ingestion 系統(tǒng)寫入使用 ZSTD 壓縮的 Parquet 文件中，這會占用大量 CPU 資源并生成非常緊湊的文件。除了這樣做之外，我們還可以讓 Ingestion 編寫輕度壓縮的 Parquet 文件，這些文件占用更多磁盤空間但 CPU 更少。然后我們有一個維護作業(yè)，它會晚一點時間運行以重新壓縮文件。通過這種方式，我們可以顯著減少前臺 CPU 的需求。

維護作業(yè)可能需要非保證的計算能力才能運行。正如我們之前所描述的，我們有足夠的資源用于此目的。

Big Data Usage 的定價機制

考慮到在多租戶的大數(shù)據(jù)平臺中，我們經(jīng)常處于很難滿足每個客戶的資源需求的情況下。我們?nèi)绾蝺?yōu)化有限的硬件預算的總效用？帶有高峰時間乘數(shù)的 Dynamic_MAX 是最佳選擇嗎？

事實上，我們相信還有更好的解決方案。然而，這需要我們想出一個更微妙的定價機制。我們想要考慮的例子包括每個團隊可以花在我們集群上的假錢（fake money），用戶可以用來提高他們工作優(yōu)先級的積分等。

總結(jié)

在這篇博文中，我們分享了在提高 Uber 大數(shù)據(jù)平臺效率方面的努力和想法，包括文件格式改進、HDFS 糾刪碼、YARN 調(diào)度策略改進、負載均衡、查詢引擎和 Apache Hudi。這些改進帶來了顯著的成本減少。此外，我們探索了一些開放性的挑戰(zhàn)，例如分析和在線托管以及定價機制。然而，正如我們之前文章中概述的框架所建立的那樣，平臺效率的提升并不能保證高效的運行?？刂茢?shù)據(jù)的供應和需求同樣重要，我們將在下一篇文章中討論。

本文翻譯自：https://eng.uber.com/cost-efficient-big-data-platform/

責任編輯：haq