個推 Spark實踐教你繞過開發那些“坑”
Spark核心概念簡介
1、RDD即彈性分布式數據集，通過RDD可以執行各種算子實現數據處理和計算。比如用Spark做統計詞頻，即拿到一串文字進行WordCount，可以把這個文字數據load到RDD之後，調用map、reducebyKey 算子，最後執行count動作觸發真正的計算。
2、寬依賴和窄依賴。工廠裏面有很多流水線，一款產品上遊有一個人操作，下遊有人進行第二個操作，窄依賴和這個很類似，下遊依賴上遊。而所謂寬依賴類似於有多條流水線，A流水線的一個操作是需要依賴一條流水線B，才可以繼續執行，要求兩條流水線之間要做材料運輸，做協調，但效率低。
從上圖可以看到，如果B只依賴A則是一種窄依賴。像圖中這種reduceByKey的操作，就是剛剛舉例的寬依賴，類似於多條流水線之間某一些操作相互依賴，如：F對E、B的依賴。寬依賴最大的問題是會導致洗牌過程。
Spark Streaming介紹
流式計算，即數據生成後，實時對數據進行處理。Spark 是一個批處理框架，那它如何實現流式處理？Spark 是把數據裁成一段一段的處理,即一個數據流離散化成許多個連續批次，然後Spark對每個批次進行處理。
個推為什麽選擇Spark？
1、Spark 比較適合叠代計算，解決我們團隊在之前使用hadoop mapreduce叠代數據計算這一塊的瓶頸。
2、Spark是一個技術棧，但可以做很多類型的數據處理：批處理，SQL，流式處理以及ML等，基本滿足我們團隊當時的訴求。
3、它的API抽象層次非常高，通過使用map、reduce、groupby等多種算子可快速實現數據處理，極大降低開發成本，並且靈活。另外Spark框架對於多語言支持也是非常好，很多負責數據挖掘算法同學對於python 熟悉，而工程開發的同學熟悉java， 多語言支持可以把開發和分析的同學快速地引入過來。
4、在2014年的時候，我們用hadoop Yarn，而Spark可以在Yarn部署起來，使用Spark大大降低了切換成本，並且可以把之前的hadoop資源利用起來。
5、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 在個推業務上的具體使用現狀
1、個推做用戶畫像、模型叠代以及一些推薦的時候直接用了MLLib，MLLib集成了很多算法，非常方便。
2、個推有一個BI工具箱，讓一些運營人員提取數據，我們是用Spark SQL+Parquet格式寬表實現，Parquet是列式存儲格式，使用它你不用加載整個表，只會去加載關心那些字段，大大減少IO消耗。
3、實時統計分析這塊：例如個推有款產品叫個圖，就是使用Spark streaming 來實時統計。
4、復雜的 ETL 任務我們也使用 Spark。例如：我們個推推送報表這一塊，每天需要做很多維度的推送報表統計。使用 Spark 通過 cache 中間結果緩存，然後再統計其他維度，大大地減少了 I/O 消耗，顯著地提升了統計處理速度。
個推Spark實踐案例分享
上圖是個推熱力圖的處理架構。左邊這一側利用業務平臺得到設備的實時位置數據，通過Spark Streaming以及計算得到每一個geohash格子上的人數，然後統計結果實時傳輸給業務服務層，在push到客戶端地圖上面去渲染，最終形成一個實時熱力圖。Spark Streaming 主要用於數據實時統計處理上。
個推教你繞過開發那些坑
1、數據處理經常出現數據傾斜，導致負載不均衡的問題，需要做統計分析找到傾斜數據特征，定散列策略。
2、使用Parquet列式存儲，減少IO，提高Spark SQL效率。
3、實時處理方面：一方面要註意數據源（Kafka）topic需要多個partition，並且數據要散列均勻，使得Spark Streaming的Recevier能夠多個並行，並且均衡地消費數據 。使用Spark Streaming，要多通過Spark History 排查DStream的操作中哪些處理慢，然後進行優化。另外一方面我們自己還做了實時處理的監控系統，用來監控處理情況如流 入、流出數據速度等。通過監控系統報警,能夠方便地運維Spark Streaming 實時處理程序。這個小監控系統主要用了 influxdb+grafana 等實現。
4、我們測試網經常出現找不到第三方jar的情況，如果是用CDH的同學一般會遇到，就是在CDH 5.4開始，CDH的技術支持人員說他們去掉了hbase等一些jar，他們認那些jar已經不需要耦合在自己的classpath中，這個情況可以通過spark.executor.extraClassPath方式添加進來。
5、一些新入門的人會遇到搞不清transform和action，沒有明白transform是lazy的，需要action觸發，並且兩個action前後調用效果可能不一樣。
6、大家使用過程當中，對需要重復使用的RDD，一定要做cache，性能提升會很明顯。
以上內容根據個推講師袁凱在11月28號Segmentfault北京D-Day沙龍的分享整理，希望給廣大開發者一些啟示。小夥伴們記得將get到的幹貨順手分享給身邊的同學們哦~
個推 Spark實踐教你繞過開發那些“坑”