個推 Spark實踐教你繞過開發那些「坑」

     Spark做爲一個開源數據處理框架，它在數據計算過程當中把中間數據直接緩存到內存裏，能大大提升處理速度，特別是複雜的迭代計算。Spark主要包括SparkSQL，SparkStreaming，Spark MLLib以及圖計算。

 

 

Spark核心概念簡介

 

一、RDD即彈性分佈式數據集，經過RDD能夠執行各類算子實現數據處理和計算。好比用Spark作統計詞頻，即拿到一串文字進行WordCount，能夠把這個文字數據load到RDD以後，調用map、reducebyKey 算子，最後執行count動做觸發真正的計算。

 

二、寬依賴和窄依賴。工廠裏面有不少流水線，一款產品上游有一我的操做，下游有人進行第二個操做，窄依賴和這個很相似，下游依賴上游。而所謂寬依賴相似於有多條流水線，A流水線的一個操做是須要依賴一條流水線B，才能夠繼續執行，要求兩條流水線之間要作材料運輸，作協調，但效率低。

 

      從上圖能夠看到，若是B只依賴A則是一種窄依賴。像圖中這種reduceByKey的操做，就是剛剛舉例的寬依賴，相似於多條流水線之間某一些操做相互依賴，如：F對E、B的依賴。寬依賴最大的問題是會致使洗牌過程。

 

Spark Streaming介紹 

 

      流式計算，即數據生成後，實時對數據進行處理。Spark 是一個批處理框架，那它如何實現流式處理？Spark 是把數據裁成一段一段的處理,即一個數據流離散化成許多個連續批次，而後Spark對每一個批次進行處理。

 

 

 

個推爲何選擇Spark

 

一、Spark 比較適合迭代計算，解決咱們團隊在以前使用hadoop mapreduce迭代數據計算這一塊的瓶頸。

 

二、Spark是一個技術棧，但能夠作不少類型的數據處理：批處理，SQL，流式處理以及ML等，基本知足咱們團隊當時的訴求。

 

三、它的API抽象層次很是高，經過使用map、reduce、groupby等多種算子可快速實現數據處理，極大下降開發成本，而且靈活。另外Spark框架對於多語言支持也是很是好，不少負責數據挖掘算法同窗對於python 熟悉，而工程開發的同窗熟悉java， 多語言支持能夠把開發和分析的同窗快速地引入過來。

 

四、在2014年的時候，咱們用hadoop Yarn，而Spark能夠在Yarn部署起來，使用Spark大大下降了切換成本，而且能夠把以前的hadoop資源利用起來。

 

五、Spark在社區很火，找資料很是方便。

 

個推數據處理架構

 

 

      上圖是一個典型的lambda架構。主要分三層。上面藍色的框，是作離線批量處理，下面一層是實時數據處理這一塊，中間這一層是對於結果數據作一些存儲和檢索。

 

       有兩種方式導入數據到HDFS，一部分數據從業務平臺日誌收集寫入到 Kafka，而後直接Linkedin Camus（咱們作過擴展） 準實時地傳輸到 HDFS，另外部分數經過運維那邊的腳本定時導入到 HDFS 上。

 

       離線處理部分咱們仍是使用兩個方式（Hadoop MR 和 Spark）。原有的hadoop MR沒有放棄掉， 由於原來不少的工程已是用MR作的了，很是穩定，沒有必要推倒重來，只有部分迭代任務使用Spark 從新實現。另外Hive是直接能夠跟Spark作結合，Spark Sql中就可使用Hive的命令 。

 

個推Spark集羣的部署情況

 

      個推最開始用Spark是1.3.1版本，用的是刀片服務器,就是刀框裏面能夠塞 16 個刀片服務器，單個內存大小192G， CPU 核數是24 核的。在Spark官方也推薦用萬兆網卡，大內存設備。咱們權衡了需求和成本後，選擇了就用刀片機器來搭建 Spark集羣。刀框有個好處就是經過背板把刀片機器鏈接起來，傳輸速度快，相對成本小。部署模式上採用的是 Spark on Yarn，實現資源複用。

 

Spark 在個推業務上的具體使用

 

一、個推作用戶畫像、模型迭代以及一些推薦的時候直接用了MLLib，MLLib集成了不少算法，很是方便。

 

二、個推有一個BI工具箱，讓一些運營人員提取數據，咱們是用Spark SQL+Parquet格式寬表實現，Parquet是列式存儲格式，使用它你不用加載整個表，只會去加載關心那些字段，大大減小IO消耗。

 

三、實時統計分析這塊：例如個推有款產品叫個圖，就是使用Spark streaming 來實時統計。

 

四、複雜的 ETL 任務咱們也使用 Spark。例如：咱們個推推送報表這一塊，天天須要作不少維度的推送報表統計。使用 Spark 經過 cache 中間結果緩存，而後再統計其餘維度，大大地減小了 I/O 消耗，顯著地提高了統計處理速度。

 

個推Spark實踐案例分享

 

      上圖是個推熱力圖的處理架構。左邊這一側利用業務平臺獲得設備的實時位置數據，經過Spark Streaming以及計算獲得每個geohash格子上的人數，而後統計結果實時傳輸給業務服務層，在push到客戶端地圖上面去渲染，最終造成一個實時熱力圖。Spark Streaming 主要用於數據實時統計處理上。

 

 

個推教你繞過開發那些坑

 

一、數據處理常常出現數據傾斜，致使負載不均衡的問題，須要作統計分析找到傾斜數據特徵，定散列策略。

 

二、使用Parquet列式存儲，減小IO，提升Spark SQL效率。

 

三、實時處理方面：一方面要注意數據源（Kafka）topic須要多個partition，而且數據要散列均勻，使得Spark Streaming的Recevier可以多個並行，而且均衡地消費數據 。使用Spark Streaming，要多經過Spark History 排查DStream的操做中哪些處理慢，而後進行優化。另一方面咱們本身還作了實時處理的監控系統，用來監控處理狀況如流 入、流出數據速度等。經過監控系統報警,可以方便地運維Spark Streaming 實時處理程序。這個小監控系統主要用了 influxdb+grafana 等實現。

 

四、咱們測試網常常出現找不到第三方jar的狀況，若是是用CDH的同窗通常會遇到，就是在CDH 5.4開始，CDH的技術支持人員說他們去掉了hbase等一些jar，他們認那些jar已經不須要耦合在本身的classpath中，這個狀況能夠經過spark.executor.extraClassPath方式添加進來。

 

五、一些新入門的人會遇到搞不清transform和action，沒有明白transform是lazy的，須要action觸發，而且兩個action先後調用效果可能不同。

 

六、你們使用過程中，對須要重複使用的RDD，必定要作cache，性能提高會很明顯。