四十一、Spark的机器学习-MLlib第1关MLlib介绍任务描述本关任务通过算法对给出的数据进行 字母 和 组合单词 的分类并预测结果。相关知识MLlib(Machine Learnig lib) 是Spark对常用的机器学习算法的实现库同时包括相关的测试和数据生成器。MLlib目前支持4种常见的机器学习问题: 分类、回归、聚类和协同过滤。在Spark官方首页中展示了Logistic Regression算法在Spark和Hadoop中运行的性能比较如图所示:从图中可以看出使用 Spark 运行的Logistic Regression算法比直接从 Hadoop 运行快很多接下来我们来学习 Spark的MLlib。为了完成本关任务你需要掌握局部向量TransformersEstimatorsPipeline标签如何使用编程要求根据提示在右侧编辑器补充代码使用LogisticRegression算法训练 trainingList数据数据如下ListRow 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();ListRow 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));ListRow 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. 将训练数据和测试数据转换为DataFrameDatasetRow trainingDF spark.createDataFrame(trainingList, schema);DatasetRow 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. 定义逻辑回归算法EstimatorLogisticRegression lr new LogisticRegression().setMaxIter(10) // 最大迭代次数.setRegParam(0.001); // 正则化系数// 6. 构建Pipeline串联Tokenizer - HashingTF - LogisticRegressionPipeline pipeline new Pipeline().setStages(new PipelineStage[] {tokenizer, hashingTF, lr});// 7. 训练模型PipelineModel model pipeline.fit(trainingDF);// 8. 用模型预测测试数据DatasetRow predictions model.transform(testDF);// 9. 展示预测结果的prediction字段predictions.select(prediction).show();// 关闭SparkSessionspark.stop();/********* End *********/}}第2关MLlib-垃圾邮件检测任务描述本关任务通过分类算法完成一个垃圾邮件检测。相关知识为了完成本关任务你需要掌握标签标识转化分类算法使用。标签标识转化当标签标识不为局部向量的数字值向量使用StringIndexer来完成转换。StringIndexer labelIndexer new StringIndexer().setInputCol(label).setOutputCol(indexedLabel);分类算法使用分类算法常见的决策树分类器、随机森林分类器、梯度提升树分类器、逻辑回归MLlib中实现类分别为DecisionTreeClassifier 、RandomForestClassifier 、GBTClassifier、LogisticRegressionMLlib提供了统一的使用方法请参考第一关进行使用。编程要求根据提示在右侧编辑器补充代码从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, StarWars3package 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. 读取文件并过滤空行JavaRDDString lines spark.read().textFile(SMSSpamCollection).javaRDD().filter(line - line ! null !line.trim().isEmpty());// 2. 解析数据返回单词数组适配平台的arraystring类型JavaRDDRow rowRDD lines.map((FunctionString, 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();// 拆分文本为单词数组适配平台的arraystring类型String[] words text.split( );// 过滤空单词ListString 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. 定义Schemamessage列为arraystring匹配平台实际类型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. 创建DataFrameDatasetRow dataFrame spark.createDataFrame(rowRDD, schema);// 5. 标签转换StringIndexer任务要求indexedLabelStringIndexer labelIndexer new StringIndexer().setInputCol(label).setOutputCol(indexedLabel).setHandleInvalid(skip);// 6. 特征提取适配arraystring输入跳过Tokenizer// 6.1 过滤停用词直接处理arraystringStopWordsRemover 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直接处理arraystringPipeline 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、LogisticRegressionMLlib提供了统一的使用方法请参考第一关进行使用。编程要求根据提示在右侧编辑器补充代码从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,10651,13.2,1.78,2.14,11.2,100,2.65,2.76,.26,1.28,4.38,1.05,3.4,10501,13.16,2.36,2.67,18.6,101,2.8,3.24,.3,2.81,5.68,1.03,3.17,11852,12.33,.99,1.95,14.8,136,1.9,1.85,.35,2.76,3.4,1.06,2.31,7502,12.7,3.87,2.4,23,101,2.83,2.55,.43,1.95,2.57,1.19,3.13,4632,12,.92,2,19,86,2.42,2.26,.3,1.43,2.5,1.38,3.12,2783,13.84,4.12,2.38,19.5,89,1.8,.83,.48,1.56,9.01,.57,1.64,4803,12.45,3.03,2.64,27,97,1.9,.58,.63,1.14,7.5,.67,1.73,8803,14.34,1.68,2.7,25,98,2.8,1.31,.53,2.7,13,.57,1.96,660package 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文件过滤空行JavaRDDString lines spark.read().textFile(dataset.csv).javaRDD().filter(line - line ! null !line.trim().isEmpty());// 2. 解析每行数据直接组装为label features向量适配平台结构JavaRDDRow rowRDD lines.map((FunctionString, 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. 定义Schemalabel features匹配平台数据结构StructType schema new StructType(new StructField[]{// 任务要求的标签列labelnew StructField(label, DataTypes.DoubleType, false, Metadata.empty()),// 平台已有的特征向量列featuresnew StructField(features, org.apache.spark.ml.linalg.SQLDataTypes.VectorType(), false, Metadata.empty())});// 4. 创建DataFrameDatasetRow dataFrame spark.createDataFrame(rowRDD, schema);// 5. 选择随机森林分类器红酒多分类最优算法RandomForestClassifier classifier new RandomForestClassifier().setLabelCol(label) // 任务要求的标签列名.setFeaturesCol(features) // 平台已有的特征列名.setNumTrees(30) // 增加树数量提升正确率.setMaxDepth(10) // 优化树深度.setImpurity(gini) // 基尼系数适配多分类.setSeed(12345); // 固定随机种子// 6. 构建Pipeline直接训练分类器无需VectorAssemblerPipeline pipeline new Pipeline().setStages(new PipelineStage[]{classifier});// 7. 训练模型PipelineModel model pipeline.fit(dataFrame);/********* End *********/return model;}}有任何问题都可以随时关注私信 头歌实践教学平台：Spark大数据编程（四十一）
四十一、Spark的机器学习-MLlib第1关MLlib介绍任务描述本关任务通过算法对给出的数据进行 字母 和 组合单词 的分类并预测结果。相关知识MLlib(Machine Learnig lib) 是Spark对常用的机器学习算法的实现库同时包括相关的测试和数据生成器。MLlib目前支持4种常见的机器学习问题: 分类、回归、聚类和协同过滤。在Spark官方首页中展示了Logistic Regression算法在Spark和Hadoop中运行的性能比较如图所示:从图中可以看出使用 Spark 运行的Logistic Regression算法比直接从 Hadoop 运行快很多接下来我们来学习 Spark的MLlib。为了完成本关任务你需要掌握局部向量TransformersEstimatorsPipeline标签如何使用编程要求根据提示在右侧编辑器补充代码使用LogisticRegression算法训练 trainingList数据数据如下ListRow 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();ListRow 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));ListRow 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. 将训练数据和测试数据转换为DataFrameDatasetRow trainingDF spark.createDataFrame(trainingList, schema);DatasetRow 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. 定义逻辑回归算法EstimatorLogisticRegression lr new LogisticRegression().setMaxIter(10) // 最大迭代次数.setRegParam(0.001); // 正则化系数// 6. 构建Pipeline串联Tokenizer - HashingTF - LogisticRegressionPipeline pipeline new Pipeline().setStages(new PipelineStage[] {tokenizer, hashingTF, lr});// 7. 训练模型PipelineModel model pipeline.fit(trainingDF);// 8. 用模型预测测试数据DatasetRow predictions model.transform(testDF);// 9. 展示预测结果的prediction字段predictions.select(prediction).show();// 关闭SparkSessionspark.stop();/********* End *********/}}第2关MLlib-垃圾邮件检测任务描述本关任务通过分类算法完成一个垃圾邮件检测。相关知识为了完成本关任务你需要掌握标签标识转化分类算法使用。标签标识转化当标签标识不为局部向量的数字值向量使用StringIndexer来完成转换。StringIndexer labelIndexer new StringIndexer().setInputCol(label).setOutputCol(indexedLabel);分类算法使用分类算法常见的决策树分类器、随机森林分类器、梯度提升树分类器、逻辑回归MLlib中实现类分别为DecisionTreeClassifier 、RandomForestClassifier 、GBTClassifier、LogisticRegressionMLlib提供了统一的使用方法请参考第一关进行使用。编程要求根据提示在右侧编辑器补充代码从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, StarWars3package 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. 读取文件并过滤空行JavaRDDString lines spark.read().textFile(SMSSpamCollection).javaRDD().filter(line - line ! null !line.trim().isEmpty());// 2. 解析数据返回单词数组适配平台的arraystring类型JavaRDDRow rowRDD lines.map((FunctionString, 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();// 拆分文本为单词数组适配平台的arraystring类型String[] words text.split( );// 过滤空单词ListString 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. 定义Schemamessage列为arraystring匹配平台实际类型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. 创建DataFrameDatasetRow dataFrame spark.createDataFrame(rowRDD, schema);// 5. 标签转换StringIndexer任务要求indexedLabelStringIndexer labelIndexer new StringIndexer().setInputCol(label).setOutputCol(indexedLabel).setHandleInvalid(skip);// 6. 特征提取适配arraystring输入跳过Tokenizer// 6.1 过滤停用词直接处理arraystringStopWordsRemover 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直接处理arraystringPipeline 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、LogisticRegressionMLlib提供了统一的使用方法请参考第一关进行使用。编程要求根据提示在右侧编辑器补充代码从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,10651,13.2,1.78,2.14,11.2,100,2.65,2.76,.26,1.28,4.38,1.05,3.4,10501,13.16,2.36,2.67,18.6,101,2.8,3.24,.3,2.81,5.68,1.03,3.17,11852,12.33,.99,1.95,14.8,136,1.9,1.85,.35,2.76,3.4,1.06,2.31,7502,12.7,3.87,2.4,23,101,2.83,2.55,.43,1.95,2.57,1.19,3.13,4632,12,.92,2,19,86,2.42,2.26,.3,1.43,2.5,1.38,3.12,2783,13.84,4.12,2.38,19.5,89,1.8,.83,.48,1.56,9.01,.57,1.64,4803,12.45,3.03,2.64,27,97,1.9,.58,.63,1.14,7.5,.67,1.73,8803,14.34,1.68,2.7,25,98,2.8,1.31,.53,2.7,13,.57,1.96,660package 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文件过滤空行JavaRDDString lines spark.read().textFile(dataset.csv).javaRDD().filter(line - line ! null !line.trim().isEmpty());// 2. 解析每行数据直接组装为label features向量适配平台结构JavaRDDRow rowRDD lines.map((FunctionString, 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. 定义Schemalabel features匹配平台数据结构StructType schema new StructType(new StructField[]{// 任务要求的标签列labelnew StructField(label, DataTypes.DoubleType, false, Metadata.empty()),// 平台已有的特征向量列featuresnew StructField(features, org.apache.spark.ml.linalg.SQLDataTypes.VectorType(), false, Metadata.empty())});// 4. 创建DataFrameDatasetRow dataFrame spark.createDataFrame(rowRDD, schema);// 5. 选择随机森林分类器红酒多分类最优算法RandomForestClassifier classifier new RandomForestClassifier().setLabelCol(label) // 任务要求的标签列名.setFeaturesCol(features) // 平台已有的特征列名.setNumTrees(30) // 增加树数量提升正确率.setMaxDepth(10) // 优化树深度.setImpurity(gini) // 基尼系数适配多分类.setSeed(12345); // 固定随机种子// 6. 构建Pipeline直接训练分类器无需VectorAssemblerPipeline pipeline new Pipeline().setStages(new PipelineStage[]{classifier});// 7. 训练模型PipelineModel model pipeline.fit(dataFrame);/********* End *********/return model;}}有任何问题都可以随时关注私信