预处理test.csv和train.csv数据集
一、这是本次实验需要用到的数据集，我在之前发过的文章《Hadoop、Hive实战：淘宝双十一数据分析》已经给出了，有同学需要的话可以去看看。
这两个数据集的字段，定义如下：
1、user_id：买家id
2、age_range：买家年龄分段：1表示年龄<18,2表示年龄在[18,24]，3表示年龄在[25,29]，4表示年龄在[30,34]，5表示年龄在[35,39]，6表示年龄在[40,49]，7和8表示年龄>=50,0和NULL则表示未知
3、gender：性别：0表示女性，1表示男性，2和NULL表示未知
4、merchant_id：商家id
5、label：是否是回头客，0值表示不是回头客，1值表示回头客，-1表示该用户已经超出我们所需要考虑的预测范围。NULL值只存在测试集，在测试集中表示需要预测的值。
二、需要预先处理test.csv数据集，把这test.csv数据集里label字段表示-1值剔除掉,保留需要预测的数据.并假设需要预测的数据中label字段均为1.
同样我们需要写一个脚本完成这个任务，将这个脚本与数据集放在一起
将以下内容写进脚本
#!/bin/bash
#下面设置输入文件，把用户执行predeal_test.sh命令时提供的第一个参数作为输入文件名称
infile=$1 #下面设置输出文件，把用户执行predeal_test.sh命令时提供的第二个参数作为输出文件名称
outfile=$2 #注意！！最后的$infile > $outfile必须跟在}’这两个字符的后面
awk -F "," 'BEGIN{
id=0; }
{
if($1 && $2 && $3 && $4 && !$5){
id=id+1;
print $1","$2","$3","$4","1
if(id==10000){
exit }
}
}' $infile > $outfile
执行：
测试一下，取出前5条数据查看：
接下来我们对train.csv数据集进行预处理，我们首先删除第一行数据
然后剔除掉train.csv中字段值为NULL的数据，还是使用脚本完成
#!/bin/bash
#下面设置输入文件，把用户执行predeal_train.sh命令时提供的第一个参数作为输入文件名称
infile=$1
#下面设置输出文件，把用户执行predeal_train.sh命令时提供的第二个参数作为输出文件名称
outfile=$2
#注意！！最后的$infile > $outfile必须跟在}’这两个字符的后面
awk -F "," 'BEGIN{
id=0;
}
{
if($1 && $2 && $3 && $4 && ($5!=-1)){
id=id+1;
print $1","$2","$3","$4","$5
if(id==10000){
exit
}
}
}' $infile > $outfile
执行：
这是我们本次实验需要用到的数据文件：
预测回头客
1、启动hadoop
在启动spark之前先启动hadoop，并将两个数据集分别存取到HDFS中
进入hadoop的安装目录的sbin文件下，输入以下命令开启HDFS
./start-dfs.sh或者./start-all.sh
然后存取到HDFS中：
bin/hadoop fs -mkdir -p /dbtaobao/dataset
bin/hadoop fs -put /usr/local/dbtaobao/dataset/train_after.csv /dbtaobao/dataset
bin/hadoop fs -put /usr/local/dbtaobao/dataset/test_after.csv /dbtaobao/dataset
2、启动mysql服务
登录到mysql中创建一个数据库，用来存储回头客，我们在之前的实验中建过数据库dbtaobao，我们在这个数据库里再建一张表
use dbtaobao;
create table rebuy (score varchar(40),label varchar(40));
3、启动Spark
在启动是Spark之前，需要把连接数据库mysql的jar包上传到spark的安装目录jars文件夹下，
接下来我们启动Spark，在sbin目录下输入以下命令
./start-all.sh
然后在命令行里查看是否开启了spark
接下来我们开启spark-shell
进入spark安装目录：cd /usr/local/spark
输入以下命令：
./bin/spark-shell --jars /root/spark/spark2.4.1/jars/mysql-connector-java-5.1.47-bin.jar --driver-class-path /root/spark/spark2.4.1/jars/mysql-connector-java-5.1.47-bin.jar
4、支持向量机SVM分类器预测回头客
a、在spark-shell中导入需要的包，分别输入以下语句：
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg.{Vectors,Vector}
import org.apache.spark.mllib.classification.{SVMModel, SVMWithSGD}
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
import java.util.Properties
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
b、读取训练文本
val train_data = sc.textFile("/dbtaobao/dataset/train_after.csv")
val test_data = sc.textFile("/dbtaobao/dataset/test_after.csv")
c、构建模型
通过map将每行的数据用“,”隔开，在数据集中，每行被分成了5部分，前4部分是用户交易的3个特征(age_range,gender,merchant_id)，最后一部分是用户交易的分类(label)。把这里我们用LabeledPoint来存储标签列和特征列。LabeledPoint在监督学习中常用来存储标签和特征，其中要求标签的类型是double，特征的类型是Vector。
val train= train_data.map{line =>
val parts = line.split(',')
LabeledPoint(parts(4).toDouble,Vectors.dense(parts(1).toDouble,parts(2).toDouble,parts(3).toDouble))
}
val test = test_data.map{line =>
val parts = line.split(',')
LabeledPoint(parts(4).toDouble,Vectors.dense(parts(1).toDouble,parts(2).toDouble,parts(3).toDouble))
}
接下来设置迭代次数为1000，
val numIterations = 1000
val model = SVMWithSGD.train(train, numIterations)
4、评估模型
输出原始的预测评分，即带有确信度的结果
model.clearThreshold()
val scoreAndLabels = test.map{point =>
val score = model.predict(point.features)
score+" "+point.label
}
scoreAndLabels.foreach(println)
输出结果是：
也可以设置阈值
model.setThreshold(0.0)
scoreAndLabels.foreach(println)
如果我们设定了阀值，则会把大于阈值的结果当成正预测，小于阈值的结果当成负预测。
接下来我们将原始预测结果添加到mysql数据库中
通过查看数据库，看是否导入进去，最好重新连接一下数据库，防止没有更新成功
到这里我们利用spark预测回头客的实验就正式结束了！
