摘要：很多初學的朋友對大數據挖掘第一直觀的印象，都只是業(yè)務模型，以及組成模型背后的各種算法原理。
很多初學的朋友對大數據挖掘第一直觀的印象，都只是業(yè)務模型，以及組成模型背后的各種算法原理。往往忽視了整個業(yè)務場景建模過程中，看似最普通，卻又最精髓的特征數據清洗。可謂是平平無奇，卻又一掌定乾坤，稍有閃失，足以功虧一簣。
前言：很多初學的朋友對大數據挖掘第一直觀的印象，都只是業(yè)務模型，以及組成模型背后的各種算法原理。往往忽視了整個業(yè)務場景建模過程中，看似最普通，卻又最精髓的特征數據清洗。可謂是平平無奇，卻又一掌定乾坤，稍有閃失，足以功虧一簣。
說明：這篇文章很早就想寫了，但是切入點一直拿捏不準，要講的內容比較大眾化，卻又是重中之重。
一、數據清洗的那些事
構建業(yè)務模型，在確定特征向量以后，都需要準備特征數據在線下進行訓練、驗證和測試。同樣，部署發(fā)布離線場景模型，也需要每天定時跑P加工模型特征表。
而這一切要做的事，都離不開數據清洗，業(yè)內話來說，也就是ETL處理(抽取Extract、轉換Transform、加載Load)，三大法寶。
來自于百度百科
在大數據圈里和圈外，很多朋友都整理過數據，我們這里稱為清洗數據。
不管你是叱咤風云的Excel大牛，還是玩轉SQL的數據庫的能人，甚至是專注HQL開發(fā)ETL工程師，以及用MapReduceScala語言處理復雜數據的程序猿。(也許你就是小白一個)
我想說的是，解決問題的技術有高低，但是解決問題的初衷只有一個——把雜亂的數據清洗干凈，讓業(yè)務模型能夠輸入高質量的數據源。
不過，既然做的是大數據挖掘，面對的至少是G級別的數據量(包括用戶基本數據、行為數據、交易數據、資金流數據以及第三方數據等等)。那么選擇正確的方式來清洗特征數據就極為重要，除了讓你事半功倍，還至少能夠保證你在方案上是可行的。
你可別告訴我，你仍然選擇用Excel，那我選擇狗帶。
二、大數據的必殺技
在大數據生態(tài)圈里，有著很多開源的數據ETL工具，每一種都私下嘗嘗鮮也可以。但是對于一個公司內部來說，穩(wěn)定性、安全性和成本都是必須考慮的。
就拿Spark Hive和Hive來說，同樣是在Yarn上來跑P，而且替換任務的執(zhí)行引擎也很方便。
修改任務執(zhí)行引擎
的確，Spark的大多數任務都會比MapReduce執(zhí)行效率要快差不多1/3時間。但是，Spark對內存的消耗是很大的，在程序運行期間，每個節(jié)點的負載都很高，隊列資源消耗很多。因此，我每次提交Spark離線模型跑任務時，都必須設置下面的參數，防止占用完集群所有資源。
spark-submit --master yarn-cluster --driver-memory 5g --executor-memory 2g --num-executors 20
其中：
· driver-memory是用于設置Driver進程的內存，一般不設置，或者1G。我這里調整到5G是因為RDD的數據全部拉取到Driver上進行處理，那要確保Driver的內存足夠大，否則會出現OOM內存溢出。
· executor-memory是用于設置每個Executor進程的內存。Executor內存的大小決定了Spark作業(yè)的性能。
· num-executors是用于設置Spark作業(yè)總共要用多少個Executor進程來執(zhí)行。這個參數如果不設置，默認啟動少量的Executor進程，會很大程度影響任務執(zhí)行效率。
單獨的提交Spark任務，優(yōu)化參數還可以解決大部分運行問題。但是完全替換每天跑P加工報表的執(zhí)行引擎，從MapReduce到Spark，總會遇到不少意想不到的問題。對于一個大數據部門而言，另可效率有所延遲，但是數據穩(wěn)定性是重中之重。
Spark運行Stage
所以，大部分數據處理，甚至是業(yè)務場景模型每天的數據清洗加工，都會優(yōu)先考慮Hive基于MapRedcue的執(zhí)行引擎，少部分會單獨使用編寫MapReduce、Spark程序來進行復雜處理。
三、實踐中的數據清洗
這節(jié)要介紹的內容其實很多，單獨對于Hive這方面，就包括執(zhí)行計劃、常用寫法、內置函數、一些自定義函數，以及優(yōu)化策略等等。
幸運的是，這方面資源在網上很全，這是一個值得欣慰的點，基本遇到的大多數問題都能夠搜到滿意答案。
因此，文章這個版塊主要順著這條主線來——(我在大數據挖掘實踐中所做的模型特征清洗)，這樣對于大數據挖掘的朋友們來說，更具有針對性。
3.1 知曉數據源
(這里不擴展數據源的抽取和行為數據的埋點)
大數據平臺的數據源集中來源于三個方面，按比重大小來排序：
60%來源于關系數據庫的同步遷移： 大多數公司都是采用MySQL和Oracle，就拿互聯網金融平臺來說，這些數據大部分是用戶基本信息，交易數據以及資金數據。
30%來源于平臺埋點數據的采集：渠道有PC、Wap、安卓和IOS，通過客戶端產生請求，經過Netty服務器處理，再進Kafka接受數據并解碼，最后到Spark Streaming劃分為離線和實時清洗。
10%來源于第三方數據：做互聯網金融都會整合第三方數據源，大體有工商、快消、車房、電商交易、銀行、運營商等等，有些是通過正規(guī)渠道來購買(已脫敏)，大部分數據來源于黑市(未脫敏)。這個市場魚龍混雜、臭氣熏天，很多真實數據被注入了污水，在這基礎上建立的模型可信度往往很差。
得數據，得天下?
3.2 業(yè)務場景模型的背景
看過我以前文章集的朋友都知道一點，我致力于做大數據產品。
在之前開發(fā)數據產品的過程中，有一次規(guī)劃了一個頁面——用戶關系網絡，底層是引用了一個組合模型。
簡單來說是對用戶群體細分，判斷用戶屬于那一類別的羊毛黨群體，再結合業(yè)務運營中的彈性因子去綜合評估用戶的風險。
截圖的原型Demo
大家看到這幅圖會有什么想法?
| 簡單來說，原型展示的是分析兩個用戶之間在很多維度方面的關聯度
當時這個功能在后端開發(fā)過程中對于特征數據的處理花了很多時間，有一部分是數據倉庫工具HQL所不能解決的，而且還需要考慮完整頁面(截圖只是其中一部分)查詢的響應時間，這就得預先標準化業(yè)務模型的輸出結果。
我可以簡單描述下需求場景：
· 拿IP地址來說，在最近30天范圍內，用戶使用互聯網金融平臺，不管是PC端，還是無線端，每個用戶每個月都會產生很多IP數據集。
· 對于擁有千萬級別用戶量的平臺，肯定會出現這樣的場景——很多用戶在最近一個月內都使用過相同的IP地址，而且數量有多有少。
· 對某個用戶來說，他就好像是一個雪花中的焦點，他使用過的IP地址就像雪花一樣圍繞著他。而每個IP地址都曾被很多用戶使用過。
簡單來說，IP地址只是一個媒介，連接著不同用戶。——你中有我，我中有你。
雪花狀
有了上面的背景描述，那么就需要每個讀者都去思考下這三個問題：
問題一、如何先通過某個用戶最近30天的IP列表去找到使用相同IP頻數最多的那一批用戶列表呢?
問題二、如何結合關系網絡的每個維度(IP、設備指紋、身份證、銀行卡和加密隱私等等)，去挖掘與該用戶關聯度最高的那一批用戶列表?
問題三、如何對接產品標準化模型輸出，讓頁面查詢的效應時間變得更快些?
| 思考就像吃大理核桃般，總是那么耐人尋味。
3.3 學會用Hive解決70%的數據清洗
對于70%的數據清洗都可以使用Hive來完美解決，而且網絡參考資料也很全，所以大多數場景我都推薦用它來清洗。——高效、穩(wěn)定
一只小蜜蜂呀，飛到花叢中
不過在使用過程中，我有兩點建議送給大家，就當作雞年禮物吧。
第一點建議：要學會顧全大局，不要急于求成，學會把復雜的查詢拆開寫，多考慮集群整個資源總量和并發(fā)任務數。
第二點建議：心要細，在線下做好充足的測試，確保安全性、邏輯正確和執(zhí)行效率才能上線。
| 禮物也送了，繼續(xù)介紹
對于上述的用戶關系網絡場景，這里舉IP維度來實踐下，如何利用Hive進行數據清洗。
下面是用戶行為日志表的用戶、IP地址和時間數據結構。
用戶、IP和時間
回到上面的第一個思考，如何先通過某個用戶最近30天的IP列表去找到使用相同IP頻數最多的那一批用戶列表呢?
我當時采取了兩個步驟。
· 步驟一：清洗最近30天所有IP對應的用戶列表，并去重用戶
清洗IP對應的用戶列表
這里解釋三個內置函數concat_ws、collect_set和cast，先更了解必須去親自實踐：
concat_ws，它是用來分隔符字符串連接函數。
collect_set，它是用來將一列多行轉換成一行多列，并去重用戶。
cast，它是用來轉換字段數據類型。
果然很方便吧，下面是第一個步驟的執(zhí)行結果。
IP馬賽克
· 步驟二：清洗用戶在IP媒介下，所有關聯的用戶集列表
清洗用戶在IP媒介下的關聯用戶集
最終對于IP媒介清洗的數據效果如下所示：
用戶ID、IP關聯的用戶集
同理對于其他維度的媒介方法一樣，到這一步，算是完成Hive階段的初步清洗，是不是很高效。
3.4 使用Scala來清洗特殊的數據
對于使用Spark框架來清洗數據，我一般都是處于下面兩個原因：
· 常規(guī)的HQL解決不了
· 用簡潔的代碼高效計算，也就是考慮開發(fā)成本和執(zhí)行效率
工欲善其事，必先利其器。有了這么好的利器，處理復雜的特征數據，那都是手到擒來。
借助于Hive清洗處理后的源數據，我們繼續(xù)回到第二個思考——如何結合關系網絡的每個維度，去初步挖掘與該用戶關聯度最高的那一批用戶列表?
看到這個問題，又產生了這幾個思考：
· 目前有五個維度，以后可能還會更多，純手工顯然不可能，再使用Hive好像也比較困難。
· 每個維度的關聯用戶量也不少，所以基本每個用戶每行數據的處理采用單機串行的程序去處理顯然很緩慢。不過每行的處理是獨立性的。
· 同一個關聯用戶會在同一個維度，以及每一個維度出現多次，還需要進行累計。
如果才剛剛處理大數據挖掘，遇到這樣的問題的確很費神，就連你們常用的Python和R估計也難拯救你們。但是如果實戰(zhàn)比較多，這樣的獨立任務，完全可以并發(fā)到每臺計算節(jié)點上去每行單獨處理，而我們只需要在處理每行時，單獨調用清洗方法即可。
這里我優(yōu)先推薦使用Spark來清洗處理(后面給一個MapReduce的邏輯)，整個核心過程主要有三個板塊
· 預處理，對所有關聯用戶去重，并統(tǒng)計每個關聯用戶在每個維度的累計次數
核心就在這兩處
· 評分，循環(huán)上述關聯用戶集，給關聯度打一個分
核心在這三處
· 標準化清洗處理，用戶關聯用json串拼接
第二個階段清洗結果
得到上面清洗結果，我們才能更好的作為模型的源數據輸出，感覺是不是很費神，所以才印證了這句話——做Data Mining，其實大部分時間都花在清洗數據
3.5 附加分：使用MapReduce來清洗特殊的數據
針對上述的數據清洗，同樣可以MapReduce來單獨處理。只是開發(fā)效率和執(zhí)行效率有所影響。
當然也不排除適用于MapReduce處理的復雜數據場景。
對于在本地Windows環(huán)境寫MapRecue代碼，可以借鑒上述文章中部署的數據挖掘環(huán)境，修改下Maven工程的pom.xml文件就可以了。
在pom.xml文件添加hadoop依賴
而我在以往做大數據挖掘的過程里，也有不少場景需要借助MR來處理，比如很早的一篇文章《一種新思想去解決大矩陣相乘》，甚至是大家比較常見的數據傾斜——特別是處理平臺行為日志數據，特別容易遇到數據傾斜。
這里提供一個上述Spark清洗數據的MR代碼邏輯，大家可以對比看看與Spark代碼邏輯的差異性。
· Map階段
邏輯思路
· Reduce階段(這里用不上)
Reduce階段的框架
· Drive階段
驅動階段的配置
上面這三個階段就是MR任務常規(guī)的流程，處理上述問題的思路其實和Spark邏輯差不多。只是這套框架性代碼量太多，有很多重復性，每寫一個MR任務的工作量也會比較大，執(zhí)行效率我并沒有去測試作比較。
如果Spark跑線上任務模型會出現不穩(wěn)定的話，我想以后我還是會遷移到MapReduce上去跑離線模型。拭目以待吧!
總結
說到這里，整篇文章概括起來有三點：
· 講述了數據清洗在業(yè)務場景建模過程中的重要性和流程操作。
· 介紹了兩款主流計算框架的適用場景和差異性。
· 更列舉了不同數據處理工具在每個業(yè)務場景下的優(yōu)勢和不同。
但是，還是那么一句話——使用什么技術不在乎，我更迷戀業(yè)務場景驅動下的技術挑戰(zhàn)。
與你溝通最關鍵的，也許會是直屬領導，也許會是業(yè)務運營人員，甚至是完全不懂技術的客戶。他們最關心的是你在業(yè)務層面上的技術方案能否解決業(yè)務痛點問題。
所以，做大數據挖掘要多關心業(yè)務，別一味只談技術。
但是對于分析用戶細分來說，還需要借助MapReduce，或者Scala來深層次處理特征數據。
京公網安備 11010502049343號