作者：王雅寧

轉載自：

0.序言

本文主要以基於AWS 搭建的EMR spark 託管叢集，使用pandas pyspark 對合作單位的業務資料進行ETL ---- EXTRACT（抽取）、TRANSFORM（轉換）、LOAD（載入） 等工作為例介紹大資料資料預處理的實踐經驗，很多初學的朋友對大資料探勘，資料分析第一直觀的印象，都只是業務模型，以及組成模型背後的各種演算法原理。往往忽視了整個業務場景建模過程中，看似最普通，卻又最精髓的資料預處理或者叫資料清洗過程。

1. 資料接入

我們經常提到的ETL是將業務系統的資料經過抽取、清洗轉換之後載入到資料倉庫的過程，首先第一步就是根據不同來源的資料進行資料接入，主要接入方式有三：

• 1.批量資料
  可以考慮採用使用備份資料庫匯出dmp，通過ftp等多種方式傳送，首先接入樣本資料，進行分析
• 2.增量資料
  考慮使用ftp，http等服務配合指令碼完成
• 2.實時資料
  訊息佇列接入，kafka，rabbitMQ 等

資料接入對應ETL 中的E----EXTRACT（抽取），接入過程中面臨多種資料來源，不同格式，不同平臺，資料吞吐量，網路頻寬等多種挑戰。

python 這種膠水語言天然可以對應這類多樣性的任務，當然如果不想程式設計，還有：Talend，Kettle，Informatica，Inaplex Inaport等工具可以使用.

e.g. 一個kettle 的作業流

以上不是本文重點，不同資料來源的匯入匯出可以參考：

2. 髒資料的清洗

比如在使用Oracle等資料庫匯出csv file時，欄位間的分隔符為英文逗號，欄位用英文雙引號引起來，我們通常使用大資料工具將這些資料載入成表格的形式，pandas ，spark中都叫做dataframe

對與欄位中含有逗號，回車等情況，pandas 是完全可以handle 的，spark也可以但是2.2之前和gbk解碼共同作用會有bug

資料樣例

1,2,3
"a","b,
c","d"
"4","6,7","8"

pandas

# -*- coding:utf-8 -*-
"""@author:
 
[email protected]:[email protected]:2018/5/3110:49"""
import  pandas

def sum_analysis(filename,col_names):
    # 讀csv檔案
    data = pandas.read_csv(filename,names=col_names,\
    engine='python', dtype=str)
    # 返回前n行
    first_rows = data.head(n=2)
    print(first_rows)
    # 返回全部列名
    cols = data.columns
    print(cols)
    # 返回維度
    dimensision = data.shape
    print(dimensision)
    print(data.info())
    return data

def main():
    col_names = ['1','2','3']
    file_test = u'''test.csv'''
    print(sum_analysis(file_test,col_names))

if __name__=='__main__':
    main()

pyspark

sdf = spark.read.option("header","true") \
				 .option("charset","gbk") \
				 .option("multiLine", "true") \
				  .csv("s3a://your_file*.csv")
pdf = sdf.limit(1000).toPandas()

linux 命令

強大的sed命令，去除兩個雙引號中的換行

**處理結果放入新檔案**
sed ':x;N;s/\nPO/ PO/;b x' INPUTFILE  > OUTPUTFILE

**處理結果覆蓋原始檔**
sed -i ':x;N;s/\nPO/ PO/;b x' INPUTFILE

2.1 檔案轉碼

當然，有些情況還有由於檔案編碼造成的亂碼情況，這時候就輪到linux命令大顯神威了。

比如 使用enconv 將檔案由漢字編碼轉換成utf-8

enconv -L zh_CN -x UTF-8 filename

或者要把當前目錄下的所有檔案都轉成utf-8

enca -L zh_CN -x utf-8 *     

在Linux中專門提供了一種工具convmv進行檔名編碼的轉換，可以將檔名從GBK轉換成UTF-8編碼,或者從UTF-8轉換到GBK。

下面看一下convmv的具體用法：

	convmv -f 源編碼 -t 新編碼 [選項] 檔名

#將目錄下所有檔名由gbk轉換為utf-8
convmv -f GBK -t UTF-8 -r --nosmart --notest /your_directory

2.2 指定列名

在spark 中

如何把別的dataframe已有的schame加到現有的dataframe 上呢？

from pyspark.sql.types import *
diagnosis_sdf_new = diagnosis_sdf.rdd.toDF(diagnosis_sdf_tmp.schema)

3. 缺失值的處理

pandas

pandas使用浮點值NaN(Not a Number)表示浮點數和非浮點陣列中的缺失值，同時python內建None值也會被當作是缺失值。

如果其中有值為None，Series會輸出None，而DataFrame會輸出NaN，但是對空值判斷沒有影響。DataFrame使用isnull方法在輸出空值的時候全為NaN

例如對於樣本資料中的年齡欄位，替換缺失值，並進行離群值清洗

pdf["AGE"] = pd.to_numeric(pdf["AGE"],"coerce").fillna(500.0).astype("int")

pdf[(pdf["AGE"] > 0) & (pdf["AGE"] < 150)]

自定義過濾器過濾

#Fix gender
def fix_gender(x):
    if x is None:
        return None
    if "男" in x:
        return "M"
    if "女" in x:
        return "F"
pdf["PI_SEX"] = pdf["PI_SEX"].map(fix_gender)
or
pdf["PI_SEX"] = pdf["PI_SEX"].apply(fix_gender)

或者直接刪除有缺失值的行

data.dropna()

pyspark
spark 同樣提供了，.dropna(…) ，.fillna(…) 等方法，是丟棄還是使用均值，方差等值進行填充就需要針對具體業務具體分析了

#檢視application_sdf每一列缺失值百分比
import pyspark.sql.functions as fn
queshi_sdf = application_sdf.agg(*[(1-(fn.count(c) /fn.count('*'))).alias(c+'_missing') for c in application_sdf.columns])

queshi_pdf  = queshi_sdf.toPandas()
queshi_pdf

4. 資料質量核查與基本的資料統計

對於多來源場景下的資料，需要敏銳的發現數據的各類特徵，為後續機器學習等業務提供充分的理解，以上這些是離不開資料的統計和質量核查工作，也就是業界常說的讓資料自己說話。

##4.1 統一單位

多來源資料 ，突出存在的一個問題是單位不統一，比如度量衡，國際標準是米，然而很多北美國際習慣使用英尺等單位，這就需要我們使用自定義函式，進行單位的統一換算。

4.2 去重操作

pandas

去重操作可以幫助我們統計業務的核心資料，從而迅速抓住主要矛盾。例如，對於網際網路公司來說，每天有很多的業務資料，然而發現其中的獨立個體的獨立行為才是資料分析人員應該注意的點。

data.drop_duplicates(['column'])

pyspark

使用dataframe api 進行去除操作和pandas 比較類似

sdf.select("column1","column2").dropDuplicates()

當然如果資料量大的話，可以在spark環境中算好再轉化到pandas的dataframe中，利用pandas豐富的統計api 進行進一步的分析。

pdf = sdf.select("column1","column2").dropDuplicates().toPandas()

使用spark sql，其實我覺的這個spark sql 對於傳統的資料庫dba 等分析師來說簡直是革命性產品， 例如：如下程式碼統計1到100測試中每一個測試次數的人員分佈情況

count_sdf.createOrReplaceTempView("testnumber")

count_sdf_testnumber = spark.sql("\
SELECT tests_count,count(1) FROM \
testnumber where tests_count < 100 and lab_tests_count > 0 \
group by tests_count \
order by count(1) desc")

count_sdf_testnumber.show()

4.3 聚合操作與統計

pyspark 和pandas 都提供了類似sql 中的groupby 以及distinct 等操作的api，使用起來也大同小異，下面是對一些樣本資料按照姓名，性別進行聚合操作的程式碼例項

sdf.groupBy("SEX").agg(F.count("NAME")).show()

labtest_count_sdf = sdf.groupBy("NAME","SEX","PI_AGE").agg(F.countDistinct("CODE").alias("tests_count"))

順帶一句，pyspark 跑出的sql 結果集合，使用toPandas() 轉換為pandas 的dataframe 之後只要通過引入matplotlib, 就能完成一個簡單的視覺化demo 了。

d2 = pd.DataFrame({
    'label': [1,2,3],
    'count': [10,2,3],})

d2.plot(kind='bar')
plt.show()
d2.plot.pie(labels=['1', '2', '3'],subplots=True, figsize=(8, 4))
plt.show()

直方圖，餅圖