2016年大數據Spark「蘑菇雲」行動代碼學習之AdClickedStreamingStats模塊分析

2016年大數據Spark「蘑菇雲」行動代碼學習之AdClickedStreamingStats模塊分析

    系統背景：用戶使用終端設備（IPAD、手機、瀏覽器）等登陸系統，系統採用js腳本發送用戶信息和廣告點擊信息到後臺日誌，進入flume監控，經過kafka消息中間件傳輸數據，由Spark Streaming消費後將信息存儲到後臺。本模塊主要就是實現將kafka發送的信息進行過濾，動態更新黑名單，生成有效的廣告點擊數據，造成廣告點擊趨勢，將相關信息經過數據庫鏈接池寫入數據庫MySql。

 

    實現思路：因爲kafka傳輸的廣告點擊數據中有一些是無效數據，須要根據必定的規則進行過濾（本方案採起簡單的黑名單過濾的方式進行過濾，還能夠採用機器學習的方式進行過濾——能夠作一個大課題研究、大數據黃金崗位、下一步重點方向）。

 

在造成有效的廣告點擊數據後，按照天爲單位，對廣告點擊次數進行累計，同時按照一天點擊次數超過？次造成新的黑名單，持久化到後臺。

 

對廣告按照時間、省份、城市進行點擊次數更新，而後插入或更新到MySql數據庫。

 

由DataFrame的GroupBy，計算出天天每一個省份的Top5排名的廣告。

 

採用窗口函數reduceByKeyAndWindow計算過去半個小時內廣告點擊的趨勢。

 

    具體的流程以下：

1. AdClickedStreamingStats模塊流程圖

   
   

2. 初始化Spark的運行環境，加載必要的JAR包

   
   

3. 設置kafka的參數（broker地址列表、Topics）

   
   

4. 從kafka獲取數據（time+IP+UserID+AdID+province+city)

   
   

5. 從數據庫獲取黑名單信息（UserID），生成黑名單RDD，格式爲map(UserID,true）

   
   

6. 將廣告點擊數據按（UserID,字串流）造成廣告點擊RDD

   
   

7. 廣告點擊RDD和黑名單RDD左外鏈接，造成UserID,(字串流,true或者爲空)的RDD

   
   

8. 重新的RDD過濾出用戶名不在黑名單上的廣告點擊RDD（UserID,字串流)

   
   

9. 累加有效的廣告點擊次數RDD

   （"time_IP_userID_adID_province_city"，long）

   
   

10. 根據一天內的點擊次數，來過濾出有效的廣告點擊數

    （代碼中一天內大於1次就過濾掉了？）

    
    

11. 造成用戶廣告點擊ListBuffer

    
    

12. 插入或者更新用戶廣告點擊數據

    （當天已有記錄則將廣告點擊數累加後更新，不然插入）

    
    

13. 根據當天、用戶、廣告，得到總的點擊數，

    若是大於81次，造成新的黑名單

    
    

14. 根據用戶ID，進行黑名單去重操做並寫入

    
    

15. 按照日期+廣告ID+省+市，造成新的廣告點擊RDD

    
    

16. updateStateByKey，更新狀態？

    
    

17. 寫入或更新廣告點擊數據（日期+廣告+省+市+點擊數）

    
    

18. 按省進行Group by，而後查詢出點擊排名前5的廣告

    
    

19. 將當天+廣告的TOP5數據庫清空，而後插入新的數據

    
    

20. 計算過去半小時廣告點擊的趨勢

    造成（廣告ID+Time)的DStream

    
    

21. 根據滑動窗口（半小時、5分鐘），計算出廣告點擊趨勢

    
    

22. 寫入廣告點擊趨勢數據