目录
2.数据处理 JieBaTokenizer -> HashingVector
一.引言
Spark 3.0 - 5.ML Pipeline 实战之电影影评情感分析 通过 LR 二分类实现了影评的情感分析，我们主观的将 <= 🌟🌟🌟 的影评标记为 Label=0 即负向评论，>= 🌟🌟🌟🌟 的标记为 Label=1 即正向评论，有失偏颇，本文通过 LR 实现多分类预测影评对应的真实评论分数。
二.LR 多分类分析
LR 二分类问题主要通过样本训练 weights 与 bias，最终将 xTw + b 送到 sigmoid 得到概率值，再通过阈值决定该样本为哪一类，LR 多分类可以看作是多个二分类 LR 的合并：
- 针对每个类别训练得到一个 weight 参数与 bias 参数
- 每个类别对应的 LR 模型分别预测当前样本的概率值
- 取概率值最大的 LR 模型对应的类别作为当前预测样本的类别
整个过程可以看到集成学习、投票法的一些影子，以电影影评预测电影评分为例，我们共有 6个类别分别对应 0 x 🌟 到 🌟🌟🌟🌟🌟，按词库 20000 计算，则训练生成一个 20000 x 6 的参数矩阵以及 1 x 6 的 bias 偏置，通过 Matrix 与 Bias 即可预测新的样本。
其中 1 代表 1 x Bias，前面 wTx 即可。
三.LR 多分类实战
通过 Comment -> RDD -> DF -> JieBaTokenizer -> HashingVector -> LR 的流程实现数据的预处理与训练，再通过得到的 LrModel 实现 Predict。
1.数据准备 Comment -> RDD -> DF
原始数据样式为：
通过 spark 读取该数据得到 movie、score 与 comment 三个字段的 DF，由于 Model.fit 需要 label 列，所以 score 列采用 label 为列名：
-
val spark =
SparkSession
-
.builder
//创建spark会话
-
.master(
"local")
//设置本地模式
-
.appName(
"MultiClassify")
//设置名称
-
.getOrCreate()
//创建会话变量
-
-
spark.sparkContext.setLogLevel(
"error")
-
-
val inputPath =
"./DouBanComment"
-
-
val commentAndLabel = spark.sparkContext.textFile(inputPath).mapPartitions(partition => {
-
-
partition.map(line => {
-
try {
-
val info = line.split(
"##")
-
val movieName = info(
0)
-
val label = info(
1).toDouble
-
val comment = info(
2)
-
(movieName, label, comment)
-
}
catch {
-
case _:
Throwable =>
-
null
-
}
-
})
-
-
}).filter(_ !=
null)
-
-
val data = spark.createDataFrame(commentAndLabel)
-
.toDF(
"movie",
"label",
"comment")
2.数据处理 JieBaTokenizer -> HashingVector
通过 JiebaTokenizer 将影评进行分词，通过 HashingTF 实现向量化，最后将 hashData 通过 randomSplit 方法划分为训练集与数据集，为了每次输出结果相同，这里使用了固定的 seed。
-
val jiebaTonkenizer =
new
JiebaTokenizer()
-
.setInputCol(
"comment")
-
.setOutputCol(
"words")
-
-
val hashingTF =
new
HashingTF()
-
.setNumFeatures(
20000)
-
.setInputCol(
"words")
-
.setOutputCol(
"features")
-
-
val hashData = hashingTF.transform(jiebaTonkenizer.transform(data))
-
-
hashData.limit(
10).show()
-
-
// 划分训练集、测试集
-
val trainAndTestRatio =
Array(
0.95,
0.05)
-
val pipelineData = hashData.randomSplit(trainAndTestRatio,
99)
-
val trainData = pipelineData(
0)
-
val testData = pipelineData(
1)
可以看到原始 comment 通过 Tokenizer 实现了分词，再通过 Hashing 实现了向量化，向量的长度由 HashingTF 的 NumFeatures 决定：
3.模型训练 LR
由于整体训练样本不多且词库相对稀疏，这里也没有设置正则化系数。
-
val lr =
new
LogisticRegression()
-
.setMaxIter(
10)
-
.setElasticNetParam(
0.8)
-
val lrModel = lr.fit(trainData)
获取参数矩阵与截距：
-
// 打印逻辑回归的系数和截距
-
println(
s"Coefficients: \n${lrModel.coefficientMatrix}")
-
println(
s"Intercepts: \n${lrModel.interceptVector}")
Coefficnents 稀疏矩阵维度为 20000 x 6，Intercepts 截距项为 1 x 6。
4.模型评估 Metrics
val trainingSummary = lrModel.summary
通过 Model.summary 可以获取训练中的相关信息。
-
// 获取每次的迭代对象
-
val objectiveHistory = trainingSummary.objectiveHistory
-
println(
"objectiveHistory:")
-
objectiveHistory.foreach(println)
-
-
// 对于多分类问题，我们可以基于每个标签观察矩阵，并打印一些汇总信息
-
println(
"False positive rate by label:")
-
trainingSummary.falsePositiveRateByLabel.zipWithIndex.foreach {
case (rate, label) =>
-
println(
s"label $label: $rate")
-
}
-
-
println(
"True positive rate by label:")
-
trainingSummary.truePositiveRateByLabel.zipWithIndex.foreach {
case (rate, label) =>
-
println(
s"label $label: $rate")
-
}
-
-
println(
"Precision by label:")
-
trainingSummary.precisionByLabel.zipWithIndex.foreach {
case (prec, label) =>
-
println(
s"label $label: $prec")
-
}
-
-
println(
"Recall by label:")
-
trainingSummary.recallByLabel.zipWithIndex.foreach {
case (rec, label) =>
-
println(
s"label $label: $rec")
-
}
-
-
-
println(
"F-measure by label:")
-
trainingSummary.fMeasureByLabel.zipWithIndex.foreach {
case (f, label) =>
-
println(
s"label $label: $f")
-
}
-
-
val accuracy = trainingSummary.accuracy
-
val falsePositiveRate = trainingSummary.weightedFalsePositiveRate
-
val truePositiveRate = trainingSummary.weightedTruePositiveRate
-
val fMeasure = trainingSummary.weightedFMeasure
-
val precision = trainingSummary.weightedPrecision
-
val recall = trainingSummary.weightedRecall
-
println(
s"Accuracy: $accuracy\nFPR: $falsePositiveRate\nTPR: $truePositiveRate\n" +
-
s"F-measure: $fMeasure\nPrecision: $precision\nRecall: $recall")
objectiveHistory 是一个双精度数组，它指定了在每个训练迭代中目标函数的优化过程，可以基于该参数评估模型是否需要进一步迭代或者调整训练参数。
上面我们制定了 Iter 迭代次数为 10，可以看到每次优化的目标函数 Loss 在前两轮训练中下降最快，除此之外，还有基于各个 Label 的参数指标，例如正负样本率、召回率等等，最后打印模型整体评估指标：
5.人工校验 DIY
下面我们通过参数矩阵 Weights 与截距 Bias 手动实现预测并分析测试数据预测结果，套用上面提到的公式计算 fx 结果即可：
-
def calcTop(vector:
Vector, real:
Double):
Unit = {
-
-
val multiply = lrModel.coefficientMatrix.multiply(vector).values
-
val result = multiply.zip(lrModel.interceptVector.toArray).map{
case (wx, b) => {
-
val fx =
1 / (
1 + math.exp(
-1 * (wx + b)))
-
fx
-
}}
-
val rank = result.zipWithIndex.maxBy(_._1)._2
-
println(
s"Pre: ${lrModel.predict(vector)} DiyPre: $rank Real: $real ${result.mkString(",")}")
-
}
下面我们以测试集中的 《肖申克的救赎》影评为例看一下评分：
-
testData.filter(
"movie == '肖申克的救赎'")
-
.select(
"movie",
"label",
"comment",
"features")
-
.take(
10).foreach{
-
case
Row(movie:
String, label:
Double, comment:
String, feature:
Vector) => {
-
println(movie +
"\t" + label +
"\t" + comment +
"\t" + feature)
-
calcTop(feature, label)
-
}}
由于训练样本中该电影评分大部分为 4-5，所以模型训练后也会相应得到较高的分数，第二个预测中，虽然将 5 分预测为 4 分，但是可以通过最终 1x6 的结果数组中看到 4 分与 5 分对应的预测分数相当接近，只相差 0.00008 左右。
四.总结
不论是二分类还是多分类，均使用了 LR 实现，相对比较简单，如果想要更复杂的实现，Python 更适合，我们在前面讲过通过循环神经网络并实现多输出模型的案例 Keras 多输出模型：
这里社交媒体发帖可以类比为电影评论，目标函数 Loss 由 3 个预测目标构成，收入分10类，使用 categorical_crossentropy 损失函数、年龄是 0-100 岁，采用 mse 损失函数、性别是二值，采用 binary_crossentropy 损失函数。这里电影类别可以看做是分为 6 个类别的收入，因此采用 categorical_crossentropy 作为损失函数，删掉另外两个输出类年龄和性别，就实现了相对复杂版的循环神经网络 NLP 语言情感分析。将训练生成的模型保存为 pb 文件，再配合 java tensorflow API 即可使用 Scala / Java 实现复杂深度模型的使用。
转载：https://blog.csdn.net/BIT_666/article/details/128038170