電信行業的實際案例研究

調試實際的分布式應用程序可能是一項艱巨的任務。 至少在一開始，最常見的Google搜索并沒有什么用。 在這篇博客文章中，我將詳細介紹如何將Apache Kafka / Spark Streaming / Apache Ignite應用程序加速近十倍，并將開發原型轉變為有用的，穩定的流媒體應用程序，該應用程序最終超過了性能目標。為應用程序。

此處學習的課程相當籠統，可以輕松地擴展到使用MapR Streams和Kafka的類似系統。

該項目是融合平臺需求的具體案例，該平臺集成了完整的軟件堆棧以支持該系統的需求：實時流，大數據分布式處理和持久性。 截至撰寫本文時， MapR融合數據平臺是此類平臺當前唯一可用的生產就緒型實現。

系統目標

為了滿足電信公司的需求，該應用程序的目標是將來自三個獨立系統的日志數據結合在一起。 加入數據后，就可以將網絡條件與任何特定客戶的特定呼叫相關聯，從而使客戶支持可以向不滿意其電話服務的客戶提供準確和有用的信息。 如果該應用程序可以實時進行而不是作為批處理工作，則它具有巨大的附加價值，因為6個小時的呼叫質量信息對客戶服務或網絡運營沒有實際價值。

基本上，這是一個相當簡單的ETL作業，通常會作為數據倉庫的批處理作業完成，但現在必須作為流式分布式體系結構實時進行。

更具體地說，總體情況是將來自遠程服務器的輸入數據流式傳輸到分布式集群中，進行一些數據清理和擴充，將三個日志中的記錄聯接在一起，并將聯接的數據作為單個表持久保存到數據庫中。

原始系統的問題

原始系統存在幾個圍繞性能和穩定性的問題。

首先，流應用程序不穩定。 在Spark Streaming應用程序中，如果每個微批處理的處理時間等于或小于批處理時間，則稱該流穩定。 在這種情況下，應用程序的流式傳輸部分正在30秒的窗口中接收數據，但處理時間為4.5-6分鐘。

其次，有一個批處理過程，一次要一個小時連接一次數據，目標是在30分鐘內運行，但要花2個小時才能完成。

第三，應用程序運行了幾個小時后隨機崩潰。

集群硬件，軟件堆棧和輸入數據

集群硬件非常好，有12個企業服務器節點，每個節點配備兩個E5 Xeon CPU，每個CPU具有16個物理核心，256GB內存和八個6TB旋轉硬盤。 網絡是10GB以太網。

為該項目選擇的技術棧圍繞著Kafka 0.8（用于將數據流式傳輸到系統中），Apache Spark 1.6（用于ETL操作）（本質上是對輸入進行過濾和轉換，然后進行聯接）以及使用Apache Ignite進行。 1.6作為內存共享緩存，可以很容易地將應用程序的流輸入部分與數據連接起來。 如果發生故障，Apache Hive還可以用作Ignite的磁盤備份以及單獨的分析應用程序。

初始集群的配置如下：

節點	k	神經網絡	HDFS	梅索斯	Mesos Master	卡夫卡	火花工人	點燃
1個	X	X	X	X		X	X	X
2	X	X	X	X		X	X	X
3	X		X	X		X	X	X
…			X	X		X	X	X
7			X	X	X		X	X
8			X	X	X		X	X
…			X	X			X	X
12			X	X			X	X

該集群運行Apache Hadoop的HDFS作為分布式存儲層，資源由Mesos 0.28管理。 最后，HBase用作最終聯接數據的最終數據存儲。 該項目范圍之外的其他系統將查詢該文件。

系統的性能要求是處理高達3GB / min或150-200,000個事件/秒的輸入吞吐量，代表已知的峰值數據吞吐量以及額外的余量。 普通吞吐量約為該值的一半，即1.5GB /分鐘和60,000-80,000個事件/秒。

原始數據源是三個遠程系統的日志，此處標記為A，B和C：日志A占條目的約84-85％，日志B約占1-2％，日志C約占14-15％。 數據不平衡這一事實是此應用程序中（許多）困難的來源之一。

Spark應用程序均使用Scala 2.10和Kafka的直接方法 （無接收器）進行編碼。 Apache Ignite具有一個非常好的Scala API和一個神奇的IgniteRDD ，它可以使應用程序共享內存數據，這是該系統降低編碼復雜性的關鍵功能。

應用架構

單個Kafka生產者將原始數據攝入到運行在6臺服務器上的Kafka中。 生產者讀取各種日志，并將每個日志的記錄添加到其自己的主題中。 由于有三個日志，因此有三個Kafka主題。 每個主題分為36個分區。 最有可能存在36個分區，因為有6個節點，每6個磁盤分配給HDFS，Kafka文檔似乎建議每個物理磁盤大約有一個分區作為指導。

Spark Streaming應用程序使用數據，該應用程序選擇每個主題，然后執行簡單的過濾器以切出不必要的字段，進行映射操作以轉換數據和執行foreachRDD操作（每個微批處理在Spark Streaming中生成RDD）將數據保存到Ignite和Hive。

流媒體應用程序非常簡單：映射，過濾器和foreach分區，保存到Ignite

第二個“常規” Spark應用程序運行在Ignite存儲在內存中的數據上，以1小時為批處理將來自三個單獨日志的記錄合并到一個表中。 這項工作是使用Spark的DataFrame API完成的，該API非常適合該任務。 第二部分涉及不超過100GB的數據，并且群集硬件的大小適當以處理該數量的數據。

三個小時的數據被累積到Ignite中，因為絕大多數調用持續不到一個小時，并且我們希望一次對一個小時的數據進行連接。 由于某些呼叫將在一批中開始而在另一批中完成，因此系統將保留三個小時并且僅處理一個小時的中間批次，因此聯接可以成功處理接近100％的記錄。

值得一提的是，更好的全流架構可以避免中間表示形式的出現。 具有更多時間和事前思考能力的示例性現實案例可以更快地結束整個項目，而不是全神貫注地編寫第一個可行的解決方案。

性能調優

這些應用程序的主要問題是由于試圖運行開發系統的代碼而造成的，這些代碼在運行于真實數據的物理，本地群集上的AWS實例上進行了測試。 從來沒有授予原始開發人員訪問生產集群或真實數據的權限。

Apache Ignite是一個巨大的問題源，主要是因為它是一個新項目，沒有人有任何實際經驗，而且還因為它不是一個非常成熟的項目。

Spark Streaming應用程序在大約4.5分鐘內運行，并且項目目標是在30秒內運行。 我們需要找到9倍于加速的改進價值，并且由于時間限制，我們無法更改任何代碼！

該系統必須在一周內準備好進行生產測試，因此從體系結構和算法角度出發的代碼被假定為正確且足夠好，以至于我們只有通過調整才能達到性能要求。

修復RPC超時異常

我們從具有相同問題的人那里找到了正確的解決方案，如JIRA的SPARK-14140中所示 。 他們建議將spark.executor.heartbeatInterval從10s增加到20s。

我認為這個問題可能是由于Kafka，Ignite或垃圾收集器暫停而導致節點由于磁盤繁忙或CPU高峰而變得忙碌所致。 由于Spark在所有節點上運行，因此問題是隨機的。 （請參閱第一部分中的集群服務布局表。）

配置更改完全解決了此問題。 從那以后我們再也沒有看到過。

增加驅動程序和執行程序的內存

通過將內存從每個執行者20g增加到每個執行者40g以及驅動程序40g，解決了內存不足問題和應用程序的隨機崩潰。 令人高興的是，生產集群中的機器配備了大量內存。 對于新應用程序，這是一個好習慣，因為您一開始不知道需要多少。

由于Spark UI報告的內存消耗非常小，因此很難精確地調試該問題，缺乏準確的信息。 實際上，由于此設置易于更改，因此根據經驗，我們將40g作為使應用程序穩定運行的最小內存大小。

增加并行度：增加Kafka中的分區數量

輸入數據不平衡，大部分應用程序處理時間都花在處理主題1（吞吐量的85％）上。 Kafka分區與輸入RDD中的分區數量進行1：1匹配，導致只有36個分區，這意味著我們只能讓36個核心忙于此任務。 為了增加并行度，我們需要增加分區數。 因此，我們將主題1分為12個主題，每個主題有6個分區，總共72個分區。 我們對生產者進行了簡單的修改，將第一個日志中的數據平均分為12個主題，而不僅僅是一個。 消費者方需要修改零代碼。

我們還根據其他兩個主題在輸入數據中的相對重要性，適當調整了分區數的大小，因此我們將主題2設置為2，將主題3設置為8。

并行運行更多任務。 調整之前，每個階段始終有36個分區！

調整執行程序的大小

原始應用程序僅運行3個執行程序，共有72個內核。 我們將應用程序配置為以80個內核運行，每個執行者最多10個內核，總共8個執行者。 請注意，在10個節點的集群中，每個節點具有16個實際核心，我們為Kafka代理，Ignite和HDFS / NN留下了足夠的資源。

將批處理窗口從30s增加到1m

生產者每隔30秒將數據分批推送到Kafka，因為它是通過FTP批處理從遠程系統收集的。 由于需要處理制造商，技術和年齡的困惑范圍內的設備和系統，因此這種布置在電信應用中很常見。

這意味著輸入流非常不完整，如Spark UI的“流”選項卡的屏幕截圖所示：

將窗口增加到1m可使我們平滑輸入，并使系統有機會在1分鐘或更短的時間內處理數據，但仍保持穩定。

為了確保這一點，該團隊生成了一個測試數據，該數據模擬了已知的最壞情況數據，并且使用新設置，火花流工作現在確實很穩定。 該團隊還能夠輕松地在測試數據和實際生產數據流之間進行切換，并通過限制生產者來配置要傳入系統的數據量。 這對于快速測試各種配置并查看我們是否取得了進展非常有幫助。

刪除要求保存到Hive，僅使用Ignite

與項目經理的討論表明，Hive實際上并不是流應用程序需求的一部分！ 主要是因為HBase中的數據可以輕松地被分析使用。 同樣，在此應用程序的上下文中，每個單獨的記錄實際上都不需要100％保證地進行處理。

確實，根據系統的目標，丟失數據的最壞情況是無法找到客戶的呼叫質量信息……情況已經如此。 換句話說，數據丟失的風險不是破壞交易的因素，而獲得數據的好處是更多的見解。 只要處理和存儲絕大多數數據，就可以實現業務目標。

所有優化的結果

流媒體應用程序最終穩定下來，優化運行時間為30-35s。

事實證明，淘汰Hive還加快了將數據連接在一起的第二個Spark應用程序的運行，因此它現在的運行時間為3500萬，這意味著這兩個應用程序現在都符合項目要求。

隨著下一部分的改進，Spark Streaming作業的最終性能下降到20s的較低范圍，最終加速了12倍以上。

我們必須在穩定性方面下大力氣。 需要采取幾種策略，如下所述。

使Spark Streaming應用程序穩定

我們為修復性能所做的工作直接影響了系統的穩定性。 如果兩個應用程序本身都穩定并且在適當大小的資源上運行，則系統最有可能總體上保持穩定。

刪除Mesos并使用Spark Standalone

Mesos最初選擇管理資源是前瞻性的，但最終我們決定將其從最終生產系統中刪除。 首先，計劃是讓Mesos管理所有應用程序。 但是團隊永遠無法讓Kafka和Ignite與Mesos保持良好的合作關系，因此他們以獨立模式運行，僅由Spark由Mesos管理。 當然，隨著時間的推移，毫無疑問，所有應用程序都可以正確配置為與Mesos一起使用。

提議刪除Mesos有點爭議，因為Mesos比在獨立模式下運行的Spark更先進，更酷。

但是Mesos的問題是雙重的：

對執行程序大小和數量的控制很差，這是Spark 1.6的一個已知問題（ SPARK-5095 ），已在Spark 2.0中修復。

Ignite和Kafka不在Mesos內部運行，只是Spark。 由于日程安排的壓力，該團隊已放棄嘗試使這兩個服務在Mesos中運行。

如果Mesos實際控制資源，則它只能分配好資源。 就此系統而言，Kafka和Ignite的運行超出了Mesos的知識范圍，這意味著它將錯誤地將資源分配給Spark應用程序。

此外，它是一個單一用途的集群，因此我們可以使用系統資源的全局視圖為每個應用程序自定義資源的大小。 幾乎不需要動態資源分配，調度隊列，多租戶和其他流行語。

更改點燃記憶模型

一個已知的問題是，當由JVM控制的堆變得很大（> 32GB）時，垃圾回收的成本會很大。 當加入應用程序運行時，我們確實可以看到這個問題：25GB隨機播放的階段中有些行的GC時間峰值很大，從10秒到超過一分鐘不等。

Ignite的初始配置是運行ONHEAP_TIERED，并在堆上緩存48GB的數據，然后溢出降至12GB的堆外內存。 該設置已更改為OFFHEAP_TIERED模型。 盡管由于序列化成本而稍慢，但是OFFHEAP_TIERED不會導致大量垃圾回收。 它仍然在內存中運行，因此我們估計這將是凈收益。

進行此更改后，每個批次的運行時間從30秒降低到了約25秒，減少了約5秒鐘。 此外，連續的批處理往往具有更多相似的處理時間，增量為1-3秒，而先前的變化會超過5至10秒。

更新Ignite JVM設置

我們遵循了Ignite文檔的性能調整部分（ http://apacheignite.gridgain.org/docs/jvm-and-system-tuning ）中推薦的JVM選項。

完善Spark代碼

代碼的某些部分假定可靠性，例如對Ignite的查詢，而實際上卻存在操作失敗的可能性。 這些問題可以在代碼中解決，現在可以更優雅地處理異常，盡管可能還有很多工作可以提高代碼的健壯性。 我們只能通過立即運行該應用程序來找到這些位置。

將ZooKeeper重新分配給節點10-12

鑒于群集是中型的，因此有必要盡可能多地擴展服務。 我們將ZooKeeper服務從節點1-3移到了節點10-12。

結論

調整此應用程序大約需要1周的全職工作。 我們使用的主要信息是Spark UI和Spark日志，可以從Spark UI輕松訪問。 作業和階段以及流UI的視圖確實非常有用。

我學到的是

• 將流應用程序從AWS上的原型遷移到本地集群需要安排測試時間
• 不使用真實數據測試AWS原型是一個大錯誤
• 包括許多對可靠性要求很高的“出血邊緣” OSS組件（Apache Ignite和Mesos）是不現實的
• 更好的架構設計可以極大地簡化系統
• 調整Kafka / Spark Streaming應用程序需要對整個系統有一個整體的了解。 這不僅僅是改變Spark的參數值； 它是數據流特征，應用程序目標和對客戶的價值，硬件和服務，應用程序代碼，然后使用Spark參數的組合。
• MapR融合數據平臺將減少該項目的開發時間，復雜性和成本。

該項目是這家特定電信公司的第一個項目，他們決定全力開發這種先進的100％開放源代碼平臺。 他們的開拓精神應受到贊揚。 但是，更好的平臺和應用程序體系結構選擇將使他們的生活更加輕松。

現在需要融合的大數據平臺

實際上，該項目的需求表明了現實世界中對具有最新的融合平臺的業務需求，該平臺具有快速分布式文件系統，用于持久性的高性能鍵值存儲以及實時流功能。

由于該架構所需的完整軟件堆棧已經內置并得到完全支持，因此MapR解決方案可能會跳過對仍然投機的開源項目（如Ignite）的要求。 鑒于該系統已開始為具有24/7可靠性預期的電信運營商量產，因此這一優勢非常可觀。

翻譯自: https://www.javacodegeeks.com/2017/01/performance-tuning-apache-kafkaspark-streaming-system.html

總結

以上是生活随笔為你收集整理的Apache Kafka / Spark流系统的性能调优的全部內容，希望文章能夠幫你解決所遇到的問題。

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