spark SQL踩坑記

作了一年延雲YDB的開發，這一年在使用spark上真心踩了很多坑，總結一下，但願對你們有所幫助。

spark 內存泄露

1.高併發狀況下的內存泄露的具體表現

很遺憾，Spark的設計架構並非爲了高併發請求而設計的，咱們嘗試在網絡條件很差的集羣下，進行100併發的查詢，在壓測3天后發現了內存泄露。

a)在進行大量小SQL的壓測過程當中發現，有大量的activejob在spark ui上一直處於pending狀態，且永遠不結束，以下圖所示

 

b)而且發現driver內存爆滿

 

c)用內存分析分析工具分析了下

 

2.高併發下AsynchronousListenerBus引發的WEB UI的內存泄露

短期內 SPARK 提交大量的SQL ，並且SQL裏面存在大量的 union與join的情形，會建立大量的event對象，使得這裏的 event數量超過10000個event ，

一旦超過10000個event就開始丟棄 event，而這個event是用來回收 資源的，丟棄了 資源就沒法回收了。 針對UI頁面的這個問題，咱們將這個隊列長度的限制給取消了。

 

 

 

 

 

 

3.AsynchronousListenerBus自己引發的內存泄露

抓包發現

 

 

 

這些event是經過post方法傳遞的，並寫入到隊列裏

 

 

可是也是由一個單線程進行postToAll的

 

 

可是在高併發狀況下，單線程的postToAll的速度沒有post的速度快，會致使隊列堆積的event愈來愈多，若是是持續性的高併發的SQL查詢，這裏就會致使內存泄露

 

接下來咱們在分析下postToAll的方法裏面，那個路徑是最慢的，致使事件處理最慢的邏輯是那個？

 

 

 

 

可能您都不敢相信，經過jstack抓取分析，程序大部分時間都阻塞在記錄日誌上

 

能夠經過禁用這個地方的log來提高event的速度

 

log4j.logger.org.apache.spark.scheduler=ERROR

 

 

 

 

4.高併發下的Cleaner的內存泄露

       說道這裏，Cleaner的設計應該算是spark最糟糕的設計。spark的ContextCleaner是用於回收與清理已經完成了的 廣播boradcast,shuffle數據的。可是高併發下，咱們發現這個地方積累的數據會愈來愈多，最終致使driver內存跑滿而掛掉。

l咱們先看下，是如何觸發內存回收的

 

      沒錯，就是經過System.gc() 回收的內存，若是咱們在jvm裏配置了禁止執行System.gc，這個邏輯就等於廢掉（並且有不少jvm的優化參數通常都推薦配置禁止system.gc 參數）

lclean過程

這是一個單線程的邏輯，並且每次清理都要協同不少機器一同清理，清理速度相對來講比較慢，可是SQL併發很大的時候，產生速度超過了清理速度，整個driver就會發生內存泄露。並且brocadcast若是佔用內存太多，也會使用很是多的本地磁盤小文件，咱們在測試中發現，高持續性併發的狀況下本地磁盤用於存儲blockmanager的目錄佔據了咱們60%的存儲空間。

 

 

咱們再來分析下 clean裏面，那個邏輯最慢

 

真正的瓶頸在於blockManagerMaster裏面的removeBroadcast，由於這部分邏輯是須要跨越多臺機器的。

 

針對這種問題，

l咱們在SQL層加了一個SQLWAITING邏輯，判斷了堆積長度，若是堆積長度超過了咱們的設定值，咱們這裏將阻塞新的SQL的執行。堆積長度能夠經過更改conf目錄下的ya100_env_default.sh中的ydb.sql.waiting.queue.size的值來設置。

 

l建議集羣的帶寬要大一些，萬兆網絡確定會比千兆網絡的清理速度快不少。

l給集羣休息的機會，不要一直持續性的高併發，讓集羣有間斷的機會。

l增大spark的線程池，能夠調節conf下的spark-defaults.conf的以下值來改善。

 

 

 

5.線程池與threadlocal引發的內存泄露

       發現spark，Hive，lucene都很是鍾愛使用threadlocal來管理臨時的session對象，期待SQL執行完畢後這些對象可以自動釋放，可是與此同時spark又使用了線程池，線程池裏的線程一直不結束，這些資源一直就不釋放，時間久了內存就堆積起來了。

針對這個問題，延雲修改了spark關鍵線程池的實現，更改成每1個小時，強制更換線程池爲新的線程池，舊的線程數可以自動釋放。

 

6.文件泄露

      您會發現，隨着請求的session變多，spark會在hdfs和本地磁盤建立海量的磁盤目錄，最終會由於本地磁盤與hdfs上的目錄過多，而致使文件系統和整個文件系統癱瘓。在YDB裏面咱們針對這種狀況也作了處理。

 

7.deleteONExit內存泄露

 

 

 

 

 

爲何會有這些對象在裏面，咱們看下源碼

 

 

 

 

 

 

 

 

8.JDO內存泄露

多達10萬多個JDOPersistenceManager

 

 

 

 

 

 

 

 

 

 

 

 

 

9.listerner內存泄露

經過debug工具監控發現，spark的listerner隨着時間的積累，通知(post)速度運來越慢

發現全部代碼都卡在了onpostevent上

 

 

 

 

 

jstack的結果以下

 

 

研究下了調用邏輯以下，發現是循環調用listerners，並且listerner都是空執行纔會產生上面的jstack截圖

 

 

經過內存發現有30多萬個linterner在裏面

 

 

發現都是大多數都是同一個listener,咱們覈對下該處源碼

 

 

最終定位問題

確係是這個地方的BUG ，每次建立JDBC鏈接的時候 ，spark就會增長一個listener， 時間久了，listener就會積累愈來愈多  針對這個問題 我簡單的修改了一行代碼，開始進入下一輪的壓測

 

 

 

 

二12、spark源碼調優

      測試發現，即便只有1條記錄，使用 spark進行一次SQL查詢也會耗時1秒，對不少即席查詢來講1秒的等待，對用戶體驗很是不友好。針對這個問題，咱們在spark與hive的細節代碼上進行了局部調優，調優後，響應時間由原先的1秒縮減到如今的200~300毫秒。

      

如下是咱們改動過的地方

1.SessionState 的建立目錄 佔用較多的時間

 

 

另外使用Hadoop namenode HA的同窗會注意到，若是第一個namenode是standby狀態，這個地方會更慢，就不止一秒，因此除了改動源碼外，若是使用namenode ha的同窗必定要注意，將active狀態的node必定要放在前面。

2.HiveConf的初始化過程佔用太多時間

頻繁的hiveConf初始化，須要讀取core-default.xml,hdfs-default.xml,yarn-default.xml

,mapreduce-default.xml,hive-default.xml等多個xml文件，而這些xml文件都是內嵌在jar包內的。

第一，解壓這些jar包須要耗費較多的時間，第二每次都對這些xml文件解析也耗費時間。

 

 

 

 

 

 

 

 

 

 

 

 

3.廣播broadcast傳遞的hadoop configuration序列化很耗時

lconfiguration的序列化，採用了壓縮的方式進行序列化，有全局鎖的問題

lconfiguration每次序列化，傳遞了太多了沒用的配置項了，1000多個配置項，佔用60多Kb。咱們剔除了不是必須傳輸的配置項後，縮減到44個配置項，2kb的大小。

 

 

 

 

 

 

4.對spark廣播數據broadcast的Cleaner的改進

 

因爲SPARK-3015 的BUG，spark的cleaner 目前爲單線程回收模式。

你們留意spark源碼註釋

 

 

 

其中的單線程瓶頸點在於廣播數據的cleaner，因爲要跨越不少臺機器，須要經過akka進行網絡交互。

若是回收併發特別大，SPARK-3015 的bug報告會出現網絡擁堵，致使大量的 timeout出現。

爲何回收量特變大呢？ 實際上是由於cleaner 本質是經過system.gc（）,按期執行的，默認積累30分鐘或者進行了gc後才觸發cleaner,這樣就會致使瞬間，大量的akka併發執行，集中釋放，網絡不瞬間癱瘓纔不怪呢。 可是單線程回收意味着回收速度恆定，若是查詢併發很大，回收速度跟不上cleaner的速度，會致使cleaner積累不少，會致使進程OOM（YDB作了修改，會限制前臺查詢的併發）。 不管是OOM仍是限制併發都不是咱們但願看到的，因此針對高併發狀況下，這種單線程的回收速度是知足不了高併發的需求的。 對於官方的這樣的作法，咱們表示並非一個完美的cleaner方案。併發回收必定要支持，只要解決akka的timeout問題便可。 因此這個問題要仔細分析一下，akka爲何會timeout，是由於cleaner佔據了太多的資源，那麼咱們是否能夠控制下cleaner的併發呢？好比說使用4個併發，而不是默認將所有的併發線程都給佔滿呢？這樣及解決了cleaner的回收速度，也解決了akka的問題不是更好麼？ 針對這個問題，咱們最終仍是選擇了修改spark的ContextCleaner對象，將廣播數據的回收 改爲多線程的方式，但如今了線程的併發數量，從而解決了該問題。