项目简介

此项目为优达学城数据分析（高级）毕业项目, 该项目需要使用Spark预测Sparkify应用程序的客户流失率.

项目目标

Udacity在其教室中提供Sparkify应用程序部分数据集, 便于学生完成项目，数据集文件大小为128MB, 文件类型为json. 同时项目要求成果通过Git Hub提交并附带项目过程报告.

项目思路

• 1.搭建spark运行环境
• 2.加载库
• 3.加载与清洗数据
• 4.探索性数据分析
• 5.获取特征
• 6.机器学习
• 7.结论汇总

项目实现记录

搭建spark运行环境

在Mac os X系统中搭建spark运行环境时我遇到很多的坑, 耽误很长时间, 为了保证项目进度, 选择在Udacity自带的运行环境中完成项目. 这里我将遇到的坑进行汇总，便于后续解决.

• 1.系统必须安装java JDK 8 （支持最新版）且在环境变量中设置
• 2.在spark官网上下载Apache Spark 2.4.5, 虽然Anaconda中也可以直接安装pyspark库, 但是版本较低, 是2.2.x版本
• 3.python版本不能高于3.6
  进行以上设置后, 使用jupyter notebook在mac os X系统中还是未能正常加载pyspark库, 提示未找到Apache Hadoop😢

之后和班上其他同学聊到此问题, 有部分同学表示, 如果电脑性能太低, 就算搭建好运行平台, 模型运算速度会很慢, 甚至不如在优达自带运行环境中快, 但从技术角度来看, 学会在本地搭建spark运行环境还是有意义的, 后续完善本地搭建spark环境.

加载项目所需运行库

回到项目本身, 根据项目思路, 会用到5个方面的库.

• 1.基本的pandas\numpy, 便于对dataframe进行操作
• 2.matplotlib\seaborn, 生成可视化图形
• 3.pyspark\在python中实例化spark对象
• 4.pyspark.sql, 将数据集转换为spark识别的格式, 使数据在spark中能够进行类似在SQL下的操作
• 5.pyspark.ml, spark中进行机器学习的库
• 6.time, 记录代码块运行时间的库

加载与清洗数据

首先，实例化spark

spark = SparkSession \
    .builder \
    .appName("Sparkify_Spark_Sql_Session") \
    .getOrCreate()

之后可以使用spark.sparkContext.getConf().getAll()方法查询实例化spark的运行参数.

其次，加载数据集

对于陌生的数据集，我的思路是查看其列名、行数，每列的前几行的值（便于了解每列的含义），有无空值，每列值的数据类型。

由于项目要求用Git Hub提交, 免费版Git Hub无法上传大于100MB的文件, 需要将128MB的文件压缩, 根据老师提示, 使用bz2压缩文件格式, spark可以直接读取其中数据. MacOS自带bz2压缩程序, 使用如下命令将文件压缩, 压缩后被压缩文件将删除.

$ bzip2 -z filename.json

加载数据集使用方法spark.read.json('filename.bz2')

数据加载后, 对数据进行观察, 观察数据集时用到的方法如下:

.printSchema()
.describe()
.take()

在.describe()与.take()方法之后加上.show()方法, 可以按照一定格式显示数据,便于阅读.

df.describe('userAgent').show()

+-------+--------------------+
|summary|           userAgent|
+-------+--------------------+
|  count|              278154|
|   mean|                null|
| stddev|                null|
|    min|"Mozilla/5.0 (Mac...|
|    max|Mozilla/5.0 (comp...|
+-------+--------------------+

遇到不能理解其含义的变量时, 可通过用.select方法具体观察

df.select('auth').dropDuplicates().sort('auth').show()

+----------+
|      auth|
+----------+
| Cancelled|
|     Guest|
| Logged In|
|Logged Out|
+----------+

.sort()方法是将查看的数据集按照某列排序

查看所有数据集后, 对数据集有以下认识:

• 1.数据集共有286500行
• 2.共有18列
• 3.连续变量有itemInSession\length\registration\ts, 剩下变量均为分类变量
• 4.每个变量的类型与含义如下:

 |-- artist: string (歌手)
 |-- auth: string (登录状态)
 |-- firstName: string (名字)
 |-- gender: string (性别)
 |-- itemInSession: long (连续变量, 具体含义暂不明确)
 |-- lastName: string (姓氏)
 |-- length: double (听歌时长)
 |-- level: string (会员等级)
 |-- location: string (地区)
 |-- method: string (分类变量, 具体含义暂不明确, 可能与用户发送\接受app信息有关)
 |-- page: string (请求页面)
 |-- registration: long (注册时间)
 |-- sessionId: long (页面ID)
 |-- song: string (歌名)
 |-- status: long (分类变量, 具体含义暂不明确, 可能与连接状态有关)
 |-- ts: long (连续变量, 具体含义暂不明确)
 |-- userAgent: string (用户使用平台信息)
 |-- userId: string (用户ID)

• 5.部分数据在查看其详情时, 发现有空值需要处理

处理空值
userId为用户ID, 在之后的数据清理过程中, 可用作数据集索引, 但发现其有空值, 需要处理
处理空值方法.dropna.

df.dropna(how= 'any', subset = ['userId', 'sessionId'])

使用此方法后, 发现还是存在ID无数值的情况, 怀疑其可能被空值填充, 使用.filter, 去除有空字符的行.

df_dropna.filter(df_dropna['userId'] != '')

之后使用.count方法计算被删除的行数, 与数据集剩余行数.

df.count() - df_dropna.count(), df_dropna.count()

最终, 数据集有8346行被删除, 删除后的数据集还剩278154行.

探索性数据分析

模型标签
项目提示使用churn作为模型的标签, 并且建议使用Cancellation Confirmation事件来定义客户流失. 基于对数据集的理解, Cancellation Confirmation与Downgrade事件为page变量中的两项, Cancellation Confirmation为确认注销, Downgrade为降级.
该问题的解决方案

• 1.新建churn列
• 2.标记page中的Cancellation Confirmation事件, 将转换后的数据改为int型, 再存入新列中
• 3.之后再通过标记找到对应的用户
  结果如下:

+------+-----+
|userId|churn|
+------+-----+
|100010|    0|
|200002|    0|
|   125|    1|
|   124|    0|
|    51|    1|
+------+-----+

探索数据
定义好客户流失后, 可以执行一些探索性数据分析, 观察留存用户和流失用户的行为.
首先把这两类用户的数据聚合到一起, 观察某个特征动作的次数, 比如会员等级\性别等.
保险起见, 将需要查看的特征, 转换为pandas下的dataframe类型, 便于之后的可视化工作.

• 1.删除账户的用户与用户等级的关系

# 提取churn与level特征，整理排序
churn_level_df = df_new.filter('page == "Cancellation Confirmation"') \
    .groupby('level') \
    .count() \
    .toPandas()

# 使用直方图探索churn与level的关系
churn_level_df.plot.bar();

付费用户注销数量高于免费用户注销数量

• 2.删除账户的用户与性别的关系

# 提取churn与gender特征，整理排序
churn_gender_df = df_new.dropDuplicates(['userId', 'gender']) \
    .groupby(['churn', 'gender']) \
    .count() \
    .sort('churn') \
    .toPandas()

# 通过直方图探索churn与gender的关系
ax = sns.barplot(x = 'churn', y = 'count', hue = 'gender', data = churn_gender_df)
plt.xlabel("Has user delete the account")
plt.ylabel("Count")
plt.title("Gender ratio of users who delete the account");

男性用户比女性用户删除账户的人数更多
删除账户的比例对一款应用来说比较高
app可能对男性吸引力更大

特征工程

熟悉数据之后，我认为以下特征可能对训练模型产生较大影响

• 1.用户听过的歌手数量
• 2.性别
• 3.用户听歌时长
• 4.用户所听歌曲总和
• 5.歌曲被加入播放列表的数量
• 6.会员等级

下面详细说说，获取每种特征值的关键点

• 1.用户听过的歌手数量
  page页面记录了用户在使用app过程中的动作，获取每个用户在点击页面NextSong时的artist信息并计数，就能获得用户听过的歌手数量

feature_artists = df_new.filter(df_new.page == 'NextSong') \ # 获取页面
    .select('userId', 'artist') \ # 获取userId、artist
    .dropDuplicates() \ # 去除重复值
    .groupBy('userId') \ # 按userId分组
    .count() \ # 记录同一用户不同artist出现的数量
    .withColumnRenamed('count', 'sum_artist') # 重命名列

• 2.性别
  gender特征的问题在于要把F、M变量转为0、1，方便模型计算

eature_gender = df_new.select('userId', 'gender') \ # 获取userId、gender
    .dropDuplicates() \ # 去除重复值
    .replace(['M', 'F'], ['0', '1'], 'gender') \ # 将`F、M`变量转为`0、1`
    .select('userId', col('gender').cast('int')) # 将值转换为int类型

• 3.用户听歌时长
  length特征关键在于需要将每个用户所有length值相加

feature_length = df_new.select('userId', 'length') \# 获取userId、length
    .groupBy('userId') \# 按userId分组
    .sum() \ # 按userId相加
    .withColumnRenamed('sum(length)', 'listening_time')# 重命名列

• 4.用户所听歌曲总和
  该特征没什么难点，之前特征的部分方法就能得到

feature_songs = df_new.select('userId', 'song') \# 获取userId、song
    .groupBy('userId') \# 按userId分组
    .count() \# 计数
    .withColumnRenamed('count', 'sum_song')# 重命名列

• 5.歌曲被加入播放列表的数量
  该特征获取原理是记录Add to Playlist的次数

feature_ATP = df_new.select('userId', 'page') \
    .where(df_new.page == 'Add to Playlist') \# 筛选出page等于Add to Playlist的页面
    .groupBy('userId') \
    .count() \
    .withColumnRenamed('count', 'add_to_play')

• 6.会员等级
  该特征的获取与性别特征获取方法一致

feature_level = df_new.select('userId', 'level') \
    .dropDuplicates() \
    .replace(['free', 'paid'], ['0', '1'], 'level') \
    .select('userId', col('level').cast('int'))

整理标签数据，之后与特征数据汇总

label_churn = df_new.select('userId', col('churn') \
    .alias('label')) \ # 对特征churn取别名为label
    .dropDuplicates()

整合特征值与标签

# 这里注意.join函数的用法
# 如果为空的数据，需要用0填充，不然最后模型计算会报错
# 一定不要忘记删除userId，userId是索引，不是特征，不能导入到模型计算
df_feature = feature_artists.join(feature_gender, 'userId', 'outer') \
    .join(feature_length, 'userId', 'outer') \
    .join(feature_songs, 'userId', 'outer') \
    .join(feature_ATP, 'userId', 'outer') \
    .join(label_churn, 'userId', 'outer') \
    .fillna(0) \
    .drop('userId')

特征工程的主要目的是提取特征，并生成线性代数矩阵，其中对标签设置别名很关键，因为建模时使用的方法需要关键字label对应的变量，没有设置别名，或者别名设置成其他值，均不能运算

建模

将完整数据集分成训练集、测试集和验证集。选用逻辑回归、支持向量机与随机森铃机器学习方法。项目说明建议评价指标选择准确率，选用 F1 score 作为优化指标。

关于参数的选择：
机器学习一般选择4个参数作为衡量模型好坏的标准，分别为准确率（Precision）、精确度（Accuracy）、召回率（Recall）、F1分数（F1-Score），简单阐述这几种参数的含义

1）准确率是对给定数据集，分类正确样本个数和总样本数的比值；
2）精确度说明判断为真的正例占所有判断为真的样例比重；
3）召回率又被称为查全率，用来说明分类器中判定为真的正例占总正例的比率；
4）精确度和召回率之间是负相关的关系，引入F1-Score作为综合指标，平衡准确率和召回率的影响。

根据项目需求，需要预测客户流失率，但流失顾客数据集很小，只占数据1%不到，Accuracy很难反映模型好坏，f1分数这时候就比较关键。

关于机器学习算法的选择：
1）逻辑回归-----优点：计算速度快，容易理解 缺点：容易产生欠拟合
2）支持向量机---优点：数据量较小情况下解决机器学习问题，可以解决非线性问题 缺点：对缺失数据敏感
3）随机森林-----优点：在当前的算法中拥有特别好的精确度（Accuracy），可以有效的运行在大数据集上，有缺失数据也能够获得更好的结果

在模型的选择上，我的思路是选用逻辑回归作为模型参考，因为其容易欠拟合，其他两种机器学习算法的分数应该比逻辑回归更高；
支持向量机在现在的小数据集上应该表现最佳，但是数据存在小部分确实，可能对分数产生影响；
随机森林的分数大概率与支持向量机类似，但是其更适合运用于大数据，在之后测试12GB大数据时从计算时间上会优于SVM。

• 转换数据

# 将特征工程中的数据集转换为可供模型计算的结构
columns = ['sum_artist', 'gender', 'listening_time', 'sum_song', 'add_to_play']
assembler = VectorAssembler(inputCols = columns, outputCol = 'features_matrix')
data = assembler.transform(df_feature)

# 标准化数据
scaler = StandardScaler(inputCol = 'features_matrix', outputCol = 'features')
scalerModel = scaler.fit(data)
data = scalerModel.transform(data)

# 将数据集分成训练集、测试集和验证集
train, test, validation = data.randomSplit([0.6, 0.2, 0.2])

这里卡了一段时间，因为转换以后的数据竟然和没转换之前一样，

• 逻辑回归

# 初始化逻辑回归，maxIter为可迭代最大次数，逻辑回归中必须设置maxIter
lr = LogisticRegression(maxIter=5)

# 设置评估标准
f1_score = MulticlassClassificationEvaluator(metricName = 'f1')

# 建立参数网格
paramGrid = ParamGridBuilder().build()
# 设置交叉验证
lr_crossval = CrossValidator(estimator=lr,
                             estimatorParamMaps=paramGrid,
                             evaluator=f1_score)
                             
# 训练之后，通过验证集计算准确度和f1分数
lr_result = crossval_model.transform(validation)

# 查看结果
# 时间也是衡量模型好坏的一个标准
evaluator = MulticlassClassificationEvaluator(predictionCol = 'prediction')
print("逻辑回归分数:")
start = time()
print("准确度: {}".format(evaluator.evaluate(lr_result, {evaluator.metricName:'accuracy'})))
print("f1分数: {}".format(evaluator.evaluate(lr_result, {evaluator.metricName:'f1'})))
end = time()
print("验证集计算准确度与f1分数用时 {}秒".format(end - start))

• 支持向量机、随机森林
  支持向量机、随机森林两个模型代码与逻辑回归结构基本一致，除了需要将代码改为对应模型外，随机森林可以不设置最大迭代次数

• 计算结果
  • 逻辑回归模型的准确度为 0.7755，f1分数为 0.6775，耗时 645秒
  • SVM（支持向量机）模型的准确度为 0.7755，f1分数为 0.6775，耗时796秒
  • 随机森林模型的准确度为 0.7143，f1分数为 0.6463，耗时 871秒

参数优化

由于使用模型的默认参数运算，得到的结果并不理想，调整参数，优化结果。

下面将对以上三个模型增加交叉验证，每种模型均交叉验证3次：

# 初始化逻辑回归
lr = LogisticRegression(maxIter=5)

# 设置评估标准
f1_score = MulticlassClassificationEvaluator(metricName = 'f1')

# 建立paramGrid
paramGrid = ParamGridBuilder().build()

lr_crossval = CrossValidator(estimator=lr,
                             estimatorParamMaps=paramGrid,
                             evaluator=f1_score,
                             numFolds=3)

代码和之前唯一的变化就是在CrossValidator中增加了numFolds=3，其余代码不变，SVM与随机森林增加交叉验证的方式与逻辑回归相同。

• 交叉验证后的计算结果
  • 逻辑回归模型的准确度为 0.7755，f1分数为 0.6775，耗时 667秒
  • SVM（支持向量机）模型的准确度为 0.7755，f1分数为 0.6775，耗时655秒
  • 随机森林模型的准确度为 0.7143，f1分数为 0.6463，耗时 682秒

优化逻辑回归
由于增加交叉验证的逻辑回归算法与SVM无明显差距，加上其易于理解，我将进一步优化该算法

# 主要修改该部分代码
paramGrid = ParamGridBuilder().addGrid(lr.regParam, [0.1, 0.01]) \
                              .addGrid(lr.fitIntercept, [False, True]) \
                              .build()

• 优化逻辑回归后的计算结果
  • 在调整逻辑回归的正则项系数及是否需要计算截距这两个参数后，发现4个结果完全一样，均为0.67487；
  • 和之前的计算结果相比，f1分数有略微下降；
  • 使用未调整参数的逻辑回归模型分数更高。

测试集结果

# 使用f1分数最高，时间最快的模型计算测试集
lr_best = LogisticRegression(maxIter=5)
lr_best_model = lr_best.fit(train)
final_result = lr_best_model.transform(test)

# 显示最终结果
evaluator = MulticlassClassificationEvaluator(predictionCol = 'prediction')
print("测试集结果:")
print("准确度: {}".format(evaluator.evaluate(final_result, {evaluator.metricName:'accuracy'})))
print("f1分数: {}".format(evaluator.evaluate(final_result, {evaluator.metricName:'f1'})))

在测试集上运算后，accuracy为0.7442，f1分数为0.6350，和最初在验证集的结果上相比，accuracy与f1分数均降低了3%左右。

总结

虽然最终采用逻辑回归作为最终模型是因为其accuracy与f1分数最高，但我认为分数还有进一步提高的可能，有这种想法主要因为以下几点：

• 第一次运算时SVM与逻辑回归分数一样，一般上讲逻辑回归容易欠拟合，但和SVM模型结果一样，说明逻辑回归欠拟合的可能性较小；
• 三种模型的f1分数并不高，说明pricision和recall始终不平衡；
• 最佳逻辑回归模型代入测试集运算后，accuracy与f1均下降了3%左右，并没有与之前差很多。

综合以上观点，我认为在特征工程部分出现了偏差导致这种现象，我选择的特征与用户流失率的关系不是特别相关。在后续完善工作中，我认为以下方案可以提高accuracy与f1分数：

• 增加参考标准，计算没有用户流失和全部用户流失的accuracy与f1分数，评估其他模型好坏；
• 回顾数据集，对未理解的特征进行探究，了解其含义，找到与数据集相关性更强的特征；
• 增加决策树、梯度提升树等其他算法，对比已使用的算法，观察accuracy与f1分数变化。

在完成项目的过程中遇到以下难点，思考其解决方案时间较久，不过最终都解决了：

• 每次运行模型，分数总会变化。造成该问题的主要原因是每次被分割的数据集均在变化，每次关掉IDE，变量就会被清空，重新加载运行，必然重新分割数据集；
• 开始将会员是否付费作为特征，代入模型计算，但分数较现在更低，思考后，感觉该特征和用户流失率并无因果关系，删去此特征后，accuracy与f1分数均有提高。

参考文献

• 【机器学习】准确率，精确度，召回率的概念和表示详解
• 【机器学习】 随机森林（Random Forest）
• 机器学习算法优缺点及其应用领域
• F1-score是越大越好吗?
• Spark ML Tuning：模型选择和超参调优
• Spark的安装（基于Mac）
• Spark Transformation —— randomSplit
• Mac terminal下解压缩命令大全
• Spark机器学习之模型选择和超参数调整
• Python Spark MLlib之逻辑回归