有了前面的基础，我们就可以利用PySpark进行数据的分析与挖掘，而通常我们需要花费80%——90%的时间用于数据的预处理，这是我们后续建模开发的基础，本文将基于PySpark对大规模数据预处理从实际业务场景出发进行分析，从而更接地气！！！
1 重复数据
spark 2.0+
df = spark.createDataFrame([
(1, 144.5, 5.9, 33, 'M'),
(2, 167.2, 5.4, 45, 'M'),
(3, 124.1, 5.2, 23, 'F'),
(4, 144.5, 5.9, 33, 'M'),
(5, 133.2, 5.7, 54, 'F'),
(3, 124.1, 5.2, 23, 'F'),
(5, 129.2, 5.3, 42, 'M')],
['id',
'weight',
'height',
'age',
'gender']
)
spark 1.6.2
df = sqlContext.createDataFrame([
(1, 144.5, 5.9, 33, 'M'),
(2, 167.2, 5.4, 45, 'M'),
(3, 124.1, 5.2, 23, 'F'),
(4, 144.5, 5.9, 33, 'M'),
(5, 133.2, 5.7, 54, 'F'),
(3, 124.1, 5.2, 23, 'F'),
(5, 129.2, 5.3, 42, 'M')],
['id',
'weight',
'height',
'age',
'gender']
)
1.1 检查重复数据
用 .distinct() 完全相同的两行或多行去重
print('Count of rows: {0}'.format(df.count()))
print('Count of distinct rows: {0}'.format(df.distinct().count()))
Count of rows: 7
Count of distinct rows: 6
用 .dropDuplicates(...) 方法去除重复行,具体用法可以参考本文集 Pyspark.SQL模块介绍
df = df.dropDuplicates()
df.show()
+---+------+------+---+------+
| id|weight|height|age|gender|
+---+------+------+---+------+
| 5| 129.2| 5.3| 42| M|
| 4| 144.5| 5.9| 33| M|
| 3| 124.1| 5.2| 23| F|
| 2| 167.2| 5.4| 45| M|
| 1| 144.5| 5.9| 33| M|
| 5| 133.2| 5.7| 54| F|
+---+------+------+---+------+
除了ID字段外的消除重复行
print('Count of ids: {0}'.format(df.count()))
#print('Count of distinct ids: {0}'.format(df.select([c for c in df.columns if c != 'id']).distinct().count()))
print('Count of distinct ids:{0}'.format(df.select(["weight","height","age","gender"]).distinct().count()))
Count of ids: 6
Count of distinct ids:5
使用 .dropDuplicates(...) 处理重复行，但是添加 subset 参数
# df = df.dropDuplicates(subset=[c for c in df.columns if c != 'id'])
df =df.dropDuplicates(["weight","height","age","gender"])
df.show()
+---+------+------+---+------+
| id|weight|height|age|gender|
+---+------+------+---+------+
| 5| 133.2| 5.7| 54| F|
| 4| 144.5| 5.9| 33| M|
| 2| 167.2| 5.4| 45| M|
| 5| 129.2| 5.3| 42| M|
| 3| 124.1| 5.2| 23| F|
+---+------+------+---+------+
利用.agg(...)函数计算ID的总数和ID唯一个数
import pyspark.sql.functions as fn
df.agg(
fn.count('id').alias('id_count'),
fn.countDistinct('id').alias('id_distinct')
).show()
+--------+-----------+
|id_count|id_distinct|
+--------+-----------+
| 5| 4|
+--------+-----------+
给每一行一个唯一的ID号，相当于新增一列数字
df.withColumn('new_id', fn.monotonically_increasing_id()).show()
+---+------+------+---+------+------------+
| id|weight|height|age|gender| new_id|
+---+------+------+---+------+------------+
| 5| 133.2| 5.7| 54| F|283467841536|
| 4| 144.5| 5.9| 33| M|352187318272|
| 2| 167.2| 5.4| 45| M|386547056640|
| 5| 129.2| 5.3| 42| M|438086664192|
| 3| 124.1| 5.2| 23| F|463856467968|
+---+------+------+---+------+------------+
1.2 缺失值观察
df_miss = sqlContext.createDataFrame([
(1, 143.5, 5.6, 28, 'M', 100000),
(2, 167.2, 5.4, 45, 'M', None),
(3, None , 5.2, None, None, None),
(4, 144.5, 5.9, 33, 'M', None),
(5, 133.2, 5.7, 54, 'F', None),
(6, 124.1, 5.2, None, 'F', None),
(7, 129.2, 5.3, 42, 'M', 76000),
], ['id', 'weight', 'height', 'age', 'gender', 'income'])
df_miss.show()
+---+------+------+----+------+------+
| id|weight|height| age|gender|income|
+---+------+------+----+------+------+
| 1| 143.5| 5.6| 28| M|100000|
| 2| 167.2| 5.4| 45| M| null|
| 3| null| 5.2|null| null| null|
| 4| 144.5| 5.9| 33| M| null|
| 5| 133.2| 5.7| 54| F| null|
| 6| 124.1| 5.2|null| F| null|
| 7| 129.2| 5.3| 42| M| 76000|
+---+------+------+----+------+------+
统计每行缺失值的个数
df_miss.rdd.\
map(lambda row:(sum([c == None for c in row]),row['id'])).\
sortByKey(ascending=False).collect() #(缺失值个数，行号)
[(4, 3), (2, 6), (1, 2), (1, 4), (1, 5), (0, 1), (0, 7)]
df_miss.where('id = 3').show()
+---+------+------+----+------+------+
| id|weight|height| age|gender|income|
+---+------+------+----+------+------+
| 3| null| 5.2|null| null| null|
+---+------+------+----+------+------+
每一列缺失的观测数据的百分比
# fn.count(c) / fn.count('*') 为每列字段非缺失的记录数占比
# fn.count('*') *表示列名的位置，指示该列方法计算所有的列
# .agg(* *之前的列指示，.agg(...)方法将该列表处理为一组独立的参数传递给函数
df_miss.agg(*[
(1.00 - (fn.count(c) / fn.count('*'))).alias(c + '_missing') for c in df_miss.columns
]).show()
+----------+------------------+--------------+------------------+------------------+------------------+
|id_missing| weight_missing|height_missing| age_missing| gender_missing| income_missing|
+----------+------------------+--------------+------------------+------------------+------------------+
| 0.0|0.1428571428571429| 0.0|0.2857142857142857|0.1428571428571429|0.7142857142857143|
+----------+------------------+--------------+------------------+------------------+------------------+
移除income列，因为income列大部分是缺失值，缺失值占比达到71%
df_miss_no_income = df_miss.select([c for c in df_miss.columns if c != 'income'])
df_miss_no_income.show()
+---+------+------+----+------+
| id|weight|height| age|gender|
+---+------+------+----+------+
| 1| 143.5| 5.6| 28| M|
| 2| 167.2| 5.4| 45| M|
| 3| null| 5.2|null| null|
| 4| 144.5| 5.9| 33| M|
| 5| 133.2| 5.7| 54| F|
| 6| 124.1| 5.2|null| F|
| 7| 129.2| 5.3| 42| M|
+---+------+------+----+------+
使用.dropna(...) 移除方法，thresh(设定移除数的阈值)
df_miss_no_income.dropna(thresh=3).show()
+---+------+------+----+------+
| id|weight|height| age|gender|
+---+------+------+----+------+
| 1| 143.5| 5.6| 28| M|
| 2| 167.2| 5.4| 45| M|
| 4| 144.5| 5.9| 33| M|
| 5| 133.2| 5.7| 54| F|
| 6| 124.1| 5.2|null| F|
| 7| 129.2| 5.3| 42| M|
+---+------+------+----+------+
使用.fillna(...) 方法，填充观测数据.
means = df_miss_no_income.agg(*[
fn.mean(c).alias(c) for c in df_miss_no_income.columns if c != 'gender']).toPandas().to_dict('records')[0]
# gender单独处理
means['gender'] = 'missing'
df_miss_no_income.fillna(means).show()
+---+-------------+------+---+-------+
| id| weight|height|age| gender|
+---+-------------+------+---+-------+
| 1| 143.5| 5.6| 28| M|
| 2| 167.2| 5.4| 45| M|
| 3|140.283333333| 5.2| 40|missing|
| 4| 144.5| 5.9| 33| M|
| 5| 133.2| 5.7| 54| F|
| 6| 124.1| 5.2| 40| F|
| 7| 129.2| 5.3| 42| M|
+---+-------------+------+---+-------+
1.3 离群值Outliers（异常数据）
df_outliers = sqlContext.createDataFrame([
(1, 143.5, 5.3, 28),
(2, 154.2, 5.5, 45),
(3, 342.3, 5.1, 99),
(4, 144.5, 5.5, 33),
(5, 133.2, 5.4, 54),
(6, 124.1, 5.1, 21),
(7, 129.2, 5.3, 42),
], ['id', 'weight', 'height', 'age'])
df_outliers.show()
+---+------+------+---+
| id|weight|height|age|
+---+------+------+---+
| 1| 143.5| 5.3| 28|
| 2| 154.2| 5.5| 45|
| 3| 342.3| 5.1| 99|
| 4| 144.5| 5.5| 33|
| 5| 133.2| 5.4| 54|
| 6| 124.1| 5.1| 21|
| 7| 129.2| 5.3| 42|
+---+------+------+---+
使用之前列出的定义来标记离群值
# approxQuantile(col, probabilities, relativeError) New in version 2.0.
cols = ['weight', 'height', 'age']
bounds = {}
for col in cols:
quantiles = df_outliers.approxQuantile(col, [0.25, 0.75], 0.05)
IQR = quantiles[1] - quantiles[0]
bounds[col] = [
quantiles[0] - 1.5 * IQR,
quantiles[1] + 1.5 * IQR
]
bounds字典保存了每个特征的上下界限
bounds
{'age': [9.0, 51.0],
'height': [4.8999999995, 5.6],
'weight': [115.0, 146.8499999997]}
标记样本数据的离群值
# 按位或运算符：只要对应的两个二进位有一个为1时，结果位就为1。
outliers = df_outliers.select(*['id'] + [((df_outliers[c] < bounds[c][0]) | (df_outliers[c] > bounds[c][1])).alias(c + '_o') \
for c in cols \
]\
)
outliers.show()
+---+--------+--------+-----+
| id|weight_o|height_o|age_o|
+---+--------+--------+-----+
| 1| false| false|false|
| 2| true| false|false|
| 3| true| false| true|
| 4| false| false|false|
| 5| false| false| true|
| 6| false| false|false|
| 7| false| false|false|
+---+--------+--------+-----+
列出weight和age与其他分部明显不同的部分(离群值)
df_outliers = df_outliers.join(outliers, on='id')
df_outliers.filter('weight_o').select('id', 'weight').show()
df_outliers.filter('age_o').select('id', 'age').show()
+---+------+
| id|weight|
+---+------+
| 3| 342.3|
| 2| 154.2|
+---+------+
+---+---+
| id|age|
+---+---+
| 5| 54|
| 3| 99|
+---+---+
通过上述方法即可获得快速清洗数据（重复值、缺失值和离群值）
2 熟悉理解你的数据
2.1描述性统计——EDA
加载数据并且转换为Spark DatFrame
# pyspark.sql.types显示了所有的我们可以使用的数据类型，IntegerType()和FloatType()
from pyspark.sql import types
fraud = sc.textFile('file:///root/ydzhao/PySpark/Chapter04/ccFraud.csv.gz') # 信用卡欺诈数据集
header = fraud.first()
header
'"custID","gender","state","cardholder","balance","numTrans","numIntlTrans","creditLine","fraudRisk"'
fraud.take(3)
['"custID","gender","state","cardholder","balance","numTrans","numIntlTrans","creditLine","fraudRisk"',
'1,1,35,1,3000,4,14,2,0',
'2,2,2,1,0,9,0,18,0']
fraud.count()
10000001
# (1) 去除标题行数据,每个元素转换成整型Integer,还是RDD
fraud = fraud.filter(lambda row: row != header).map(lambda row: [int(x) for x in row.split(',')])
fraud.take(3)
[[1, 1, 35, 1, 3000, 4, 14, 2, 0],
[2, 2, 2, 1, 0, 9, 0, 18, 0],
[3, 2, 2, 1, 0, 27, 9, 16, 0]]
# (2) 创建DataFrame模式
# h[1:-1]代表第一行到最后一行
schema = [
*[
types.StructField(h[1:-1], types.IntegerType(), True) for h in header.split(',')
]
]
schema = types.StructType(schema)
# (3) 创建DataFrame
# spark2.0+
# fraud_df = spark.createDataFrame(fraud, schema)
# spark1.6.2
fraud_df = sqlContext.createDataFrame(fraud, schema)
fraud_df.printSchema()
root
|-- custID: integer (nullable = true)
|-- gender: integer (nullable = true)
|-- state: integer (nullable = true)
|-- cardholder: integer (nullable = true)
|-- balance: integer (nullable = true)
|-- numTrans: integer (nullable = true)
|-- numIntlTrans: integer (nullable = true)
|-- creditLine: integer (nullable = true)
|-- fraudRisk: integer (nullable = true)
fraud_df.show()
+------+------+-----+----------+-------+--------+------------+----------+---------+
|custID|gender|state|cardholder|balance|numTrans|numIntlTrans|creditLine|fraudRisk|
+------+------+-----+----------+-------+--------+------------+----------+---------+
| 1| 1| 35| 1| 3000| 4| 14| 2| 0|
| 2| 2| 2| 1| 0| 9| 0| 18| 0|
| 3| 2| 2| 1| 0| 27| 9| 16| 0|
| 4| 1| 15| 1| 0| 12| 0| 5| 0|
| 5| 1| 46| 1| 0| 11| 16| 7| 0|
| 6| 2| 44| 2| 5546| 21| 0| 13| 0|
| 7| 1| 3| 1| 2000| 41| 0| 1| 0|
| 8| 1| 10| 1| 6016| 20| 3| 6| 0|
| 9| 2| 32| 1| 2428| 4| 10| 22| 0|
| 10| 1| 23| 1| 0| 18| 56| 5| 0|
| 11| 1| 46| 1| 4601| 54| 0| 4| 0|
| 12| 1| 10| 1| 3000| 20| 0| 2| 0|
| 13| 1| 6| 1| 0| 45| 2| 4| 0|
| 14| 2| 38| 1| 9000| 41| 3| 8| 0|
| 15| 1| 27| 1| 5227| 60| 0| 17| 0|
| 16| 1| 44| 1| 0| 22| 0| 5| 0|
| 17| 2| 18| 1| 13970| 20| 0| 13| 0|
| 18| 1| 35| 1| 3113| 13| 6| 8| 0|
| 19| 1| 5| 1| 9000| 20| 2| 8| 0|
| 20| 2| 31| 1| 1860| 21| 10| 8| 0|
+------+------+-----+----------+-------+--------+------------+----------+---------+
only showing top 20 rows
使用.groupby(...) 方法计算性别列的使用频率 我们所面对的样本一个性别比例失衡的样本.
fraud_df.groupby('gender').count().show()
+------+-------+
|gender| count|
+------+-------+
| 1|6178231|
| 2|3821769|
+------+-------+
使用describe()方法进行数值统计
desc = fraud_df.describe(['balance', 'numTrans', 'numIntlTrans'])
desc.show()
+-------+-----------------+------------------+-----------------+
|summary| balance| numTrans| numIntlTrans|
+-------+-----------------+------------------+-----------------+
| count| 10000000| 10000000| 10000000|
| mean| 4109.9199193| 28.9351871| 4.0471899|
| stddev|3996.847309737077|26.553781024522852|8.602970115863767|
| min| 0| 0| 0|
| max| 41485| 100| 60|
+-------+-----------------+------------------+-----------------+
检查balance偏度（skewness）
fraud_df.agg({'balance': 'skewness'}).show()
+------------------+
| skewness(balance)|
+------------------+
|1.1818315552996432|
+------------------+
2.2 相关性
目前 .corr()方法仅仅支持Pearson相关性系数，且两两相关性。
当然你可以把Spark DataFrame 转换为Python的DataFrame之后，你就可以随便怎么弄了，但是开销可能会很大，由数据量决定。期待后续版本的更新。
fraud_df.corr('balance', 'numTrans')
创建相关系数矩阵
numerical = ['balance', 'numTrans', 'numIntlTrans']
n_numerical = len(numerical)
corr = []
for i in range(0, n_numerical):
temp = [None] * i
for j in range(i, n_numerical):
temp.append(fraud_df.corr(numerical[i], numerical[j]))
corr.append(temp)
corr
[[1.0, 0.00044523140172659576, 0.00027139913398184604],
[None, 1.0, -0.0002805712819816179],
[None, None, 1.0]]
2.3 数据可视化
加载matplotlib 和 bokeh包，只会调用python的解释器
%matplotlib inline
import matplotlib.pyplot as plt
plt.style.use('ggplot')
import bokeh.charts as chrt
from bokeh.io import output_notebook
output_notebook()
Loading BokehJS ...
2.3.1 直方图Histograms
方法一 聚集工作节点中的数据并返回一个汇总bins列表和直方图每个bin中的计数给驱动（适用大数据集）
(1) 先对数据进行聚合
hists = fraud_df.select('balance').rdd.flatMap(lambda row: row).histogram(20)
type(hists)
tuple
(2)
使用matplotlib绘制直方图
data = {
'bins': hists[0][:-1],
'freq': hists[1]
}
fig = plt.figure(figsize=(12,9))
ax = fig.add_subplot(1, 1, 1)
ax.bar(data['bins'], data['freq'], width=2000)
ax.set_title('Histogram of balance')
plt.savefig('balance.png', dpi=300)
image.png
使用Bokeh绘制直方图
b_hist = chrt.Bar(data, values='freq', label='bins', title='Histogram of balance')
chrt.show(b_hist)
方法二 所有数据给驱动程序（适用于小数据集） 当然方法一是性能更好的
data_driver = {'obs': fraud_df.select('balance').rdd.flatMap(lambda row: row).collect()}
fig = plt.figure(figsize=(12,9))
ax = fig.add_subplot(1, 1, 1)
ax.hist(data_driver['obs'], bins=20)
ax.set_title('Histogram of balance using .hist()')
plt.savefig('balance_hist.png', dpi=300)
image.png
使用Bokeh绘制直方图
b_hist_driver = chrt.Histogram(data_driver, values='obs', title='Histogram of balance using .Histogram()', bins=20)
chrt.show(b_hist_driver)
2.3.2 特征值之间的交互
10000000条数据先抽样0.02%
data_sample = fraud_df.sampleBy('gender', {1: 0.0002, 2: 0.0002}).select(numerical)
data_sample.show()
+-------+--------+------------+
|balance|numTrans|numIntlTrans|
+-------+--------+------------+
| 0| 15| 0|
| 4000| 3| 0|
| 6000| 21| 50|
| 0| 28| 10|
| 8630| 100| 0|
| 4000| 17| 0|
| 5948| 76| 0|
| 3000| 9| 4|
| 0| 5| 0|
| 1588| 55| 0|
| 3882| 87| 0|
| 1756| 12| 0|
| 4000| 4| 0|
| 0| 10| 0|
| 5000| 17| 0|
| 4188| 50| 0|
| 3141| 2| 1|
| 8000| 52| 5|
| 9000| 40| 0|
| 2423| 11| 1|
+-------+--------+------------+
only showing top 20 rows
data_multi = dict([
(x, data_sample.select(x).rdd.flatMap(lambda row: row).collect())
for x in numerical
])
sctr = chrt.Scatter(data_multi, x='balance', y='numTrans')
chrt.show(sctr)
网友评论