Pyspark:ml.Classification包的使用

  本篇博客主要介绍pyspark.ml中Classification、Clustering、Regression包的使用。

1. Classification

1.1 概览

  ml.Classification包中包含的分类算法及其相关类主要如下：

ml.Classification包中的类主要有以下几种：

• 基础类：可以指定分类算法的各项参数；
• Model类：通过基础类的fit()方法等到；
• Summary类：通过Model对象.evaluate(data)得到，其中data为测试集。通过该类既可以对测试集进行预测，也可以获得分类算法的各项评估指标；
• TrainingSummary类：通过Model对象.summary。可以通过该类获得算法在训练集上的各项评估指标；

1.2 使用案例

  本篇仅以逻辑回归为例对ml.Classification中的分类算法的使用进行说明。本案例中使用的是Iris鸢尾花数据集，可以从网上自行下载。

1.2.1 读取数据集

from pyspark.sql import SparkSession
import os
from pyspark.ml.feature import *
from pyspark.ml.classification import *
os.environ['SPARK_HOME'] ='/Users/sherry/documents/spark/spark-3.2.1-bin-hadoop3.2'
spark=SparkSession.builder.appName('ml').getOrCreate()

#读取数据集和测试集
trainSet=spark.read.csv(r'/Users/sherry/Downloads/iris_training.csv',
                        schema='Feat1 FLOAT,Feat2 FLOAT,Feat3 FLOAT,Feat4 FLOAT,label INT')
testSet=spark.read.csv(r'/Users/sherry/Downloads/iris_test.csv',
                       schema='Feat1 FLOAT,Feat2 FLOAT,Feat3 FLOAT,Feat4 FLOAT,label INT')
print('训练集数量:{},测试集数量:{}'.format(trainSet.count(),testSet.count()))
trainSet.show(5)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

其结果如下：

1.2.2 模型训练

将特征组合成向量

vect_assemble=VectorAssembler(inputCols=['Feat1','Feat2','Feat3','Feat4'],
                              outputCol='features')
trainSet=vect_assemble.transform(trainSet)
testSet=vect_assemble.transform(testSet)
1
2
3
4

和模型训练

#训练
LR=LogisticRegression(featuresCol='features',
                      labelCol='label',
                      predictionCol='prediction')
LR_model=LR.fit(trainSet)
trainSet=LR_model.transform(trainSet)
trainSet.printSchema()
1
2
3
4
5
6
7

训练集trainSet经过训练拟合之后默认情况会新增三个列，具体如下：

• rawPrediction: 每个样本在各个类别上的线性回归结果；
• probability: 每个样本属于各个类的概率；
• prediction: 每个样本预测的最终类别；
  

查看模型参数
  经过训练后的Model类即可以获得相关参数，可通过以下方法获得：

iris数据集是个三分类数据集，所以想要获得模型参数及截距需要通过coefficientMatrix和interceptVector。具体举例如下：

coefficient=LR_model.coefficientMatrix
intercept=LR_model.interceptVector
print(coefficient)
print(intercept)
1
2
3
4

其结果如下：

多分类逻辑回归采用OneVsRest模式训练，所以会有三个线性回归模型(可以参考：。这里仅以第一组结果为例进行说明：

coefficient=LR_model.coefficientMatrix
intercept=LR_model.interceptVector

getVectItem=func.udf(lambda x:x.toArray().tolist()[0],FloatType())
y=trainSet.select('features','rawPrediction',
    getVectItem('rawPrediction').alias('y'))
get_y=func.udf(lambda x:float(x.dot(coefficient.toArray()[0])+intercept.toArray()[0])),
               FloatType())
y=y.select('features','y',get_y('features').alias('y1'))
y.select('y','y1').show()
1
2
3
4
5
6
7
8
9
10

其结果如下：

模型评估指标
  Summary类和TrainingSummary类提供的评估指标类似，所以这里仅以Summary类为例进行介绍。该类提供的属性和方法主要如下：

用法举例如下：

testSummary=LR_model.evaluate(testSet)
accuracy=testSummary.accuracy
FPR=testSummary.falsePositiveRateByLabel
precision=testSummary.precisionByLabel
Recall=testSummary.recallByLabel
TPR=testSummary.truePositiveRateByLabel
F_meature=testSummary.fMeasureByLabel()
1
2
3
4
5
6
7

这里要注意，testSet必须是未经模型训练的，否则会抱如下错误：IllegalArgumentException: requirement failed: Column prediction already exists.

2. Clustering

2.1 概览

  ml.Clustering包中提供的聚类算法及其类如下：

2.2 使用案例

  这里使用K均值++对iris数据集进行聚类。其具体代码如下：

from pyspark.sql import SparkSession
import os
from pyspark.ml.feature import *
from pyspark.ml.clustering import *
from pyspark.sql.types import *
os.environ['SPARK_HOME'] ='/Users/sherry/documents/spark/spark-3.2.1-bin-hadoop3.2'
spark=SparkSession.builder.appName('ml').getOrCreate()

#读取数据集和测试集
trainSet=spark.read.csv(r'/Users/sherry/Downloads/iris_training.csv',
                        schema='Feat1 FLOAT,Feat2 FLOAT,Feat3 FLOAT,Feat4 FLOAT,label INT')
testSet=spark.read.csv(r'/Users/sherry/Downloads/iris_test.csv',
                       schema='Feat1 FLOAT,Feat2 FLOAT,Feat3 FLOAT,Feat4 FLOAT,label INT')

vect_assemble=VectorAssembler(inputCols=['Feat1','Feat2','Feat3','Feat4'],
                              outputCol='features')
trainSet=vect_assemble.transform(trainSet)
testSet=vect_assemble.transform(testSet)

kmean=KMeans(featuresCol='features',
              k=3)
kmean_Model=kmean.fit(trainSet)
trainSet=kmean_Model.transform(trainSet)
trainSet.select('label','prediction').show(5)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

其结果如下：

3. Regression

3.1 概览

  ml.Regression包中提供的回归算法及其类如下：

参考

1. https://spark.apache.org/docs/latest/api/python/reference/pyspark.ml.html#classification