步骤四、将数据从Hive导入到MySQL，spark预测回头客

上一篇：一套简单但完整的伪分布式大数据分析流程(三)(图文详解)，适用于Windows系统https://blog.csdn.net/qq_42800581/article/details/106268361

12.Hive导入数据到MySQL

因为需要借助于MySQL保存Hive的元数据，所以，首先启动MySQL数据库，在终端中输入下面命令：

service mysql start  # 可以在Linux的任何目录下执行该命令

再启动Hadoop，然后启动Hive。

执行下面命令启动Hadoop（如果你已经启动了Hadoop就不用再次启动了）：

cd /usr/local/hadoop
./sbin/start-all.sh

然后，执行jps命令看一下当前运行的进程：

jps

如果出现下面这些进程，说明Hadoop启动成功了。

3765 NodeManager
3639 ResourceManager
3800 Jps
3261 DataNode
3134NameNode
3471 SecondaryNameNode

继续执行下面命令启动进入Hive：

cd /usr/local/hive
./bin/hive   #启动Hive

通过上述过程，我们就完成了MySQL、Hadoop和Hive三者的启动。
启动成功以后，就进入了“hive>”命令提示符状态，可以输入类似SQL语句的HiveQL语句。

然后，在“hive>”命令提示符状态下执行下面命令：

1、创建临时表inner_user_log和inner_user_info(这个命令 又有点长)

create table dbtaobao.inner_user_log(user_id INT,item_id INT,cat_id INT,merchant_id INT,brand_id INT,month STRING,day STRING,action INT,age_range INT,gender INT,province STRING) COMMENT 'Welcome to XMU dblab! Now create inner table inner_user_log ' ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE;

这个命令执行完以后，Hive会自动在HDFS文件系统中创建对应的数据文件“/user/hive/warehouse/dbtaobao.db/inner_user_log”。

2、将user_log表中的数据插入到inner_user_log,
面把dbtaobao.user_log数据插入到dbtaobao.inner_user_log表中：

INSERT OVERWRITE TABLE dbtaobao.inner_user_log select * from dbtaobao.user_log;

执行下面命令查询上面的插入命令是否成功执行：

select * from inner_user_log limit 10;

退出hive环境

exit;

使用Sqoop将数据从Hive导入MySQL
1、启动Hadoop集群、MySQL服务
前面我们已经启动了Hadoop集群和MySQL服务。这里请确认已经按照前面操作启动成功。

2、将前面生成的临时表数据从Hive导入到 MySQL 中，包含如下四个步骤。
(1)在Linux系统中新建一个终端，执行下面命令：

mysql -u root -p

为了简化操作，本教程直接使用root用户登录MySQL数据库，但是，在实际应用中，建议在MySQL中再另外创建一个用户。
执行上面命令以后，就进入了“mysql>”命令提示符状态。

(2)创建数据库

show databases; 	#显示所有数据库
create database dbtaobao; 	#创建dbtaobao数据库
use dbtaobao;	 #使用数据库

注意：请使用下面命令查看数据库的编码：

show variables like "char%";

请确认当前结果与下边类似，当前编码为utf8，否则无法导入中文：
±-------------------------±---------------------------+
| Variable_name | Value |
±-------------------------±---------------------------+
| character_set_client | utf8 |
| character_set_connection | utf8 |
| character_set_database | utf8 |
| character_set_filesystem | binary |
| character_set_results | utf8 |
| character_set_server | utf8 |
| character_set_system | utf8 |
| character_sets_dir | /usr/share/mysql/charsets/ |
±-------------------------±---------------------------+
8 rows in set (0.00 sec)

(3)创建表
下面在MySQL的数据库dbtaobao中创建一个新表user_log，并设置其编码为utf-8：

CREATE TABLE `dbtaobao`.`user_log` (`user_id` varchar(20),`item_id` varchar(20),`cat_id` varchar(20),`merchant_id` varchar(20),`brand_id` varchar(20), `month` varchar(6),`day` varchar(6),`action` varchar(6),`age_range` varchar(6),`gender` varchar(6),`province` varchar(10)) ENGINE=InnoDB DEFAULT CHARSET=utf8;

提示：语句中的引号是反引号`，不是单引号’。需要注意的是，sqoop抓数据的时候会把类型转为string类型，所以mysql设计字段的时候，设置为varchar。

创建成功后，输入下面命令退出MySQL：

exit;

(4)导入数据(执行时间：20秒左右)
注意，刚才已经退出MySQL，回到了Shell命令提示符状态。下面就可以执行数据导入操作，

cd /usr/local/sqoop
bin/sqoop export --connect jdbc:mysql://localhost:3306/dbtaobao?useSSL=false --username root --password hadoop --table user_log --export-dir '/user/hive/warehouse/dbtaobao.db/inner_user_log' --fields-terminated-by ',';

字段解释： ./bin/sqoop export ##表示数据从 hive 复制到 mysql 中
–connect jdbc:mysql://localhost:3306/dbtaobao
–username root #mysql登陆用户名
–password root #登录密码
–table user_log #mysql 中的表，即将被导入的表名称
–export-dir ‘/user/hive/warehouse/dbtaobao.db/user_log ‘
#hive 中被导出的文件
–fields-terminated-by ‘,’ #Hive 中被导出的文件字段的分隔符

3、查看MySQL中user_log或user_info表中的数据
下面需要再次启动MySQL，进入“mysql>”命令提示符状态：

mysql -u root -p

然后执行下面命令查询user_action表中的数据：

use dbtaobao;
select * from user_log limit 10;

会得到类似下面的查询结果：
±--------±--------±-------±------------±---------±------±-----±-------±----------±-------±----------+
| user_id | item_id | cat_id | merchant_id | brand_id | month | day | action | age_range | gender | province |
±--------±--------±-------±------------±---------±------±-----±-------±----------±-------±----------+
| 414196 | 1109106 | 1188 | 3518 | 4805 | 11 | 11 | 0 | 4 | 0 | 宁夏 |
| 414196 | 380046 | 4 | 231 | 6065 | 11 | 11 | 0 | 5 | 2 | 陕西 |
| 414196 | 1109106 | 1188 | 3518 | 4805 | 11 | 11 | 0 | 7 | 0 | 山西 |
| 414196 | 1109106 | 1188 | 3518 | 4805 | 11 | 11 | 0 | 6 | 0 | 河南 |
| 414196 | 1109106 | 1188 | 3518 | 763 | 11 | 11 | 2 | 2 | 0 | 四川 |
| 414196 | 944554 | 1432 | 323 | 320 | 11 | 11 | 2 | 7 | 2 | 青海 |
| 414196 | 1110009 | 1188 | 298 | 7907 | 11 | 11 | 2 | 3 | 1 | 澳门 |
| 414196 | 146482 | 513 | 2157 | 6539 | 11 | 11 | 0 | 1 | 0 | 上海市 |
| 414196 | 944554 | 1432 | 323 | 320 | 11 | 11 | 0 | 2 | 1 | 宁夏 |
| 414196 | 1109106 | 1188 | 3518 | 4805 | 11 | 11 | 0 | 7 | 0 | 新疆 |
±--------±--------±-------±------------±---------±------±-----±-------±----------±-------±----------+
10 rows in set (0.03 sec)

从Hive导入数据到MySQL中，成功！
接下来退出MySQL环境：

exit;

13.利用Spark预测回头客

预处理test.csv和train.csv数据集
这里列出test.csv和train.csv中字段的描述，字段定义如下：

user_id | 买家id age_range |
买家年龄分段：1表示年龄<18,2表示年龄在[18,24]，3表示年龄在[25,29]，4表示年龄在[30,34]，5表示年龄在[35,39]，6表示年龄在[40,49]，7和8表示年龄>=50,0和NULL则表示未知
gender | 性别:0表示女性，1表示男性，2和NULL表示未知 merchant_id | 商家id label |
是否是回头客，0值表示不是回头客，1值表示回头客，-1值表示该用户已经超出我们所需要考虑的预测范围。NULL值只存在测试集，在测试集中表示需要预测的值。
这里需要预先处理test.csv数据集，把这test.csv数据集里label字段表示-1值剔除掉,保留需要预测的数据.并假设需要预测的数据中label字段均为1.

cd /usr/local/dbtaobao/dataset
vim predeal_test.sh

上面使用vim编辑器新建了一个predeal_test.sh脚本文件，请在这个脚本文件中加入下面代码：

#!/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","1if(id==10000){exit}}}' $infile > $outfile

下面就可以执行predeal_test.sh脚本文件，截取测试数据集需要预测的数据到test_after.csv，命令如下：

chmod +x ./predeal_test.sh
./predeal_test.sh ./test.csv ./test_after.csv

train.csv的第一行都是字段名称，不需要第一行字段名称,这里在对train.csv做数据预处理时，删除第一行

sed -i '1d' train.csv

然后剔除掉train.csv中字段值部分字段值为空的数据。

cd /usr/local/dbtaobao/dataset
vim predeal_train.sh

上面使用vim编辑器新建了一个predeal_train.sh脚本文件，请在这个脚本文件中加入下面代码：

#!/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","$5if(id==10000){exit}}}' $infile > $outfile

保存并退出文件。
下面就可以执行predeal_train.sh脚本文件，截取测试数据集需要预测的数据到train_after.csv，命令如下：

chmod +x ./predeal_train.sh
./predeal_train.sh ./train.csv ./train_after.csv

预测回头客

启动hadoop，如果Hadoop没有运行，执行如下命令：

cd /usr/local/hadoop/
sbin/start-dfs.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

启动MySQL服务

service mysql start
mysql -uroot -p #会提示让你输入数据库密码

输入密码后，你就可以进入“mysql>”命令提示符状态，然后就可以输入下面的SQL语句完成表的创建：

use dbtaobao;
create table rebuy (score varchar(40),label varchar(40));

退出MySQL shell环境：

exit;

启动Spark Shell
Spark支持通过JDBC方式连接到其他数据库获取数据生成DataFrame。

cd ~/下载/
sudo tar -zxvf ./mysql-connector-java-5.1.49.tar.gz -C /usr/local/spark/jars

接下来正式启动spark-shell

cd /usr/local/spark
./bin/spark-shell --jars /usr/local/spark/jars/mysql-connector-java-5.1.49/mysql-connector-java-5.1.49-bin.jar --driver-class-path /usr/local/spark/jars/mysql-connector-java-5.1.49/mysql-connector-java-5.1.49-bin.jar

启动成功后如图所示

14.向量机SVM分类器预测回头客

这里使用Spark MLlib自带的支持向量机SVM分类器进行预测回头客
在spark-shell中执行如下操作：

1.导入需要的包
首先，我们导入需要的包(这部分可以整段复制，最后别忘敲回车)：

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

2.读取训练数据
首先，读取训练文本文件；然后，通过map将每行的数据用“,”隔开，

在数据集中，每行被分成了5部分，前4部分是用户交易的3个特征(age_range,gender,merchant_id)，最后一部分是用户交易的分类(label)。把这里我们用LabeledPoint来存储标签列和特征列。LabeledPoint在监督学习中常用来存储标签和特征，其中要求标签的类型是double，特征的类型是Vector。

val train_data = sc.textFile("/dbtaobao/dataset/train_after.csv")
val test_data = sc.textFile("/dbtaobao/dataset/test_after.csv")

3.构建模型(这部分整段复制，最后别忘敲回车)

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))
}

接下来，通过训练集构建模型SVMWithSGD。这里的SGD即著名的随机梯度下降算法（Stochastic Gradient Descent）。设置迭代次数为1000，除此之外还有stepSize（迭代步伐大小），regParam（regularization正则化控制参数），miniBatchFraction（每次迭代参与计算的样本比例），initialWeights（weight向量初始值）等参数可以进行设置。

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)

spark-shell会打印出如下结果

…
-59045.132228013084 1.0
-81550.17634254562 1.0
-87393.69932070676 1.0
-34743.183626268634 1.0
-42541.544145105494 1.0
-75530.22669142077 1.0
-84157.31973688163 1.0
-18673.911440386535 1.0
-43765.52530945006 1.0
-80524.44350315288 1.0
-61709.836501153935 1.0
-37486.854426141384 1.0
-79793.17112276069 1.0
-21754.021986991942 1.0
-50378.971923247285 1.0
-11646.722569368836 1.0
…

如果我们设定了阀值，则会把大于阈值的结果当成正预测，小于阈值的结果当成负预测。

model.setThreshold(0.0)
scoreAndLabels.foreach(println)

把结果添加到mysql数据库中
现在我们上面没有设定阀值的测试集结果存入到MySQL数据中。
(这部分整段复制，最后别忘敲回车)

model.clearThreshold()
val scoreAndLabels = test.map{point =>val score = model.predict(point.features)score+" "+point.label
}
//设置回头客数据
val rebuyRDD = scoreAndLabels.map(_.split(" "))
/下面要设置模式信息
val schema = StructType(List(StructField("score", StringType, true),StructField("label", StringType, true)))
//下面创建Row对象，每个Row对象都是rowRDD中的一行
val rowRDD = rebuyRDD.map(p => Row(p(0).trim, p(1).trim))
//建立起Row对象和模式之间的对应关系，也就是把数据和模式对应起来
val rebuyDF = spark.createDataFrame(rowRDD, schema)
//下面创建一个prop变量用来保存JDBC连接参数
val prop = new Properties()
prop.put("user", "root") //表示用户名是root
prop.put("password", "hadoop") //表示密码是hadoop
prop.put("driver","com.mysql.jdbc.Driver") //表示驱动程序是com.mysql.jdbc.Driver
//下面就可以连接数据库，采用append模式，表示追加记录到数据库dbtaobao的rebuy表中
rebuyDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/dbtaobao?useSSL=false", "dbtaobao.rebuy", prop)

退出Scala shell环境；

:quit

到这里，第四个步骤的实验内容顺利结束。

行百里者半于九十，你已经走到九十里啦!