• Java应用|使用Apache Spark MLlib构建机器学习模型

    如果您觉得本博客的内容对您有所帮助或启发，请关注我的博客，以便第一时间获取最新技术文章和教程。同时，也欢迎您在评论区留言，分享想法和建议。谢谢支持！

    一、引言

    1.1 Spark MLlib简介

    Apache Spark MLlib（Machine Learning library）是一个开源机器学习框架，建立在Apache Spark之上，支持分布式计算和大规模数据处理。它提供了许多经典机器学习算法和工具，如分类、回归、聚类、协同过滤、特征提取和数据预处理等。

    Spark MLlib使用基于DataFrame的API，提供了一个易于使用的高级API，使得用户能够快速构建、训练和调整机器学习模型，而无需担心底层分布式计算的复杂性。它还支持分布式模型选择和调整，以及与其他Apache Spark组件的集成，如Spark SQL、Spark Streaming和GraphX。

    Spark MLlib还提供了Python、Java和Scala等多种编程语言的API，使得不同的开发人员可以使用他们最喜欢的编程语言来开发机器学习应用程序。

    总之，Spark MLlib是一个非常强大和灵活的机器学习框架，适用于处理大规模数据和需要分布式计算的场景。

    1.2 为什么选择使用Spark MLlib

    1. 处理大规模数据：Spark MLlib支持分布式计算和大规模数据处理，使得处理大规模数据集变得容易。
    2. 丰富的算法库：Spark MLlib包含了许多经典的机器学习算法和工具，如分类、回归、聚类、协同过滤、特征提取和数据预处理等，覆盖了大部分机器学习应用场景。
    3. 高性能：Spark MLlib基于Apache Spark，使用内存计算和RDD（弹性分布式数据集）等优化技术，可以在处理大规模数据时提供高性能和可扩展性。
    4. 易于使用：Spark MLlib提供了一个易于使用的高级API，使得用户可以快速构建、训练和调整机器学习模型，而无需担心底层分布式计算的复杂性。
    5. 多语言支持：Spark MLlib支持多种编程语言的API，包括Python、Java和Scala等，使得不同的开发人员可以使用他们最喜欢的编程语言来开发机器学习应用程序。

    二、Spark MLlib基础

    2.1 RDD和DataFrame的比较

    1. 数据类型：基础RDD可以包含任意类型的数据，包括对象、原始类型、数组和集合等；DataFrame则是一种表格化的数据结构，其数据类型必须是统一的，且可以使用SQL-like的语法进行查询。
    2. 内存计算：DataFrame利用内存计算技术，相比基础RDD更加高效。
    3. 可读性：DataFrame比基础RDD更加易于阅读和理解，可以使用SQL-like的语法进行查询，更加直观。
    4. 类型安全：DataFrame是类型安全的，可以在编译期间捕获类型错误，避免运行时错误；而基础RDD则是类型不安全的，需要在运行时进行类型检查。
    5. 执行计划：基础RDD提供了更加灵活的执行计划，用户可以控制计算的方式和顺序，但这也增加了开发复杂度；而DataFrame则有一个自动优化的执行计划，可以自动优化查询性能。

    总之，基础RDD更加灵活和可控，但需要开发人员自己掌握计算的方式和顺序；而DataFrame则更加易于使用和高效，适合快速开发和迭代。选择使用哪种数据结构，取决于具体的场景和需求。

    2.2 数据准备和预处理

    在使用Spark MLlib进行机器学习之前，需要对原始数据进行预处理和准备。以下是一些常见的数据准备和预处理步骤：

    1. 数据清洗：删除缺失值、处理异常值和重复值等。
    2. 特征选择：选择对模型有用的特征，去除冗余和无关的特征。
    3. 特征缩放：对特征进行缩放，以便它们具有相似的范围和重要性。
    4. 特征变换：将原始特征转换为更有意义的特征，如使用对数、指数、平方根等函数进行变换。
    5. 特征归一化：将特征值归一化为标准正态分布，使得模型更容易学习。
    6. 数据转换：将数据转换为适合模型训练的格式，如将分类变量转换为二进制变量、将文本转换为向量等。

    在Spark MLlib中，可以使用各种预处理和数据准备工具，如：

    1. Imputer：用于填充缺失值。
    2. StandardScaler：用于特征缩放和归一化。
    3. VectorAssembler：用于将多个特征列组合成一个向量列。
    4. OneHotEncoder：用于将分类变量转换为二进制变量。
    5. StringIndexer和IndexToString：用于将字符串类型的变量转换为数字类型的变量。
    6. Tokenizer和StopWordsRemover：用于将文本转换为向量。

    总之，在使用Spark MLlib进行机器学习之前，需要对原始数据进行预处理和准备。Spark MLlib提供了许多工具和功能，可以帮助我们轻松地完成这些任务。

    2.3 特征提取和转换

    在Spark MLlib中，有许多常用的特征提取和转换工具，包括：

    1. Tokenizer：用于将文本转换为单词或词条。
    2. StopWordsRemover：用于去除文本中的停用词，如“the”、“and”等。
    3. CountVectorizer：用于将文本转换为词频向量。
    4. HashingTF：用于将文本转换为哈希向量，可以减少维度并提高计算效率。
    5. IDF：用于计算逆文档频率，可以减少常见词语的权重，提高稀有词语的权重。
    6. Word2Vec：用于将文本转换为向量，可以捕捉词语之间的语义关系。
    7. PCA：用于将高维特征空间降维，可以提高计算效率并避免过拟合。
    8. StringIndexer：用于将分类变量转换为数字类型的变量。
    9. OneHotEncoder：用于将数字类型的变量转换为二进制变量。

    以上这些工具都可以用于特征提取和转换，帮助我们将原始数据转换为模型可以处理的格式。我们可以根据具体的任务和数据类型选择适当的工具，以获得更好的结果。值得注意的是，这些工具的使用通常需要进行适当的参数设置和调整，以达到最佳的效果。

    三、监督学习

    3.1 分类问题

    3.1.1 逻辑回归

    逻辑回归是一种二元分类模型，它的目标是根据已知数据对一个事物进行分类。逻辑回归的输出是一个概率值，代表该事物属于某个类别的概率。如果概率值大于阈值，则将其分类为正类，否则分类为负类。

    在 Spark MLlib 中，可以使用 LogisticRegression 类来实现逻辑回归。下面是一个 Java 版本的示例代码：

    pom引用：

    1. <dependencies>
    2.     <!-- Spark core dependencies -->
    3.     <dependency>
    4.       <groupId>org.apache.spark</groupId>
    5.       <artifactId>spark-core_2.12</artifactId>
    6.       <version>3.2.0</version>
    7.     </dependency>
    8.     <dependency>
    9.       <groupId>org.apache.spark</groupId>
    10.       <artifactId>spark-sql_2.12</artifactId>
    11.       <version>3.2.0</version>
    12.     </dependency>
    13.     <dependency>
    14.       <groupId>org.apache.spark</groupId>
    15.       <artifactId>spark-mllib_2.12</artifactId>
    16.       <version>3.2.0</version>
    17.     </dependency>
    18.  
    19.     <!-- Spark testing dependencies (optional) -->
    20.     <dependency>
    21.       <groupId>org.apache.spark</groupId>
    22.       <artifactId>spark-streaming_2.12</artifactId>
    23.       <version>3.2.0</version>
    24.       <scope>test</scope>
    25.     </dependency>
    26.     <dependency>
    27.       <groupId>org.apache.spark</groupId>
    28.       <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
    29.       <version>3.2.0</version>
    30.       <scope>test</scope>
    31.     </dependency>
    32.     <dependency>
    33.       <groupId>org.apache.spark</groupId>
    34.       <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
    35.       <version>3.2.0</version>
    36.       <scope>test</scope>
    37.     </dependency>
    38.   </dependencies>
    1. import org.apache.spark.ml.classification.LogisticRegression;
    2. import org.apache.spark.ml.classification.LogisticRegressionModel;
    3. import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator;
    4. import org.apache.spark.ml.feature.VectorAssembler;
    5. import org.apache.spark.ml.linalg.Vector;
    6. import org.apache.spark.sql.Dataset;
    7. import org.apache.spark.sql.Row;
    8. import org.apache.spark.sql.SparkSession;
    9.  
    10. public class LogisticRegressionDemo {
    11.  
    12.     public static void main(String[] args) {
    13.         SparkSession spark = SparkSession
    14.                 .builder()
    15.                 .appName("LogisticRegressionDemo")
    16.                 .master("local[*]")
    17.                 .getOrCreate();
    18.  
    19.         // 加载数据
    20.         Dataset<Row> data = spark.read().format("libsvm").load("data/sample_libsvm_data.txt");
    21.  
    22.         // 将特征向量转换成一列
    23.         VectorAssembler assembler = new VectorAssembler()
    24.                 .setInputCols(new String[]{"features"})
    25.                 .setOutputCol("feature");
    26.  
    27.         Dataset<Row> newData = assembler.transform(data).select("label", "feature");
    28.  
    29.         // 将数据集分为训练集和测试集
    30.         Dataset<Row>[] splits = newData.randomSplit(new double[]{0.7, 0.3});
    31.         Dataset<Row> trainData = splits[0];
    32.         Dataset<Row> testData = splits[1];
    33.  
    34.         // 创建逻辑回归模型
    35.         LogisticRegression lr = new LogisticRegression();
    36.  
    37.         // 训练模型
    38.         LogisticRegressionModel lrModel = lr.fit(trainData);
    39.  
    40.         // 在测试集上进行预测
    41.         Dataset<Row> predictions = lrModel.transform(testData);
    42.  
    43.         // 计算模型评估指标
    44.         BinaryClassificationEvaluator evaluator = new BinaryClassificationEvaluator();
    45.         double auc = evaluator.evaluate(predictions);
    46.  
    47.         System.out.println("Area under ROC curve = " + auc);
    48.  
    49.         spark.stop();
    50.     }
    51. }

    这个示例代码首先加载了一个 libsvm 格式的数据集，然后将特征向量转换成一列，将数据集分为训练集和测试集，创建逻辑回归模型并训练模型，最后在测试集上进行预测并计算模型评估指标。在这个例子中，我们使用了 BinaryClassificationEvaluator 来计算模型的 AUC 指标，它是评估二元分类器性能的一种常用指标。

    需要注意的是，以上代码仅供参考，实际情况可能需要根据数据集的特点和任务的要求进行相应的修改。

    3.1.2 决策树

    Spark MLlib 分类决策树是一种基于树结构的分类算法，通过一系列特征对数据进行划分和分类。该算法在 Spark MLlib 中的实现采用 CART（Classification And Regression Tree）算法，使用信息熵或 Gini 系数等指标进行特征选择和划分。Spark MLlib 分类决策树可用于二分类、多分类和概率预测问题。

    1. import org.apache.spark.ml.Pipeline;
    2. import org.apache.spark.ml.PipelineModel;
    3. import org.apache.spark.ml.PipelineStage;
    4. import org.apache.spark.ml.classification.DecisionTreeClassificationModel;
    5. import org.apache.spark.ml.classification.DecisionTreeClassifier;
    6. import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
    7. import org.apache.spark.ml.feature.IndexToString;
    8. import org.apache.spark.ml.feature.StringIndexer;
    9. import org.apache.spark.ml.feature.StringIndexerModel;
    10. import org.apache.spark.ml.feature.VectorAssembler;
    11. import org.apache.spark.sql.Dataset;
    12. import org.apache.spark.sql.Row;
    13. import org.apache.spark.sql.SparkSession;
    14.  
    15. public class DecisionTreeClassificationExample {
    16.   public static void main(String[] args) {
    17.     SparkSession spark = SparkSession.builder()
    18.       .appName("DecisionTreeClassificationExample")
    19.       .master("local[*]")
    20.       .getOrCreate();
    21.  
    22.     // 读取数据集
    23.     Dataset<Row> data = spark.read().format("csv")
    24.       .option("header", "true")
    25.       .option("inferSchema", "true")
    26.       .load("path/to/data.csv");
    27.  
    28.     // 将标签列转换为数值类型
    29.     StringIndexerModel labelIndexer = new StringIndexer()
    30.       .setInputCol("label")
    31.       .setOutputCol("indexedLabel")
    32.       .fit(data);
    33.     data = labelIndexer.transform(data);
    34.  
    35.     // 将特征列转换为特征向量
    36.     VectorAssembler featureAssembler = new VectorAssembler()
    37.       .setInputCols(new String[]{"feature1", "feature2", "feature3"})
    38.       .setOutputCol("features");
    39.     data = featureAssembler.transform(data);
    40.  
    41.     // 将数据集分为训练集和测试集
    42.     Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3}, 12345);
    43.     Dataset<Row> trainData = splits[0];
    44.     Dataset<Row> testData = splits[1];
    45.  
    46.     // 创建决策树分类器
    47.     DecisionTreeClassifier dt = new DecisionTreeClassifier()
    48.       .setLabelCol("indexedLabel")
    49.       .setFeaturesCol("features");
    50.  
    51.     // 将标签数值转换回原始标签
    52.     IndexToString labelConverter = new IndexToString()
    53.       .setInputCol("prediction")
    54.       .setOutputCol("predictedLabel")
    55.       .setLabels(labelIndexer.labels());
    56.  
    57.     // 创建管道并拟合模型
    58.     Pipeline pipeline = new Pipeline()
    59.       .setStages(new PipelineStage[]{labelIndexer, featureAssembler, dt, labelConverter});
    60.     PipelineModel model = pipeline.fit(trainData);
    61.  
    62.     // 在测试集上进行预测和评估
    63.     Dataset<Row> predictions = model.transform(testData);
    64.     MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
    65.       .setLabelCol("indexedLabel")
    66.       .setPredictionCol("prediction")
    67.       .setMetricName("accuracy");
    68.     double accuracy = evaluator.evaluate(predictions);
    69.     System.out.println("Test Error = " + (1.0 - accuracy));
    70.     // 输出决策树结构
    71.     DecisionTreeClassificationModel treeModel =
    72.     (DecisionTreeClassificationModel) (model.stages()[2]);
    73.     System.out.println("Learned classification tree model:\n" + treeModel.toDebugString());
    74.  
    75.     spark.stop();
    76.     }
    77. }

    以上示例中，我们首先使用 SparkSession 读取 CSV 格式的数据集。然后，使用 StringIndexer 将标签列转换为数值类型，并使用 VectorAssembler 将特征列转换为特征向量。接着，将数据集分为训练集和测试集，并创建 DecisionTreeClassifier 决策树分类器。最后，将管道中的各个阶段组合在一起，拟合模型并在测试集上进行预测和评估。

    3.1.3 随机森林

    随机森林是一种集成学习算法，它将多棵决策树组合起来，通过投票或平均来决定分类结果。该算法在 Spark MLlib 中的实现使用基于 CART（Classification And Regression Tree）算法的决策树作为基分类器，可以用于二分类、多分类和概率预测问题。

    以下是一个基于 Java 的 Spark MLlib 分类随机森林示例：

    1. import org.apache.spark.ml.Pipeline;
    2. import org.apache.spark.ml.PipelineModel;
    3. import org.apache.spark.ml.PipelineStage;
    4. import org.apache.spark.ml.classification.RandomForestClassificationModel;
    5. import org.apache.spark.ml.classification.RandomForestClassifier;
    6. import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
    7. import org.apache.spark.ml.feature.IndexToString;
    8. import org.apache.spark.ml.feature.StringIndexer;
    9. import org.apache.spark.ml.feature.StringIndexerModel;
    10. import org.apache.spark.ml.feature.VectorAssembler;
    11. import org.apache.spark.sql.Dataset;
    12. import org.apache.spark.sql.Row;
    13. import org.apache.spark.sql.SparkSession;
    14.  
    15. public class RandomForestClassificationExample {
    16.   public static void main(String[] args) {
    17.     SparkSession spark = SparkSession.builder()
    18.       .appName("RandomForestClassificationExample")
    19.       .master("local[*]")
    20.       .getOrCreate();
    21.  
    22.     // 读取数据集
    23.     Dataset<Row> data = spark.read().format("csv")
    24.       .option("header", "true")
    25.       .option("inferSchema", "true")
    26.       .load("path/to/data.csv");
    27.  
    28.     // 将标签列转换为数值类型
    29.     StringIndexerModel labelIndexer = new StringIndexer()
    30.       .setInputCol("label")
    31.       .setOutputCol("indexedLabel")
    32.       .fit(data);
    33.     data = labelIndexer.transform(data);
    34.  
    35.     // 将特征列转换为特征向量
    36.     VectorAssembler featureAssembler = new VectorAssembler()
    37.       .setInputCols(new String[]{"feature1", "feature2", "feature3"})
    38.       .setOutputCol("features");
    39.     data = featureAssembler.transform(data);
    40.  
    41.     // 将数据集分为训练集和测试集
    42.     Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3}, 12345);
    43.     Dataset<Row> trainData = splits[0];
    44.     Dataset<Row> testData = splits[1];
    45.  
    46.     // 创建随机森林分类器
    47.     RandomForestClassifier rf = new RandomForestClassifier()
    48.       .setLabelCol("indexedLabel")
    49.       .setFeaturesCol("features")
    50.       .setNumTrees(10);
    51.  
    52.     // 将标签数值转换回原始标签
    53.     IndexToString labelConverter = new IndexToString()
    54.       .setInputCol("prediction")
    55.       .setOutputCol("predictedLabel")
    56.       .setLabels(labelIndexer.labels());
    57.  
    58.     // 创建管道并拟合模型
    59.     Pipeline pipeline = new Pipeline()
    60.       .setStages(new PipelineStage[]{labelIndexer, featureAssembler, rf, labelConverter});
    61.     PipelineModel model = pipeline.fit(trainData);
    62.  
    63.     // 在测试集上进行预测和评估
    64.     Dataset<Row> predictions = model.transform(testData);
    65.     MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
    66.       .setLabelCol("indexedLabel")
    67.       .setPredictionCol("prediction")
    68.       .setMetricName("accuracy");
    69.     double accuracy = evaluator.evaluate(predictions);
    70.     System.out.println("Test Error = " + (1.0 - accuracy));
    71.     // 获取训练好的随机森林模型并打印树的重要性
    72.     RandomForestClassificationModel rfModel = (RandomForestClassificationModel) model.stages()[2];
    73.     System.out.println("Learned classification forest model:\n" + rfModel.toDebugString());
    74.  
    75.     spark.stop();
    76.   }
    77. }

    该示例代码首先使用 SparkSession 读取 CSV 格式的数据集。接下来，使用 StringIndexer 将标签列转换为数值类型，并使用 VectorAssembler 将特征列转换为特征向量。然后，将数据集分为训练集和测试集。创建 RandomForestClassifier，并将其作为管道的一部分进行拟合。拟合后，使用 MulticlassClassificationEvaluator 对测试集进行预测和评估。最后，获取训练好的随机森林模型并打印树的重要性。

    请注意，上面的示例中，数据集的路径应该被替换为实际数据集的路径，特征列的名称也应该被替换为实际特征列的名称。

    3.1.4 梯度提升树

    Spark MLlib 提供了一个强大的算法——分类梯度提升树（Gradient-Boosted Trees, GBT），它可以用于二元分类和多类分类。GBT 是一种集成学习算法，它通过在先前树的残差上逐步拟合一系列决策树来提高模型的准确性。

    在 Spark MLlib 中，可以使用 ​​GBTClassifier​​ 类来构建分类 GBT 模型。GBT 分类器使用一系列决策树来逐步提高模型的准确性，每个决策树都是在之前决策树的残差上训练得到的。通过这种方式，GBT 可以在更少的迭代次数下得到比随机森林更准确的模型。

    与其他 Spark MLlib 分类器类似，GBT 分类器也使用管道（Pipeline）来处理数据。管道通常包括以下几个步骤：

    1. 数据预处理：包括数据清洗、特征提取、特征转换等操作。
    2. 特征工程：根据特定的特征工程需求，对特征进行过滤、选择、转换等操作。
    3. 模型训练：使用训练集对模型进行拟合。
    4. 模型评估：使用测试集对模型进行评估。
    5. 模型应用：将模型应用到新的数据集上进行预测。

    在使用 GBT 分类器时，你需要指定以下参数：

    • ​​featuresCol​​：特征列的名称。
    • ​​labelCol​​：标签列的名称。
    • ​​maxIter​​：训练迭代次数。
    • ​​maxDepth​​：决策树的最大深度。
    • ​​minInstancesPerNode​​：每个节点上的最小实例数。
    • ​​stepSize​​：每个迭代步骤的步长。
    • ​​subsamplingRate​​：用于训练每棵树的数据子样本的比例。

    1. import org.apache.spark.ml.Pipeline;
    2. import org.apache.spark.ml.PipelineModel;
    3. import org.apache.spark.ml.PipelineStage;
    4. import org.apache.spark.ml.classification.GBTClassificationModel;
    5. import org.apache.spark.ml.classification.GBTClassifier;
    6. import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
    7. import org.apache.spark.ml.feature.IndexToString;
    8. import org.apache.spark.ml.feature.StringIndexer;
    9. import org.apache.spark.ml.feature.StringIndexerModel;
    10. import org.apache.spark.ml.feature.VectorIndexer;
    11. import org.apache.spark.ml.feature.VectorIndexerModel;
    12. import org.apache.spark.sql.Dataset;
    13. import org.apache.spark.sql.Row;
    14. import org.apache.spark.sql.SparkSession;
    15.  
    16.  
    17. public class GBTExample {
    18.     public static void main(String[] args) {
    19.         // 创建一个 SparkSession
    20.         SparkSession spark = SparkSession
    21.                 .builder()
    22.                 .appName("GBTExample")
    23.                 .getOrCreate();
    24.  
    25.         // 读取数据集
    26.         Dataset<Row> data = spark.read()
    27.                 .format("libsvm")
    28.                 .load("data/mllib/sample_libsvm_data.txt");
    29.  
    30.         // 对标签列进行索引
    31.         StringIndexerModel labelIndexer = new StringIndexer()
    32.                 .setInputCol("label")
    33.                 .setOutputCol("indexedLabel")
    34.                 .fit(data);
    35.  
    36.         // 对特征列进行索引
    37.         VectorIndexerModel featureIndexer = new VectorIndexer()
    38.                 .setInputCol("features")
    39.                 .setOutputCol("indexedFeatures")
    40.                 .setMaxCategories(4) // 特征具有少于 4 个不同的值
    41.                 .fit(data);
    42.  
    43.         // 将数据集拆分为训练集和测试集
    44.         Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3});
    45.         Dataset<Row> trainingData = splits[0];
    46.         Dataset<Row> testData = splits[1];
    47.  
    48.         // 定义 GBT 分类器
    49.         GBTClassifier gbt = new GBTClassifier()
    50.                 .setLabelCol("indexedLabel")
    51.                 .setFeaturesCol("indexedFeatures")
    52.                 .setMaxIter(10)
    53.                 .setFeatureSubsetStrategy("auto");
    54.  
    55.         // 将索引的标签转换回原始标签
    56.         IndexToString labelConverter = new IndexToString()
    57.                 .setInputCol("prediction")
    58.                 .setOutputCol("predictedLabel")
    59.                 .setLabels(labelIndexer.labels());
    60.  
    61.         // 创建管道
    62.         Pipeline pipeline = new Pipeline()
    63.                 .setStages(new PipelineStage[]{
    64.                         labelIndexer,
    65.                         featureIndexer,
    66.                         gbt,
    67.                         labelConverter
    68.                 });
    69.  
    70.         // 训练模型
    71.         PipelineModel model = pipeline.fit(trainingData);
    72.  
    73.         // 进行预测
    74.         Dataset<Row> predictions = model.transform(testData);
    75.  
    76.         // 选择样例行显示
    77.         predictions.select("predictedLabel", "label", "features").show(5);
    78.  
    79.         // 评估模型
    80.         MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
    81.                 .setLabelCol("indexedLabel")
    82.                 .setPredictionCol("prediction")
    83.                 .setMetricName("accuracy");
    84.         double accuracy = evaluator.evaluate(predictions);
    85.         System.out.println("Test Error = " + (1.0 - accuracy));
    86.  
    87.         // 获取训练得到的 GBT 模型
    88.         GBTClassificationModel gbtModel = (GBTClassificationModel) (model.stages()[2]);
    89.         System.out.println("Learned classification GBT model:\n" + gbtModel.toDebugString());
    90.  
    91.         spark.stop();
    92.     }
    93. }

    该示例使用了 Spark MLlib 内置的 ​​sample_libsvm_data.txt​​​ 数据集。首先，将数据集加载到 ​​DataFrame​​ 中。接下来，对标签列和特征列进行索引。然后，将数据集拆分为训练集和测试集。接下来，创建 GBT 分类器，并使用管道将标签转换回原始标签。最后，使用训练数据拟合管道并进行预测。最终评估模型并输出模型学习到的 GBT 分类模型的调试字符串。该字符串显示了树的结构和分裂标准，以及在每个节点处对特征的使用情况和分裂点。

    3.2 回归问题

    3.2.1 线性回归

    Spark MLlib 的线性回归算法是一种广泛使用的预测模型。它将输入特征映射到连续的输出值。这是通过训练模型来确定最佳拟合线性函数的系数完成的。

    以下是一个使用 Spark MLlib 实现线性回归的 Java 示例程序。

    1. import org.apache.spark.ml.feature.VectorAssembler;
    2. import org.apache.spark.ml.regression.LinearRegression;
    3. import org.apache.spark.ml.regression.LinearRegressionModel;
    4. import org.apache.spark.ml.regression.LinearRegressionTrainingSummary;
    5. import org.apache.spark.sql.Dataset;
    6. import org.apache.spark.sql.Row;
    7. import org.apache.spark.sql.SparkSession;
    8.  
    9. public class LinearRegressionExample {
    10.  
    11.     public static void main(String[] args) {
    12.         // 创建一个 SparkSession
    13.         SparkSession spark = SparkSession
    14.                 .builder()
    15.                 .appName("LinearRegressionExample")
    16.                 .getOrCreate();
    17.  
    18.         // 读取数据集
    19.         Dataset<Row> data = spark.read()
    20.                 .format("libsvm")
    21.                 .load("data/mllib/sample_linear_regression_data.txt");
    22.  
    23.         // 将数据集拆分为训练集和测试集
    24.         Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3});
    25.         Dataset<Row> trainingData = splits[0];
    26.  
    27.         // 将特征列合并到一个向量列中
    28.         VectorAssembler assembler = new VectorAssembler()
    29.                 .setInputCols(new String[]{"features"})
    30.                 .setOutputCol("featuresVector");
    31.  
    32.         // 定义线性回归模型
    33.         LinearRegression lr = new LinearRegression()
    34.                 .setMaxIter(10)
    35.                 .setRegParam(0.3)
    36.                 .setElasticNetParam(0.8);
    37.  
    38. //        // 将数据集拟合到线性回归模型中
    39.         Dataset<Row> trainingDataWithFeatures = assembler.transform(trainingData);
    40.  
    41.         //训练模型
    42.         LinearRegressionModel lrModel = lr.fit(trainingDataWithFeatures);
    43.         //打印线性回归的系数和截距
    44.         System.out.println("系数Coefficients: "+lrModel.coefficients() + "");
    45.         System.out.println(" 截距Intercept: " + lrModel.intercept()+ "");
    46.         //总结训练集上的模型并打印出一些指标。
    47.         LinearRegressionTrainingSummary trainingSummary = lrModel.summary();
    48.         Dataset<Row> dataset = trainingSummary.predictions().select("prediction", "label", "featuresVector");
    49.         dataset.show(5);
    50.         spark.stop();
    51.     }
    52. }
    3.2.2 决策树回归

    Spark MLlib 提供了决策树回归（Decision Tree Regression）算法来解决回归问题。决策树回归是一种基于树结构的非参数统计方法，能够处理多维输入和输出，并且具有良好的可解释性和鲁棒性。

    下面是一个简单的 Java 代码示例，演示如何使用 Spark MLlib 的决策树回归算法对数据进行训练和预测：

    1. import org.apache.spark.ml.Pipeline;
    2. import org.apache.spark.ml.PipelineModel;
    3. import org.apache.spark.ml.PipelineStage;
    4. import org.apache.spark.ml.feature.VectorAssembler;
    5. import org.apache.spark.ml.regression.DecisionTreeRegressionModel;
    6. import org.apache.spark.ml.regression.DecisionTreeRegressor;
    7. import org.apache.spark.sql.Dataset;
    8. import org.apache.spark.sql.Row;
    9. import org.apache.spark.sql.SparkSession;
    10. public class DecisionTreeRegressionExample {
    11.     public static void main(String[] args) {
    12.         // 创建 SparkSession
    13.         SparkSession spark = SparkSession.builder()
    14.                 .appName("DecisionTreeRegressionExample")
    15.                 .master("local[*]")
    16.                 .getOrCreate();
    17.  
    18.         // 读取数据集
    19.         Dataset<Row> data = spark.read().format("csv")
    20.                 .option("header", "true")
    21.                 .option("inferSchema", "true")
    22.                 .load("path/to/your/data.csv");
    23.  
    24.         // 定义特征列和标签列
    25.         String[] featureCols = data.columns();
    26.         String labelCol = "label";
    27.  
    28.         // 将特征列转换为向量
    29.         VectorAssembler assembler = new VectorAssembler()
    30.                 .setInputCols(featureCols)
    31.                 .setOutputCol("features");
    32.  
    33.         Dataset<Row> dataWithFeatures = assembler.transform(data).select("features", labelCol);
    34.  
    35.         // 将数据集拆分为训练集和测试集
    36.         double[] weights = {0.7, 0.3};
    37.         Dataset<Row>[] datasets = dataWithFeatures.randomSplit(weights);
    38.         Dataset<Row> trainData = datasets[0];
    39.         Dataset<Row> testData = datasets[1];
    40.  
    41.         // 创建决策树回归器
    42.         DecisionTreeRegressor dt = new DecisionTreeRegressor()
    43.                 .setLabelCol(labelCol)
    44.                 .setFeaturesCol("features");
    45.  
    46.         // 创建 Pipeline
    47.         Pipeline pipeline = new Pipeline().setStages(new PipelineStage[] { dt });
    48.  
    49.         // 训练模型
    50.         PipelineModel model = pipeline.fit(trainData);
    51.  
    52.         // 预测测试集
    53.         Dataset<Row> predictions = model.transform(testData);
    54.  
    55.         // 显示预测结果
    56.         predictions.show();
    57.  
    58.         // 获取训练好的决策树模型
    59.         DecisionTreeRegressionModel dtModel = (DecisionTreeRegressionModel) model.stages()[0];
    60.         System.out.println("Learned regression tree model:\n" + dtModel.toDebugString());
    61.  
    62.         // 停止 SparkSession
    63.         spark.stop();
    64.     }
    65. }

    这只是一个简单的示例，实际应用中可能需要更复杂的特征工程和模型调整。另外，如果数据集过大，可能需要在集群上运行以获得更好的性能。

    3.2.3 随机森林回归

    Spark MLlib提供了随机森林回归算法，可以用于预测连续的数值型数据。随机森林是一种集成学习算法，它基于决策树，通过随机选择样本和特征来减少过拟合的风险。随机森林回归使用多个决策树对数据进行拟合和预测，并取这些决策树的平均值作为最终预测结果。本文将介绍如何使用Spark MLlib中的随机森林回归算法，并提供一个完整可运行的Java示例。

    示例说明： 在这个示例中，我们将使用Spark MLlib中的随机森林回归算法，对一组汽车数据进行建模，然后使用模型来预测汽车的燃油效率（MPG）。我们将使用UCI Machine Learning Repository中的Auto MPG数据集。该数据集包含8个输入特征，如汽车的气缸数、排量、马力、重量等，以及一个输出特征MPG，表示汽车的燃油效率。我们将使用70%的数据来训练模型，30%的数据用于测试模型性能。

    1.准备数据 我们需要下载Auto MPG数据集，将其保存为CSV文件，并将其加载到Spark DataFrame中。

    数据集下载地址：​​https://archive.ics.uci.edu/ml/datasets/auto+mpg​​

    CSV文件格式如下：

    1. mpg,cylinders,displacement,horsepower,weight,acceleration,modelyear,origin
    2. 18.0,8,307.0,130.0,3504.0,12.0,70,1
    3. 15.0,8,350.0,165.0,3693.0,11.5,70,1

    其中，第一列为输出特征MPG，后面的列为输入特征。

    2.构建随机森林回归模型

    1. // 创建随机森林回归模型
    2. RandomForestRegressor rf = new RandomForestRegressor()
    3.         .setLabelCol("label")
    4.         .setFeaturesCol("features")
    5.         .setNumTrees(10);
    6.  
    7. // 训练模型
    8. RandomForestRegressionModel model = rf.fit(trainingData);

    3.使用模型进行预测

    1. // 使用模型进行预测
    2. Dataset<Row> predictions = model.transform(testData);

    4.评估模型性能

    1. // 评估模型性能
    2. RegressionEvaluator evaluator = new RegressionEvaluator()
    3.         .setLabelCol("label")
    4.         .setPredictionCol("prediction")
    5.         .setMetricName("rmse");
    6. double rmse = evaluator.evaluate(predictions);

    下面是完整 Java 代码示例：

    1. import org.apache.spark.SparkConf;
    2. import org.apache.spark.ml.Pipeline;
    3. import org.apache.spark.ml.evaluation.RegressionEvaluator;
    4. import org.apache.spark.ml.feature.VectorAssembler;
    5. import org.apache.spark.ml.regression.RandomForestRegressor;
    6. import org.apache.spark.ml.tuning.CrossValidator;
    7. import org.apache.spark.ml.tuning.CrossValidatorModel;
    8. import org.apache.spark.ml.tuning.ParamGridBuilder;
    9. import org.apache.spark.sql.Dataset;
    10. import org.apache.spark.sql.Row;
    11. import org.apache.spark.sql.SparkSession;
    12. public class RandomForestRegressionExample {
    13.     public static void main(String[] args) {
    14.  
    15.         // Create a Spark session
    16.         SparkConf conf = new SparkConf().setAppName("RandomForestRegressionExample").setMaster("local[*]");
    17.         SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
    18.  
    19.         // Load data
    20.         Dataset<Row> data = spark.read().format("libsvm").load("data/sample_libsvm_data.txt");
    21.  
    22.         // Split the data into training and test sets
    23.         Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3});
    24.         Dataset<Row> trainingData = splits[0];
    25.         Dataset<Row> testData = splits[1];
    26.  
    27.         // Define the feature column names
    28.         String[] featureCols = new String[data.schema().fieldNames().length - 1];
    29.         for (int i = 0; i < featureCols.length; i++) {
    30.             featureCols[i] = "feature" + (i + 1);
    31.         }
    32.  
    33.         // Assemble features into a vector
    34.         VectorAssembler assembler = new VectorAssembler()
    35.                 .setInputCols(featureCols)
    36.                 .setOutputCol("features");
    37.  
    38.         Dataset<Row> trainingDataWithFeatures = assembler.transform(trainingData);
    39.         Dataset<Row> testDataWithFeatures = assembler.transform(testData);
    40.  
    41.         // Create a RandomForestRegressor model
    42.         RandomForestRegressor rf = new RandomForestRegressor()
    43.                 .setLabelCol("label")
    44.                 .setFeaturesCol("features")
    45.                 .setMaxDepth(5)
    46.                 .setNumTrees(20);
    47.  
    48.         // Set up a pipeline
    49.         Pipeline pipeline = new Pipeline().setStages(new RandomForestRegressor[]{rf});
    50.  
    51.         // Set up a grid of hyperparameters to search over using 3-fold cross validation
    52.         ParamGridBuilder paramGridBuilder = new ParamGridBuilder()
    53.                 .addGrid(rf.maxDepth(), new int[]{5, 10})
    54.                 .addGrid(rf.numTrees(), new int[]{20, 50});
    55.         CrossValidator crossValidator = new CrossValidator()
    56.                 .setEstimator(pipeline)
    57.                 .setEvaluator(new RegressionEvaluator())
    58.                 .setEstimatorParamMaps(paramGridBuilder.build())
    59.                 .setNumFolds(3);
    60.  
    61.         // Train the model using cross-validation
    62.         crossValidator.setSeed(12345);
    63.         CrossValidatorModel crossValidatorModel = crossValidator.fit(trainingDataWithFeatures);
    64.  
    65.         // Evaluate the model on the test set
    66.         Dataset<Row> predictions = crossValidatorModel.transform(testDataWithFeatures);
    67.         RegressionEvaluator evaluator = new RegressionEvaluator()
    68.                 .setLabelCol("label")
    69.                 .setPredictionCol("prediction")
    70.                 .setMetricName("rmse");
    71.         double rmse = evaluator.evaluate(predictions);
    72.         System.out.println("Root Mean Squared Error (RMSE) on test data = " + rmse);
    73.  
    74.         // Stop the Spark session
    75.         spark.stop();
    76.     }
    77. }
    3.2.4 梯度提升回归树

    Spark MLlib提供了梯度提升回归树（Gradient-Boosted Trees，GBT）的算法，它是一种强大的回归模型，可以用于连续的数值预测。GBT在每一次迭代中，使用决策树模型去拟合残差值，然后将所有的模型的预测结果相加，得到最终的预测结果。

    以下是一个使用Spark MLlib进行梯度提升回归树的示例Java程序。这个程序将使用一个数据集，该数据集包含了关于自行车租赁量的信息。它将使用梯度提升回归树来预测一天中自行车的租赁量。

    1. import org.apache.spark.ml.evaluation.RegressionEvaluator;
    2. import org.apache.spark.ml.feature.VectorAssembler;
    3. import org.apache.spark.ml.regression.GBTRegressionModel;
    4. import org.apache.spark.ml.regression.GBTRegressor;
    5. import org.apache.spark.sql.*;
    6.  
    7. public class GBTRegressionDemo {
    8.     public static void main(String[] args) {
    9.  
    10.         // 创建 SparkSession
    11.         SparkSession spark = SparkSession.builder()
    12.                 .appName("GradientBoostedTreeRegressionDemo")
    13.                 .master("local[*]")
    14.                 .getOrCreate();
    15.  
    16.         // 读取数据集
    17.         Dataset<Row> data = spark.read().format("libsvm")
    18.                 .load("data/sample_libsvm_data.txt");
    19.  
    20.         // 将数据集划分为训练集和测试集
    21.         Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3});
    22.         Dataset<Row> trainingData = splits[0];
    23.         Dataset<Row> testData = splits[1];
    24.  
    25.         // 将特征向量合并为一个向量
    26.         VectorAssembler assembler = new VectorAssembler()
    27.                 .setInputCols(trainingData.columns())
    28.                 .setOutputCol("features");
    29.         Dataset<Row> trainingDataWithFeatures = assembler.transform(trainingData);
    30.         Dataset<Row> testDataWithFeatures = assembler.transform(testData);
    31.  
    32.         // 创建梯度提升回归树模型
    33.         GBTRegressor gbt = new GBTRegressor()
    34.                 .setLabelCol("label")
    35.                 .setFeaturesCol("features")
    36.                 .setMaxIter(10);
    37.  
    38.         // 训练模型
    39.         GBTRegressionModel model = gbt.fit(trainingDataWithFeatures);
    40.  
    41.         // 在测试集上进行预测
    42.         Dataset<Row> predictions = model.transform(testDataWithFeatures);
    43.  
    44.         // 评估模型
    45.         RegressionEvaluator evaluator = new RegressionEvaluator()
    46.                 .setLabelCol("label")
    47.                 .setPredictionCol("prediction")
    48.                 .setMetricName("rmse");
    49.         double rmse = evaluator.evaluate(predictions);
    50.         System.out.println("Root Mean Squared Error (RMSE) on test data = " + rmse);
    51.  
    52.         // 输出模型的节点信息
    53.         System.out.println("Learned regression GBT model:\n" + model.toDebugString());
    54.  
    55.         // 关闭 SparkSession
    56.         spark.close();
    57.     }
    58. }

    该程序与之前的程序类似，只是将 ​​DecisionTreeRegressor​​​ 类替换为 ​​GBTRegressor​​​ 类。需要注意的是，​​GBTRegressor​​​ 类在设置参数时需要设置 ​​maxIter​​​ 参数，表示最大迭代次数。同样需要用 ​​setFeaturesCol​​​ 方法设置特征列，用 ​​setLabelCol​​​ 方法设置标签列。最后需要调用 ​​fit​​​ 方法训练模型，然后用 ​​transform​​​ 方法在测试集上进行预测。最后用 ​​RegressionEvaluator​​ 类评估模型，并输出模型的节点信息。

    关于无监督学习，模型评估和调优，模型部署的内容请看下篇《Java应用|使用Apache Spark MLlib构建机器学习模型【下】》。

    如果您觉得本博客的内容对您有所帮助或启发，请关注我的博客，以便第一时间获取最新技术文章和教程。同时，也欢迎您在评论区留言，分享想法和建议。谢谢支持！

  • 相关阅读:
    WPF-UI布局
    计算机竞赛 深度学习+python+opencv实现动物识别 - 图像识别
    中国纺织服装市场深度全景调研及投资前景分析报告
    offer突击训练营，给你一个offer的保障，求职跳槽的看过来！
    oracle_申明与赋值
    kafka下载|安装
    Spring 6【方法参数校验、SpingAOP介绍、Schema-based方式实现AOP 】(十四)-全面详解（学习总结---从入门到深化）
    Centos7.6安装FTP
    centos7中python3.10找不到openssl解决方案
    在Linux服务器上部署Tornado项目
  • 原文地址：https://blog.csdn.net/bairo007/article/details/133066868