Apache SparkMLlib构建机器学习分类模型
刘润森
共 27447字，需浏览 55分钟
·
2024-04-11 14:41
一、引言
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
- 处理大规模数据：Spark MLlib支持分布式计算和大规模数据处理，使得处理大规模数据集变得容易。1. 丰富的算法库：Spark MLlib包含了许多经典的机器学习算法和工具，如分类、回归、聚类、协同过滤、特征提取和数据预处理等，覆盖了大部分机器学习应用场景。1. 高性能：Spark MLlib基于Apache Spark，使用内存计算和RDD（弹性分布式数据集）等优化技术，可以在处理大规模数据时提供高性能和可扩展性。1. 易于使用：Spark MLlib提供了一个易于使用的高级API，使得用户可以快速构建、训练和调整机器学习模型，而无需担心底层分布式计算的复杂性。1. 多语言支持：Spark MLlib支持多种编程语言的API，包括Python、Java和Scala等，使得不同的开发人员可以使用他们最喜欢的编程语言来开发机器学习应用程序。
二、Spark MLlib基础
2.1 RDD和DataFrame的比较
- 数据类型：基础RDD可以包含任意类型的数据，包括对象、原始类型、数组和集合等；DataFrame则是一种表格化的数据结构，其数据类型必须是统一的，且可以使用SQL-like的语法进行查询。1. 内存计算：DataFrame利用内存计算技术，相比基础RDD更加高效。1. 可读性：DataFrame比基础RDD更加易于阅读和理解，可以使用SQL-like的语法进行查询，更加直观。1. 类型安全：DataFrame是类型安全的，可以在编译期间捕获类型错误，避免运行时错误；而基础RDD则是类型不安全的，需要在运行时进行类型检查。1. 执行计划：基础RDD提供了更加灵活的执行计划，用户可以控制计算的方式和顺序，但这也增加了开发复杂度；而DataFrame则有一个自动优化的执行计划，可以自动优化查询性能。 总之，基础RDD更加灵活和可控，但需要开发人员自己掌握计算的方式和顺序；而DataFrame则更加易于使用和高效，适合快速开发和迭代。选择使用哪种数据结构，取决于具体的场景和需求。
2.2 数据准备和预处理
在使用Spark MLlib进行机器学习之前，需要对原始数据进行预处理和准备。以下是一些常见的数据准备和预处理步骤：
- 数据清洗：删除缺失值、处理异常值和重复值等。1. 特征选择：选择对模型有用的特征，去除冗余和无关的特征。1. 特征缩放：对特征进行缩放，以便它们具有相似的范围和重要性。1. 特征变换：将原始特征转换为更有意义的特征，如使用对数、指数、平方根等函数进行变换。1. 特征归一化：将特征值归一化为标准正态分布，使得模型更容易学习。1. 数据转换：将数据转换为适合模型训练的格式，如将分类变量转换为二进制变量、将文本转换为向量等。 在Spark MLlib中，可以使用各种预处理和数据准备工具，如：
- Imputer：用于填充缺失值。1. StandardScaler：用于特征缩放和归一化。1. VectorAssembler：用于将多个特征列组合成一个向量列。1. OneHotEncoder：用于将分类变量转换为二进制变量。1. StringIndexer和IndexToString：用于将字符串类型的变量转换为数字类型的变量。1. Tokenizer和StopWordsRemover：用于将文本转换为向量。 总之，在使用Spark MLlib进行机器学习之前，需要对原始数据进行预处理和准备。Spark MLlib提供了许多工具和功能，可以帮助我们轻松地完成这些任务。
2.3 特征提取和转换
在Spark MLlib中，有许多常用的特征提取和转换工具，包括：
- Tokenizer：用于将文本转换为单词或词条。1. StopWordsRemover：用于去除文本中的停用词，如“the”、“and”等。1. CountVectorizer：用于将文本转换为词频向量。1. HashingTF：用于将文本转换为哈希向量，可以减少维度并提高计算效率。1. IDF：用于计算逆文档频率，可以减少常见词语的权重，提高稀有词语的权重。1. Word2Vec：用于将文本转换为向量，可以捕捉词语之间的语义关系。1. PCA：用于将高维特征空间降维，可以提高计算效率并避免过拟合。1. StringIndexer：用于将分类变量转换为数字类型的变量。1. OneHotEncoder：用于将数字类型的变量转换为二进制变量。 以上这些工具都可以用于特征提取和转换，帮助我们将原始数据转换为模型可以处理的格式。我们可以根据具体的任务和数据类型选择适当的工具，以获得更好的结果。值得注意的是，这些工具的使用通常需要进行适当的参数设置和调整，以达到最佳的效果。
三、监督学习
3.1 分类问题
3.1.1 逻辑回归
逻辑回归是一种二元分类模型，它的目标是根据已知数据对一个事物进行分类。逻辑回归的输出是一个概率值，代表该事物属于某个类别的概率。如果概率值大于阈值，则将其分类为正类，否则分类为负类。
在 Spark MLlib 中，可以使用 LogisticRegression 类来实现逻辑回归。下面是一个 Java 版本的示例代码：
pom引用：
<dependencies>
<!-- Spark core dependencies -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-mllib_2.12</artifactId>
<version>3.2.0</version>
</dependency>
<!-- Spark testing dependencies (optional) -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId>
<version>3.2.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>3.2.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql-kafka-0-10_2.12</artifactId>
<version>3.2.0</version>
<scope>test</scope>
</dependency>
</dependencies>
import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.classification.LogisticRegressionModel;
import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator;
import org.apache.spark.ml.feature.VectorAssembler;
import org.apache.spark.ml.linalg.Vector;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
public class LogisticRegressionDemo {
public static void main(String[] args) {
SparkSession spark = SparkSession
.builder()
.appName("LogisticRegressionDemo")
.master("local[*]")
.getOrCreate();
// 加载数据
Dataset<Row> data = spark.read().format("libsvm").load("data/sample_libsvm_data.txt");
// 将特征向量转换成一列
VectorAssembler assembler = new VectorAssembler()
.setInputCols(new String[]{"features"})
.setOutputCol("feature");
Dataset<Row> newData = assembler.transform(data).select("label", "feature");
// 将数据集分为训练集和测试集
Dataset<Row>[] splits = newData.randomSplit(new double[]{0.7, 0.3});
Dataset<Row> trainData = splits[0];
Dataset<Row> testData = splits[1];
// 创建逻辑回归模型
LogisticRegression lr = new LogisticRegression();
// 训练模型
LogisticRegressionModel lrModel = lr.fit(trainData);
// 在测试集上进行预测
Dataset<Row> predictions = lrModel.transform(testData);
// 计算模型评估指标
BinaryClassificationEvaluator evaluator = new BinaryClassificationEvaluator();
double auc = evaluator.evaluate(predictions);
System.out.println("Area under ROC curve = " + auc);
spark.stop();
}
}
这个示例代码首先加载了一个 libsvm 格式的数据集，然后将特征向量转换成一列，将数据集分为训练集和测试集，创建逻辑回归模型并训练模型，最后在测试集上进行预测并计算模型评估指标。在这个例子中，我们使用了 BinaryClassificationEvaluator 来计算模型的 AUC 指标，它是评估二元分类器性能的一种常用指标。
需要注意的是，以上代码仅供参考，实际情况可能需要根据数据集的特点和任务的要求进行相应的修改。
3.1.2 决策树
Spark MLlib 分类决策树是一种基于树结构的分类算法，通过一系列特征对数据进行划分和分类。该算法在 Spark MLlib 中的实现采用 CART（Classification And Regression Tree）算法，使用信息熵或 Gini 系数等指标进行特征选择和划分。Spark MLlib 分类决策树可用于二分类、多分类和概率预测问题。
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.DecisionTreeClassificationModel;
import org.apache.spark.ml.classification.DecisionTreeClassifier;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.ml.feature.IndexToString;
import org.apache.spark.ml.feature.StringIndexer;
import org.apache.spark.ml.feature.StringIndexerModel;
import org.apache.spark.ml.feature.VectorAssembler;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
public class DecisionTreeClassificationExample {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder()
.appName("DecisionTreeClassificationExample")
.master("local[*]")
.getOrCreate();
// 读取数据集
Dataset<Row> data = spark.read().format("csv")
.option("header", "true")
.option("inferSchema", "true")
.load("path/to/data.csv");
// 将标签列转换为数值类型
StringIndexerModel labelIndexer = new StringIndexer()
.setInputCol("label")
.setOutputCol("indexedLabel")
.fit(data);
data = labelIndexer.transform(data);
// 将特征列转换为特征向量
VectorAssembler featureAssembler = new VectorAssembler()
.setInputCols(new String[]{"feature1", "feature2", "feature3"})
.setOutputCol("features");
data = featureAssembler.transform(data);
// 将数据集分为训练集和测试集
Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3}, 12345);
Dataset<Row> trainData = splits[0];
Dataset<Row> testData = splits[1];
// 创建决策树分类器
DecisionTreeClassifier dt = new DecisionTreeClassifier()
.setLabelCol("indexedLabel")
.setFeaturesCol("features");
// 将标签数值转换回原始标签
IndexToString labelConverter = new IndexToString()
.setInputCol("prediction")
.setOutputCol("predictedLabel")
.setLabels(labelIndexer.labels());
// 创建管道并拟合模型
Pipeline pipeline = new Pipeline()
.setStages(new PipelineStage[]{labelIndexer, featureAssembler, dt, labelConverter});
PipelineModel model = pipeline.fit(trainData);
// 在测试集上进行预测和评估
Dataset<Row> predictions = model.transform(testData);
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
.setLabelCol("indexedLabel")
.setPredictionCol("prediction")
.setMetricName("accuracy");
double accuracy = evaluator.evaluate(predictions);
System.out.println("Test Error = " + (1.0 - accuracy));
// 输出决策树结构
DecisionTreeClassificationModel treeModel =
(DecisionTreeClassificationModel) (model.stages()[2]);
System.out.println("Learned classification tree model:\n" + treeModel.toDebugString());
spark.stop();
}
}
以上示例中，我们首先使用 SparkSession 读取 CSV 格式的数据集。然后，使用 StringIndexer 将标签列转换为数值类型，并使用 VectorAssembler 将特征列转换为特征向量。接着，将数据集分为训练集和测试集，并创建 DecisionTreeClassifier 决策树分类器。最后，将管道中的各个阶段组合在一起，拟合模型并在测试集上进行预测和评估。
3.1.3 随机森林
随机森林是一种集成学习算法，它将多棵决策树组合起来，通过投票或平均来决定分类结果。该算法在 Spark MLlib 中的实现使用基于 CART（Classification And Regression Tree）算法的决策树作为基分类器，可以用于二分类、多分类和概率预测问题。
以下是一个基于 Java 的 Spark MLlib 分类随机森林示例：
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.RandomForestClassificationModel;
import org.apache.spark.ml.classification.RandomForestClassifier;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.ml.feature.IndexToString;
import org.apache.spark.ml.feature.StringIndexer;
import org.apache.spark.ml.feature.StringIndexerModel;
import org.apache.spark.ml.feature.VectorAssembler;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
public class RandomForestClassificationExample {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder()
.appName("RandomForestClassificationExample")
.master("local[*]")
.getOrCreate();
// 读取数据集
Dataset<Row> data = spark.read().format("csv")
.option("header", "true")
.option("inferSchema", "true")
.load("path/to/data.csv");
// 将标签列转换为数值类型
StringIndexerModel labelIndexer = new StringIndexer()
.setInputCol("label")
.setOutputCol("indexedLabel")
.fit(data);
data = labelIndexer.transform(data);
// 将特征列转换为特征向量
VectorAssembler featureAssembler = new VectorAssembler()
.setInputCols(new String[]{"feature1", "feature2", "feature3"})
.setOutputCol("features");
data = featureAssembler.transform(data);
// 将数据集分为训练集和测试集
Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3}, 12345);
Dataset<Row> trainData = splits[0];
Dataset<Row> testData = splits[1];
// 创建随机森林分类器
RandomForestClassifier rf = new RandomForestClassifier()
.setLabelCol("indexedLabel")
.setFeaturesCol("features")
.setNumTrees(10);
// 将标签数值转换回原始标签
IndexToString labelConverter = new IndexToString()
.setInputCol("prediction")
.setOutputCol("predictedLabel")
.setLabels(labelIndexer.labels());
// 创建管道并拟合模型
Pipeline pipeline = new Pipeline()
.setStages(new PipelineStage[]{labelIndexer, featureAssembler, rf, labelConverter});
PipelineModel model = pipeline.fit(trainData);
// 在测试集上进行预测和评估
Dataset<Row> predictions = model.transform(testData);
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
.setLabelCol("indexedLabel")
.setPredictionCol("prediction")
.setMetricName("accuracy");
double accuracy = evaluator.evaluate(predictions);
System.out.println("Test Error = " + (1.0 - accuracy));
// 获取训练好的随机森林模型并打印树的重要性
RandomForestClassificationModel rfModel = (RandomForestClassificationModel) model.stages()[2];
System.out.println("Learned classification forest model:\n" + rfModel.toDebugString());
spark.stop();
}
}
该示例代码首先使用 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. 特征工程：根据特定的特征工程需求，对特征进行过滤、选择、转换等操作。1. 模型训练：使用训练集对模型进行拟合。1. 模型评估：使用测试集对模型进行评估。1. 模型应用：将模型应用到新的数据集上进行预测。 在使用 GBT 分类器时，你需要指定以下参数：
-
featuresCol：特征列的名称。-labelCol：标签列的名称。-maxIter：训练迭代次数。-maxDepth：决策树的最大深度。-minInstancesPerNode：每个节点上的最小实例数。-stepSize：每个迭代步骤的步长。-subsamplingRate：用于训练每棵树的数据子样本的比例。
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.GBTClassificationModel;
import org.apache.spark.ml.classification.GBTClassifier;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.ml.feature.IndexToString;
import org.apache.spark.ml.feature.StringIndexer;
import org.apache.spark.ml.feature.StringIndexerModel;
import org.apache.spark.ml.feature.VectorIndexer;
import org.apache.spark.ml.feature.VectorIndexerModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
public class GBTExample {
public static void main(String[] args) {
// 创建一个 SparkSession
SparkSession spark = SparkSession
.builder()
.appName("GBTExample")
.getOrCreate();
// 读取数据集
Dataset<Row> data = spark.read()
.format("libsvm")
.load("data/mllib/sample_libsvm_data.txt");
// 对标签列进行索引
StringIndexerModel labelIndexer = new StringIndexer()
.setInputCol("label")
.setOutputCol("indexedLabel")
.fit(data);
// 对特征列进行索引
VectorIndexerModel featureIndexer = new VectorIndexer()
.setInputCol("features")
.setOutputCol("indexedFeatures")
.setMaxCategories(4) // 特征具有少于 4 个不同的值
.fit(data);
// 将数据集拆分为训练集和测试集
Dataset<Row>[] splits = data.randomSplit(new double[]{0.7, 0.3});
Dataset<Row> trainingData = splits[0];
Dataset<Row> testData = splits[1];
// 定义 GBT 分类器
GBTClassifier gbt = new GBTClassifier()
.setLabelCol("indexedLabel")
.setFeaturesCol("indexedFeatures")
.setMaxIter(10)
.setFeatureSubsetStrategy("auto");
// 将索引的标签转换回原始标签
IndexToString labelConverter = new IndexToString()
.setInputCol("prediction")
.setOutputCol("predictedLabel")
.setLabels(labelIndexer.labels());
// 创建管道
Pipeline pipeline = new Pipeline()
.setStages(new PipelineStage[]{
labelIndexer,
featureIndexer,
gbt,
labelConverter
});
// 训练模型
PipelineModel model = pipeline.fit(trainingData);
// 进行预测
Dataset<Row> predictions = model.transform(testData);
// 选择样例行显示
predictions.select("predictedLabel", "label", "features").show(5);
// 评估模型
MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
.setLabelCol("indexedLabel")
.setPredictionCol("prediction")
.setMetricName("accuracy");
double accuracy = evaluator.evaluate(predictions);
System.out.println("Test Error = " + (1.0 - accuracy));
// 获取训练得到的 GBT 模型
GBTClassificationModel gbtModel = (GBTClassificationModel) (model.stages()[2]);
System.out.println("Learned classification GBT model:\n" + gbtModel.toDebugString());
spark.stop();
}
}
该示例使用了 Spark MLlib 内置的 sample_libsvm_data.txt 数据集。首先，将数据集加载到 DataFrame 中。接下来，对标签列和特征列进行索引。然后，将数据集拆分为训练集和测试集。接下来，创建 GBT 分类器，并使用管道将标签转换回原始标签。最后，使用训练数据拟合管道并进行预测。最终评估模型并输出模型学习到的 GBT 分类模型的调试字符串。该字符串显示了树的结构和分裂标准，以及在每个节点处对特征的使用情况和分裂点。
评论