头歌实践教学平台：Spark大数据编程（四十一）
四十一、Spark的机器学习-MLlib
第1关：MLlib介绍
任务描述
本关任务：通过算法，对给出的数据进行 字母 和 组合单词 的分类，并预测结果。
相关知识
MLlib(Machine Learnig lib) 是Spark对常用的机器学习算法的实现库，同时包括相关的测试和数据生成器。MLlib目前支持4种常见的机器学习问题: 分类、回归、聚类和协同过滤。在Spark官方首页中展示了Logistic Regression算法在Spark和Hadoop中运行的性能比较，如图所示:
从图中可以看出，使用 Spark 运行的Logistic Regression算法比直接从 Hadoop 运行快很多，接下来我们来学习 Spark的MLlib。
为了完成本关任务，你需要掌握：
局部向量
Transformers
Estimators
Pipeline
标签
如何使用
编程要求
根据提示，在右侧编辑器补充代码，使用LogisticRegression算法训练 trainingList数据，数据如下：
List<Row> trainingList = Arrays.asList(
RowFactory.create(1.0, "a b c d E spark"),
RowFactory.create(0.0, "b d"),
RowFactory.create(1.0, "hadoop Mapreduce"),
RowFactory.create(0.0, "f g h"));
其中1.0 和 0.0为标签类别，字符串中字母和组合单词都为特征数据。
任务要求：对testList数据的标签类别进行预测，把输出标签prediction字段以表结构进行展示。
package com.educoder.bigData.sparksql5;
import java.util.Arrays;
import java.util.List;
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.*;
import org.apache.spark.ml.feature.HashingTF;
import org.apache.spark.ml.feature.Tokenizer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
public class Test1 {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder().appName("test1").master("local").getOrCreate();
List<Row> trainingList = Arrays.asList(
RowFactory.create(1.0, "a b c d E spark"),
RowFactory.create(0.0, "b d"),
RowFactory.create(1.0, "hadoop Mapreduce"),
RowFactory.create(0.0, "f g h"));
List<Row> testList = Arrays.asList(
RowFactory.create(0.0, "spark I j k"),
RowFactory.create(0.0, "l M n"),
RowFactory.create(0.0, "f g"),
RowFactory.create(0.0, "apache hadoop")
);
/********* Begin *********/
// 1. 定义数据结构（label标签列 + text文本列）
StructType schema = new StructType(new StructField[] {
new StructField("label", DataTypes.DoubleType, false, Metadata.empty()),
new StructField("text", DataTypes.StringType, false, Metadata.empty())
});
// 2. 将训练数据和测试数据转换为DataFrame
Dataset<Row> trainingDF = spark.createDataFrame(trainingList, schema);
Dataset<Row> testDF = spark.createDataFrame(testList, schema);
// 3. 定义Tokenizer分词器：将text列分词后存入words列
Tokenizer tokenizer = new Tokenizer()
.setInputCol("text")
.setOutputCol("words");
// 4. 定义HashingTF：将words列转换为特征向量存入features列
HashingTF hashingTF = new HashingTF()
.setNumFeatures(1000) // 特征维度，可根据需求调整
.setInputCol(tokenizer.getOutputCol())
.setOutputCol("features");
// 5. 定义逻辑回归算法Estimator
LogisticRegression lr = new LogisticRegression()
.setMaxIter(10) // 最大迭代次数
.setRegParam(0.001); // 正则化系数
// 6. 构建Pipeline：串联Tokenizer -> HashingTF -> LogisticRegression
Pipeline pipeline = new Pipeline()
.setStages(new PipelineStage[] {tokenizer, hashingTF, lr});
// 7. 训练模型
PipelineModel model = pipeline.fit(trainingDF);
// 8. 用模型预测测试数据
Dataset<Row> predictions = model.transform(testDF);
// 9. 展示预测结果的prediction字段
predictions.select("prediction").show();
// 关闭SparkSession
spark.stop();
/********* End *********/
}
}
第2关：MLlib-垃圾邮件检测
任务描述
本关任务：通过分类算法完成一个垃圾邮件检测。
相关知识
为了完成本关任务，你需要掌握：
标签标识转化；
分类算法使用。
标签标识转化
当标签标识不为局部向量的数字值向量，使用StringIndexer来完成转换。
StringIndexer labelIndexer = new StringIndexer().setInputCol("label").setOutputCol("indexedLabel");
分类算法使用
分类算法常见的决策树分类器、随机森林分类器、梯度提升树分类器、逻辑回归，MLlib中实现类分别为：DecisionTreeClassifier 、RandomForestClassifier 、GBTClassifier、LogisticRegression，MLlib提供了统一的使用方法，请参考第一关进行使用。
编程要求
根据提示，在右侧编辑器补充代码，从SMSSpamCollection文件读取信息进行训练，并设置向量标签为indexedLabel，平台会对生成的PipelineModel训练模型进行准确性检测。
SMSSpamCollection文件每行开头第一列为标签类别：垃圾邮件和非垃圾邮件，每行数据内容都以空格隔开。如下图：
ham Go until jurong point, crazy.. Available only in bugis n great world la e buffet... Cine there got amore wat...
ham Ok lar... Joking wif u oni...
spam Congrats! 1 year special cinema pass for 2 is yours. call 09061209465 now! C Suprman V, Matrix3, StarWars3
package com.educoder.bigData.sparksql5;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.feature.*;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import java.util.Arrays;
import java.util.List;
public class Case2 {
public static PipelineModel training(SparkSession spark) {
/********* Begin *********/
// 1. 读取文件并过滤空行
JavaRDD<String> lines = spark.read()
.textFile("SMSSpamCollection")
.javaRDD()
.filter(line -> line != null && !line.trim().isEmpty());
// 2. 解析数据：返回单词数组（适配平台的array<string>类型）
JavaRDD<Row> rowRDD = lines.map((Function<String, Row>) line -> {
int firstSpaceIndex = line.indexOf(" ");
if (firstSpaceIndex == -1) {
return RowFactory.create("", new String[0]);
}
String label = line.substring(0, firstSpaceIndex).trim();
String text = line.substring(firstSpaceIndex + 1).trim();
// 拆分文本为单词数组（适配平台的array<string>类型）
String[] words = text.split(" ");
// 过滤空单词
List<String> validWords = Arrays.asList(words);
String[] filteredWords = validWords.stream()
.filter(word -> !word.trim().isEmpty())
.toArray(String[]::new);
return RowFactory.create(label, filteredWords);
})
.filter(row -> !row.getString(0).isEmpty())
.filter(row -> ((String[]) row.get(1)).length > 0);
// 3. 定义Schema：message列为array<string>（匹配平台实际类型）
StructType schema = new StructType(new StructField[]{
new StructField("label", DataTypes.StringType, false, Metadata.empty()),
new StructField("message", DataTypes.createArrayType(DataTypes.StringType), false, Metadata.empty())
});
// 4. 创建DataFrame
Dataset<Row> dataFrame = spark.createDataFrame(rowRDD, schema);
// 5. 标签转换：StringIndexer（任务要求indexedLabel）
StringIndexer labelIndexer = new StringIndexer()
.setInputCol("label")
.setOutputCol("indexedLabel")
.setHandleInvalid("skip");
// 6. 特征提取（适配array<string>输入，跳过Tokenizer）
// 6.1 过滤停用词（直接处理array<string>）
StopWordsRemover stopWordsRemover = new StopWordsRemover()
.setInputCol("message")
.setOutputCol("filtered_words");
// 6.2 词频统计
CountVectorizer countVectorizer = new CountVectorizer()
.setInputCol("filtered_words")
.setOutputCol("raw_features")
.setVocabSize(8000) // 扩大词汇表，提升特征覆盖
.setMinDF(3); // 降低最小文档频率，保留更多特征
// 6.3 TF-IDF转换（核心特征）
IDF idf = new IDF()
.setInputCol("raw_features")
.setOutputCol("features")
.setMinDocFreq(2); // 进一步降低阈值
// 7. 逻辑回归（终极调优，确保正确率>95%）
LogisticRegression lr = new LogisticRegression()
.setLabelCol("indexedLabel")
.setFeaturesCol("features")
.setMaxIter(500) // 最大化迭代次数
.setRegParam(0.0001) // 极弱正则化
.setElasticNetParam(0.0) // 纯L2正则
.setThreshold(0.45); // 微调分类阈值，提升垃圾邮件识别率
// 8. 构建Pipeline（跳过Tokenizer，直接处理array<string>）
Pipeline pipeline = new Pipeline()
.setStages(new PipelineStage[]{
labelIndexer,
stopWordsRemover,
countVectorizer,
idf,
lr
});
// 9. 训练模型
PipelineModel model = pipeline.fit(dataFrame);
/********* End *********/
return model;
}
}
第3关：MLlib-红酒分类预测
任务描述
本关任务：编写实现红酒分类的功能。
相关知识
为了完成本关任务，你需要掌握：
分类算法使用。
分类算法使用
分类算法常见的决策树分类器、随机森林分类器、梯度提升树分类器、逻辑回归，MLlib中实现类分别为：DecisionTreeClassifier 、RandomForestClassifier 、GBTClassifier、LogisticRegression，MLlib提供了统一的使用方法，请参考第一关进行使用。
编程要求
根据提示，在右侧编辑器补充代码，从dataset.csv文件读取信息进行训练，并设置向量标签为label，平台会对生成的PipelineModel训练模型进行准确性检测。
dataset.csv文件每行第一列为标签，代表红酒三个类别，后面的为红酒特征值。内容如下图：
1,14.23,1.71,2.43,15.6,127,2.8,3.06,.28,2.29,5.64,1.04,3.92,1065
1,13.2,1.78,2.14,11.2,100,2.65,2.76,.26,1.28,4.38,1.05,3.4,1050
1,13.16,2.36,2.67,18.6,101,2.8,3.24,.3,2.81,5.68,1.03,3.17,1185
2,12.33,.99,1.95,14.8,136,1.9,1.85,.35,2.76,3.4,1.06,2.31,750
2,12.7,3.87,2.4,23,101,2.83,2.55,.43,1.95,2.57,1.19,3.13,463
2,12,.92,2,19,86,2.42,2.26,.3,1.43,2.5,1.38,3.12,278
3,13.84,4.12,2.38,19.5,89,1.8,.83,.48,1.56,9.01,.57,1.64,480
3,12.45,3.03,2.64,27,97,1.9,.58,.63,1.14,7.5,.67,1.73,880
3,14.34,1.68,2.7,25,98,2.8,1.31,.53,2.7,13,.57,1.96,660
package com.educoder.bigData.sparksql5;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.RandomForestClassifier;
import org.apache.spark.ml.linalg.Vectors;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
public class Case3 {
public static PipelineModel training(SparkSession spark) {
/********* Begin *********/
// 1. 读取dataset.csv文件，过滤空行
JavaRDD<String> lines = spark.read()
.textFile("dataset.csv")
.javaRDD()
.filter(line -> line != null && !line.trim().isEmpty());
// 2. 解析每行数据：直接组装为label + features向量（适配平台结构）
JavaRDD<Row> rowRDD = lines.map((Function<String, Row>) line -> {
// 按逗号拆分每行数据
String[] parts = line.split(",");
// 第一列是标签（红酒类别：1/2/3），转为Double类型
double label = Double.parseDouble(parts[0]);
// 剩余列是特征值，组装为MLlib向量
double[] featureValues = new double[parts.length - 1];
for (int i = 1; i < parts.length; i++) {
featureValues[i - 1] = Double.parseDouble(parts[i]);
}
// 直接创建features向量（平台已识别的字段名）
return RowFactory.create(label, Vectors.dense(featureValues));
});
// 3. 定义Schema：label + features（匹配平台数据结构）
StructType schema = new StructType(new StructField[]{
// 任务要求的标签列：label
new StructField("label", DataTypes.DoubleType, false, Metadata.empty()),
// 平台已有的特征向量列：features
new StructField("features", org.apache.spark.ml.linalg.SQLDataTypes.VectorType(), false, Metadata.empty())
});
// 4. 创建DataFrame
Dataset<Row> dataFrame = spark.createDataFrame(rowRDD, schema);
// 5. 选择随机森林分类器（红酒多分类最优算法）
RandomForestClassifier classifier = new RandomForestClassifier()
.setLabelCol("label") // 任务要求的标签列名
.setFeaturesCol("features") // 平台已有的特征列名
.setNumTrees(30) // 增加树数量，提升正确率
.setMaxDepth(10) // 优化树深度
.setImpurity("gini") // 基尼系数，适配多分类
.setSeed(12345); // 固定随机种子
// 6. 构建Pipeline：直接训练分类器（无需VectorAssembler）
Pipeline pipeline = new Pipeline()
.setStages(new PipelineStage[]{classifier});
// 7. 训练模型
PipelineModel model = pipeline.fit(dataFrame);
/********* End *********/
return model;
}
}
有任何问题都可以随时关注私信！