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 的經驗和教訓。

用例：為實體排序（entity ranking）做特性準備

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

以前的 Hive 作業實現

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

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

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

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

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 來完成這項工作？

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

可靠性修復（Reliability fixes）

處理節點頻繁重啟

為了可靠地執行長時間運行的作業，我們希望系統能夠容錯并從故障中恢復（主要是由于正常維護或軟件錯誤導致的機器重新啟動）。雖然 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?，F在，我們修復了這個問題，這個功能使得 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 運行上面實體排名程序性能比較

我們使用以下性能指標來比較 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倍的延遲），并且已經在生產環境中運行了幾個月。

原文鏈接
本文為阿里云原創內容，未經允許不得轉載

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的60TB 数据量的作业从 Hive 迁移到 Spark 在 Facebook 的实践的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。