Facebook 經常使用分析來進行數據驅動的決策。在過去的幾年裡，用戶和産品都得到了增長，使得我們分析引擎中單個查詢的數據量達到了數十TB。我們的一些批處理分析都是基于 Hive 平台（Apache Hive 是 Facebook 在2009年貢獻給社區的）和 Corona（ Facebook 内部的 MapReduce 實現）進行的。Facebook 還針對包括 Hive 在内的多個内部數據存儲，繼續增加了其 Presto 的 ANSI-SQL 查詢的覆蓋範圍。Facebook 内部還支持其他類型的分析，如圖計算、機器學習（Apache Giraph）和流處理（如 Puma、Swift 和 Stylus)。

盡管 Facebook 提供的服務涵蓋了分析領域的廣泛領域，但我們仍在不斷地與開源社區互動，以分享我們的經驗，并向他人學習。Apache Spark 于2009年由加州大學伯克利分校（UC-Berkeley）的 Matei Zaharia 創辦，并于2013年貢獻給 Apache。它是目前增長最快的數據處理平台之一，因為它能夠支持流處理、批處理、命令式(RDD)、聲明式(SQL)、圖計算和機器學習用例，所有這些都在相同的 API 和底層計算引擎中。Spark 可以有效地利用大量内存，跨整個管道（pipelines）優化代碼，并跨任務（tasks）重用 jvm 以獲得更好的性能。Facebook 認為 Spark 已經成熟到可以在許多批處理用例中與 Hive 進行比較的地步。在本文的後面部分，将介紹 Facebook 使用 Spark 替代 Hive 的經驗和教訓。

實時實體排名在 Facebook 有着多種使用場景。對于一些在線服務平台，原始的特性值是使用 Hive 離線生成的，并将生成的數據加載到這些實時關聯查詢系統中。這些 Hive 作業是數年前開發的，占用了大量的計算資源，并且難以維護，因為這些作業被拆分成數百個 Hive 小作業。為了使得業務能夠使用到新的特征數據，并且讓系統變得可維護，我們開始着手将這些作業遷移到 Spark 中。

基于 Hive 的作業由三個邏輯階段組成，每個階段對應數百個由 entity_id 分割的較小 Hive 作業，因為為每個階段運行較大的 Hive 作業不太可靠，并且受到每個作業的最大任務數限制。具體如下：

以上三個邏輯階段可以概括如下：

• 過濾掉非生産需要的特性和噪音；
• 對每個(entity_id、target_id)對進行聚合；
• 将表分為 N 個分片，并對每個切分通過自定義 UDF 生成一個用于在線查詢的自定義索引文件。

基于 Hive 構建索引的作業大約需要運行三天。管理起來也很有挑戰性，因為這條管道包含數百個分片作業，因此很難進行監控。沒有簡單的方法來衡量作業的整體進度或計算 ETA。考慮到現有 Hive 作業的上述局限性，我們決定嘗試使用 Spark 來構建一個更快、更易于管理的作業。

如果使用 Spark 全部替換上面的作業可能會很慢，并且很有挑戰性，需要大量的資源。所以我們首先将焦點投入在 Hive 作業中資源最密集的部分：第二階段。我們從50GB的壓縮輸入樣本開始，然後逐步擴展到 300 GB、1 TB 和20 TB。在每次增加大小時，我們都解決了性能和穩定性問題，但是嘗試 20 TB 時我們發現了最大改進的地方。

在運行 20 TB 的輸入時，我們發現由于任務太多，生成了太多的輸出文件（每個文件的大小大約為100 MB）。在作業運行的10個小時中，有3個小時用于将文件從 staging 目錄移動到 HDFS 中的最終目錄。最初，我們考慮了兩個方案：要麼改進 HDFS 中的批量重命名以支持我們的用例；要麼配置 Spark 以生成更少的輸出文件（這一階段有大量的任務——70,000個）。經過認真思考，我們得到了第三種方案。由于我們在作業的第二步中生成的 tmp_table2 表是臨時的，并且隻用于存儲作業的中間輸出。最後，我們把上面 Hive 實現的三個階段的作業用一個 Spark 作業表示，該作業讀取 60 TB 的壓縮數據并執行 90 TB的 shuffle 和排序，最後的 Spark job 如下:

當然，在如此大的數據量上運行單個 Spark 作業在第一次嘗試甚至第十次嘗試時都不會起作用。據我們所知，這是生産環境中 shuffle 數據量最大的 Spark 作業（Databricks 的 PB 級排序是在合成數據上進行的）。我們對 Spark 内核和應用程序進行了大量的改進和優化，才使這項工作得以運行。這項工作的好處在于，其中許多改進都适用于 Spark 的其他大型工作負載，并且我們能夠将所有工作重新貢獻給開源 Apache Spark 項目 - 有關更多詳細信息，請參見下面相關的 JIRA。下面我們将重點介紹将一個實體排名作業部署到生産環境的主要改進。

處理節點頻繁重啟

為了可靠地執行長時間運行的作業，我們希望系統能夠容錯并從故障中恢複（主要是由于正常維護或軟件錯誤導緻的機器重新啟動）。雖然 Spark 最初的設計可以容忍機器重動，但我們還是發現了各種各樣的 bug/問題，我們需要在系統正式投入生産之前解決這些問題。

• 使得 PipedRDD 容忍節點重啟（SPARK-13793）：PipedRDD 之前在處理節點重啟設計不夠健壯，當它獲取數據失敗時，這個作業就會失敗。我們重新設計了 PipedRDD，使得它能夠友好的處理這種異常，并且從這種類型的異常中恢複。
• 最大的獲取失敗次數可配置（ SPARK-13369 ）：對于長期運行的作業而言，由于計算機重動而導緻獲取失敗的可能性大大增加。 在 Spark 中每個階段允許的最大獲取失敗次數是寫死的，因此，當達到最大失敗次數時，作業通常會失敗。我們做了一個更改，使其變得可配置，并将這個參數的值從 4 增加到 20，使得作業對于 fetch 失敗更加健壯。
• Less disruptive cluster restart：長時間運行的作業應該能夠在集群重啟後繼續運行，這樣我們就不會浪費到目前為止完成的所有處理。Spark 的可重啟 shuffle service 讓我們在節點重啟後保留 shuffle 文件。最重要的是，我們在 Spark driver 中實現了能夠暫停任務調度的功能，這樣作業就不會因為集群重啟而導緻任務失敗。

• Unresponsive driver（SPARK-13279）：Spark driver 添加任務會進行一項時間複雜度為 O(N2) 的操作，這可能會導緻其被卡住，最終導緻作業被 killed。我們删除這個不必要的 O(N2) 操作來解決這個問題。
• Excessive driver speculation：我們發現，Spark driver 在管理大量任務時，會花費了大量時間進行推測（speculation）。在短期内，在運行這個作業時我們禁止了 speculation。我們目前正在對 Spark Driver 進行修改，以減少 speculation 的時間。
• TimSort issue due to integer overflow for large buffer（ SPARK-13850 ）：我們發現 Spark 的 unsafe 内存操作有一個 bug，這會導緻 TimSort 中的内存出現問題。不過 Databricks 的工作人員已經修複了這個問題，使我們能夠在大型内存緩沖區上進行操作。
• Tune the shuffle service to handle large number of connections：在 shuffle 階段，我們看到許多 executors 在試圖連接 shuffle service 時超時。通過增加 Netty 服務線程（spark.shuffle.io.serverThreads）和 backlog （spark.shuffle.io.backLog）的數量解決了這個問題。
• Fix Spark executor OOM（ SPARK-13958 ）：一開始在每個節點上運行四個以上的 reduce 任務是很有挑戰性的。Spark executors 的内存不足，因為 sorter 中存在一個 bug，該 bug 會導緻指針數組無限增長。我們通過在指針數組沒有更多可用内存時強制将數據溢寫到磁盤來修複這個問題。因此，現在我們可以在一個節點上運行 24個任務而不會導緻内存不足。

在實現了上述可靠性改進之後，我們能夠可靠地運行 Spark 作業。此時，我們将工作重心轉移到與性能相關的問題上，以最大限度地利用 Spark。我們使用Spark 的指标和 profilers 來發現一些性能瓶頸。

我們用來發現性能瓶頸的工具

• Spark UI Metrics：Spark UI 可以很好地洞察特定階段的時間花在哪裡。每個任務的執行時間被劃分為子階段，以便更容易地找到作業中的瓶頸。
• Jstack：Spark UI 中還提供 executor 進程的 jstack 功能，這個可以幫助我們找到代碼中的熱點問題。
• Spark Linux Perf/Flame Graph support：盡管上面的兩個工具非常方便，但它們并沒有提供同時運行在數百台機器上作業的 CPU 概要的聚合視圖。在每個作業的基礎上，我們增加了對性能分析的支持，并且可以定制采樣的持續時間/頻率。

• Fix memory leak in the sorter（SPARK-14363）性能提升 30%：我們發現當任務釋放所有内存頁，但指針數組沒有被釋放。結果，大量内存未被使用，導緻頻繁溢出和 executor OOMs。現在，我們修複了這個問題，這個功能使得 CPU 性能提高了30%；
• Snappy optimization （ SPARK-14277 ）性能提升 10%：對于每一行的讀/寫，都會調用 JNI 方法（Snappy.ArrayCopy）。我們發現了這個問題，并且将這個調用修改成非 JNI 的System.ArrayCopy調用，修改完之後 CPU 性能提高了10%；
• Reduce shuffle write latency（SPARK-5581）性能提升近 50%：在 map 端，當将 shuffle 數據寫入磁盤時，map 任務的每個分區打開和關閉相同的文件。我們修複了這個問題，以避免不必要的打開/關閉，修改完之後 CPU 性能提高近 50%；
• Fix duplicate task run issue due to fetch failure （SPARK-14649）：當獲取失敗（fetch failure）發生時，Spark driver 會重新提交已經運行的任務，這會導緻性能低下。我們通過避免重新運行正在運行的任務修複了這個問題，并且我們發現當發生獲取操作失敗時，作業也更加穩定。
• Configurable buffer size for PipedRDD（SPARK-14542）性能提升近 10%：在使用 PipedRDD 時，我們發現用于将數據從排序器（sorter）傳輸到管道處理的默認緩沖區大小太小，我們的作業花費了超過 10% 的時間來複制數據。我們使這個緩沖區大小變得可配置，以避免這個瓶頸。
• Cache index files for shuffle fetch speed-up（SPARK-15074）：我們發現，shuffle service 經常成為瓶頸，reduce 端花費 10% 到 15% 的時間來等待獲取 map 端的數據。通過更深入的研究這個問題，我們發現 shuffle service 為每次 shuffle fetch 都需要打開/關閉 shuffle index 文件。我們通過緩存索引信息，這樣我們就可以避免重複打開/關閉文件，這一變化減少了50%的 shuffle fetch 時間；
• Reduce update frequency of shuffle bytes written metrics（SPARK-15569）性能提升近 20%：使用 Spark Linux Perf 集成，我們發現大約 20% 的 CPU 時間花在探測和更新随機字節寫的指标上。
• Configurable initial buffer size for Sorter（ SPARK-15958 ）性能提升近 5%：Sorter 的默認初始緩沖區大小太小(4 KB)，對于大的工作負載來說這個值太小了，因此我們浪費了大量的時間來複制内容。我們将這個緩沖區大小變得可配置（備注：spark.shuffle.sort.initialBufferSize）， 當将這個參數設置為 64 MB 時，可以避免大量的數據複制，使得性能提升近 5%；
• Configuring number of tasks：由于我們輸入的數據大小為 60 T，每個 HDFS 塊大小為 256 M，因此我們要生成超過250,000個任務。盡管我們能夠運行具有如此多任務的 Spark 作業，但我們發現，當任務數量過高時，性能會顯著下降。我們引入了一個配置參數，使 map 輸入大小可配置，我們通過将輸入的 split 大小設置為 2 GB ，使得 task 的數據減少了八倍。

在所有這些可靠性和性能改進之後，我們的實體排名系統變成了一個更快、更易于管理的管道，并且我們提供了在 Spark 中運行其他類似作業的能力。

我們使用以下性能指标來比較 Spark 和 Hive 運行性能。

CPU time：這是從操作系統的角度來看 CPU 使用情況。例如，如果您的作業在32核機器上僅運行一個進程，使用所有 CPU 的50%持續10秒，那麼您的 CPU 時間将是 32 0.5 10 = 160 CPU 秒。

CPU reservation time：從資源管理框架的角度來看，這是 CPU 預留（CPU reservation）。例如，如果我們将32核機器預留10秒來運行這個作業，那麼 CPU 預留時間是 32 * 10 = 320 CPU秒。CPU 時間與 CPU 預留時間的比率反映了我們集群預留 CPU 資源的情況。準确地說，當運行相同的工作負載時，與 CPU 時間相比，預留時間可以更好地比較執行引擎。例如，如果一個進程需要1個 CPU 秒來運行，但是必須保留100個 CPU 秒，那麼根據這個指标，它的效率低于需要10個 CPU 秒但隻預留10個 CPU 秒來做相同數量的工作的進程。我們還計算了内存預留時間，但這裡沒有列出來，因為這些數字與 CPU 預留時間類似，而且使用 Spark 和 Hive 運行這個程序時都沒有在内存中緩存數據。Spark 有能力在内存中緩存數據，但由于集群内存的限制，我們并沒有使用這個功能。

Latency：作業從開始到結束運行時間。

Facebook 使用高性能和可擴展的分析引擎來幫助産品開發。Apache Spark 提供了将各種分析用例統一到單個 API ，并且提供了高效的計算引擎。我們将分解成數百個 Hive 作業管道替換為一個 Spark 作業，通過一系列的性能和可靠性改進，我們能夠使用 Spark 來處理生産中的實體數據排序的用例。在這個特殊的用例中，我們展示了 Spark 可以可靠地 shuffle 并排序 90 TB 以上的中間數據，并在一個作業中運行 250,000個 tasks。與舊的基于 Hive 計算引擎管道相比，基于 Spark 的管道産生了顯著的性能改進（4.5-6倍 CPU性能提升、節省了 3-4 倍資源的使用，并降低了大約5倍的延遲），并且已經在生産環境中運行了幾個月。

作者：過往記憶大數據

本文為阿裡雲原創内容，未經允許不得轉載。

更多精彩资讯请关注tft每日頭條，我们将持续为您更新最新资讯!