做Data Mining，其實大部分時間都花在清洗數據

• 前言：很多初學的朋友對大數據挖掘第一直觀的印象，都只是業務模型，以及組成模型背後的各種算法原理。往往忽視了整個業務場景建模過程中，看似最普通，卻又最精髓的特征數據清洗。可謂是平平無奇，卻又一掌定乾坤，稍有閃失，足以功虧一簣。

大數據圈裏的一位掃地僧

說明：這篇文章很早就想寫了，但是切入點一直拿捏不準，要講的內容比較大眾化，卻又是重中之重。

一、數據清洗的那些事

構建業務模型，在確定特征向量以後，都需要準備特征數據在線下進行訓練、驗證和測試。同樣，部署發布離線場景模型，也需要每天定時跑P加工模型特征表。

而這一切要做的事，都離不開數據清洗，業內話來說，也就是ETL處理(抽取Extract、轉換Transform、加載Load)，三大法寶。

來自於百度百科

在大數據圈裏和圈外，很多朋友都整理過數據，我們這裏稱為清洗數據。

不管你是叱咤風雲的Excel大牛，還是玩轉SQL的數據庫的能人，甚至是專註HQL開發ETL工程師，以及用MapReduce\Scala語言處理復雜數據的程序猿。(也許你就是小白一個)

我想說的是，解決問題的技術有高低，但是解決問題的初衷只有一個——把雜亂的數據清洗幹凈，讓業務模型能夠輸入高質量的數據源。

不過，既然做的是大數據挖掘，面對的至少是G級別的數據量(包括用戶基本數據、行為數據、交易數據、資金流數據以及第三方數據等等)。那麽選擇正確的方式來清洗特征數據就極為重要，除了讓你事半功倍，還至少能夠保證你在方案上是可行的。

你可別告訴我，你仍然選擇用Excel，那我選擇狗帶。

二、大數據的必殺技

在大數據生態圈裏，有著很多開源的數據ETL工具，每一種都私下嘗嘗鮮也可以。但是對於一個公司內部來說，穩定性、安全性和成本都是必須考慮的。

就拿Spark Hive和Hive來說，同樣是在Yarn上來跑P，而且替換任務的執行引擎也很方便。

修改任務執行引擎

的確，Spark的大多數任務都會比MapReduce執行效率要快差不多1/3時間。但是，Spark對內存的消耗是很大的，在程序運行期間，每個節點的負載都很高，隊列資源消耗很多。因此，我每次提交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作業的性能。
• num-executors是用於設置Spark作業總共要用多少個Executor進程來執行。這個參數如果不設置，默認啟動少量的Executor進程，會很大程度影響任務執行效率。

單獨的提交Spark任務，優化參數還可以解決大部分運行問題。但是完全替換每天跑P加工報表的執行引擎，從MapReduce到Spark，總會遇到不少意想不到的問題。對於一個大數據部門而言，另可效率有所延遲，但是數據穩定性是重中之重。

Spark運行Stage

所以，大部分數據處理，甚至是業務場景模型每天的數據清洗加工，都會優先考慮Hive基於MapRedcue的執行引擎，少部分會單獨使用編寫MapReduce、Spark程序來進行復雜處理。

三、實踐中的數據清洗

這節要介紹的內容其實很多，單獨對於Hive這方面，就包括執行計劃、常用寫法、內置函數、一些自定義函數，以及優化策略等等。

幸運的是，這方面資源在網上很全，這是一個值得欣慰的點，基本遇到的大多數問題都能夠搜到滿意答案。

因此，文章這個版塊主要順著這條主線來——(我在大數據挖掘實踐中所做的模型特征清洗)，這樣對於大數據挖掘的朋友們來說，更具有針對性。

3.1 知曉數據源

(這裏不擴展數據源的抽取和行為數據的埋點)

大數據平臺的數據源集中來源於三個方面，按比重大小來排序：

60%來源於關系數據庫的同步遷移： 大多數公司都是采用MySQL和Oracle，就拿互聯網金融平臺來說，這些數據大部分是用戶基本信息，交易數據以及資金數據。

30%來源於平臺埋點數據的采集：渠道有PC、Wap、安卓和IOS，通過客戶端產生請求，經過Netty服務器處理，再進Kafka接受數據並解碼，最後到Spark Streaming劃分為離線和實時清洗。

10%來源於第三方數據：做互聯網金融都會整合第三方數據源，大體有工商、快消、車房、電商交易、銀行、運營商等等，有些是通過正規渠道來購買(已脫敏)，大部分數據來源於黑市(未脫敏)。這個市場魚龍混雜、臭氣熏天，很多真實數據被註入了汙水，在這基礎上建立的模型可信度往往很差。

得數據，得天下?

3.2 業務場景模型的背景

看過我以前文章集的朋友都知道一點，我致力於做大數據產品。

在之前開發數據產品的過程中，有一次規劃了一個頁面——用戶關系網絡，底層是引用了一個組合模型。

簡單來說是對用戶群體細分，判斷用戶屬於那一類別的羊毛黨群體，再結合業務運營中的彈性因子去綜合評估用戶的風險。

截圖的原型Demo

大家看到這幅圖會有什麽想法?

簡單來說，原型展示的是分析兩個用戶之間在很多維度方面的關聯度

當時這個功能在後端開發過程中對於特征數據的處理花了很多時間，有一部分是數據倉庫工具HQL所不能解決的，而且還需要考慮完整頁面(截圖只是其中一部分)查詢的響應時間，這就得預先標準化業務模型的輸出結果。

我可以簡單描述下需求場景：

• 拿IP地址來說，在最近30天範圍內，用戶使用互聯網金融平臺，不管是PC端，還是無線端，每個用戶每個月都會產生很多IP數據集。
• 對於擁有千萬級別用戶量的平臺，肯定會出現這樣的場景——很多用戶在最近一個月內都使用過相同的IP地址，而且數量有多有少。
• 對某個用戶來說，他就好像是一個雪花中的焦點，他使用過的IP地址就像雪花一樣圍繞著他。而每個IP地址都曾被很多用戶使用過。

簡單來說，IP地址只是一個媒介，連接著不同用戶。——你中有我，我中有你。

雪花狀

有了上面的背景描述，那麽就需要每個讀者都去思考下這三個問題：

問題一、如何先通過某個用戶最近30天的IP列表去找到使用相同IP頻數最多的那一批用戶列表呢?

問題二、如何結合關系網絡的每個維度(IP、設備指紋、身份證、銀行卡和加密隱私等等)，去挖掘與該用戶關聯度最高的那一批用戶列表?

問題三、如何對接產品標準化模型輸出，讓頁面查詢的效應時間變得更快些?

思考就像吃大理核桃般，總是那麽耐人尋味。

3.3 學會用Hive解決70%的數據清洗

對於70%的數據清洗都可以使用Hive來完美解決，而且網絡參考資料也很全，所以大多數場景我都推薦用它來清洗。——高效、穩定

一只小蜜蜂呀，飛到花叢中

不過在使用過程中，我有兩點建議送給大家，就當作雞年禮物吧。

第一點建議：要學會顧全大局，不要急於求成，學會把復雜的查詢拆開寫，多考慮集群整個資源總量和並發任務數。

第二點建議：心要細，在線下做好充足的測試，確保安全性、邏輯正確和執行效率才能上線。

禮物也送了，繼續介紹

對於上述的用戶關系網絡場景，這裏舉IP維度來實踐下，如何利用Hive進行數據清洗。

下面是用戶行為日誌表的用戶、IP地址和時間數據結構。

用戶、IP和時間

回到上面的第一個思考， 如何先通過某個用戶最近30天的IP列表去找到使用相同IP頻數最多的那一批用戶列表呢?

我當時采取了兩個步驟。

步驟一：清洗最近30天所有IP對應的用戶列表，並去重用戶

清洗IP對應的用戶列表

這裏解釋三個內置函數concat_ws、collect_set和cast，先更了解必須去親自實踐：

• concat_ws，它是用來分隔符字符串連接函數。
• collect_set，它是用來將一列多行轉換成一行多列，並去重用戶。
• cast，它是用來轉換字段數據類型。

果然很方便吧，下面是第一個步驟的執行結果。

IP馬賽克

步驟二：清洗用戶在IP媒介下，所有關聯的用戶集列表

清洗用戶在IP媒介下的關聯用戶集

最終對於IP媒介清洗的數據效果如下所示：

用戶ID、IP關聯的用戶集

同理對於其他維度的媒介方法一樣，到這一步，算是完成Hive階段的初步清洗，是不是很高效。

最終結果的樣式

但是對於分析用戶細分來說，還需要借助MapReduce，或者Scala來深層次處理特征數據。

3.4 使用Scala來清洗特殊的數據

對於使用Spark框架來清洗數據，我一般都是處於下面兩個原因：

• 常規的HQL解決不了
• 用簡潔的代碼高效計算，也就是考慮開發成本和執行效率

對於部署本機的大數據挖掘環境，可以查看這兩篇文章來實踐動手下：

• 《簡單之極，搭建屬於自己的Data Mining環境(Spark版本)》
• 《深入淺出，在Data Mining環境下Code第一個算法(Spark版本)》

工欲善其事，必先利其器。有了這麽好的利器，處理復雜的特征數據，那都是手到擒來。

借助於Hive清洗處理後的源數據，我們繼續回到第二個思考—— 如何結合關系網絡的每個維度，去初步挖掘與該用戶關聯度最高的那一批用戶列表?

看到這個問題，又產生了這幾個思考：

• 目前有五個維度，以後可能還會更多，純手工顯然不可能，再使用Hive好像也比較困難。
• 每個維度的關聯用戶量也不少，所以基本每個用戶每行數據的處理采用單機串行的程序去處理顯然很緩慢。不過每行的處理是獨立性的。
• 同一個關聯用戶會在同一個維度，以及每一個維度出現多次，還需要進行累計。

如果才剛剛處理大數據挖掘，遇到這樣的問題的確很費神，就連你們常用的Python和R估計也難拯救你們。但是如果實戰比較多，這樣的獨立任務，完全可以並發到每臺計算節點上去每行單獨處理，而我們只需要在處理每行時，單獨調用清洗方法即可。

這裏我優先推薦使用Spark來清洗處理(後面給一個MapReduce的邏輯)，整個核心過程主要有三個板塊

• 預處理，對所有關聯用戶去重，並統計每個關聯用戶在每個維度的累計次數

核心就在這兩處

• 評分，循環上述關聯用戶集，給關聯度打一個分

核心在這三處

• 標準化清洗處理，用戶關聯用json串拼接

第二個階段清洗結果

得到上面清洗結果，我們才能更好的作為模型的源數據輸出，感覺是不是很費神，所以才印證了這句話——做Data Mining，其實大部分時間都花在清洗數據

3.5 附加分：使用MapReduce來清洗特殊的數據

針對上述的數據清洗，同樣可以MapReduce來單獨處理。只是開發效率和執行效率有所影響。

當然也不排除適用於MapReduce處理的復雜數據場景。

對於在本地Windows環境寫MapRecue代碼，可以借鑒上述文章中部署的數據挖掘環境，修改下Maven工程的pom.xml文件就可以了。

在pom.xml文件添加hadoop依賴

而我在以往做大數據挖掘的過程裏，也有不少場景需要借助MR來處理，比如很早的一篇文章《一種新思想去解決大矩陣相乘》，甚至是大家比較常見的數據傾斜——特別是處理平臺行為日誌數據，特別容易遇到數據傾斜。

這裏提供一個上述Spark清洗數據的MR代碼邏輯，大家可以對比看看與Spark代碼邏輯的差異性。

• Map階段

邏輯思路

• Reduce階段(這裏用不上)

Reduce階段的框架

• Drive階段

驅動階段的配置

上面這三個階段就是MR任務常規的流程，處理上述問題的思路其實和Spark邏輯差不多。只是這套框架性代碼量太多，有很多重復性，每寫一個MR任務的工作量也會比較大，執行效率我並沒有去測試作比較。

如果Spark跑線上任務模型會出現不穩定的話，我想以後我還是會遷移到MapReduce上去跑離線模型。拭目以待吧!

總結

說到這裏，整篇文章概括起來有三點：

• 講述了數據清洗在業務場景建模過程中的重要性和流程操作。
• 介紹了兩款主流計算框架的適用場景和差異性。
• 更列舉了不同數據處理工具在每個業務場景下的優勢和不同。

但是，還是那麽一句話——使用什麽技術不在乎，我更迷戀業務場景驅動下的技術挑戰。

與你溝通最關鍵的，也許會是直屬領導，也許會是業務運營人員，甚至是完全不懂技術的客戶。他們最關心的是你在業務層面上的技術方案能否解決業務痛點問題。

所以，做大數據挖掘要多關心業務，別一味只談技術。

作者介紹

汪榕，3年場景建模經驗，曾累計獲得8次數學建模一等獎，包括全國大學生國家一等獎，在國內期刊發表過相關學術研究。兩年電商數據挖掘實踐，負責開發精準營銷產品中的用戶標簽體系。發表過數據挖掘相關的多篇文章。目前在互聯網金融行業從事數據挖掘工作，參與開發反欺詐實時監控系統。微博：樂平汪二。

做Data Mining，其實大部分時間都花在清洗數據