文章目录

第1关：Iris 分类
- 任务描述
- 相关知识
- - 1：观察数据集
  - 2：RFormula 特征提取
  - 3：pandas 的 concat 函数
- 编程要求
代码实现
————————————————————————————————————————
第2关：图片识别 - 坦克类型分类
- 任务描述
- 相关知识
- - 1：数据集介绍
  - 2：加载图片数据集
  - 3：将一维数组转换成 Spark 中的向量
  - 4：将向量与标签进行绑定并将其转换成Dataframe
  - 5：Spark 加载数据集
  - 6：将数据集拆分训练集和测试集
  - 7：创建LR分类器
  - 8：训练模型
- 编程要求
- 测试说明
代码实现

第1关：Iris 分类

任务描述

本关任务：使用 pyspark ml 的LogisticRegression分类器完成 Iris 分类任务。

相关知识

1：观察数据集

我们本次使用的数据集是sklearn自带的数据集Iris。

接下来，我们来了解下Iris数据集的数据吧！

示例代码：

iris = load_iris()
features = pandas.DataFrame(iris.data, columns=iris.feature_names)
target = pandas.DataFrame(iris.target, columns=['label'])
print(features.head())
print(target.head())

打印结果：

   sepal length (cm)  sepal width (cm)  petal length (cm)  petal width (cm)
0                5.1               3.5                1.4               0.2
1                4.9               3.0                1.4               0.2
2                4.7               3.2                1.3               0.2
3                4.6               3.1                1.5               0.2
4                5.0               3.6                1.4               0.2
   label
0      0
1      0
2      0
3      0
4      0

简单来说明下数据集，我们第一个打印输出的结果集是Iris的特征，第二个打印输出的结果集是Iris的标签，也就是说我们接下来就是通过这些特征来训练模型，再通过模型预期一些没有标签的特征数据的标签。

2：RFormula 特征提取

RFormula 借鉴了 R 语言的 R model formula 进行特征选择。

RFormula 仅支持部分 R 操作，包括：

~ 分隔因变量（LHS 左侧）和自变量（RHS 右侧）；
+ 合并变量，+0 意为移除截距；
- 移除变量，-1 意为移除截距；
: 变量相交，（数字型值相乘，类别型值二值化）；
. 除了因变量的所有变量。

上面的介绍是否有点抽象？

接下来，我们看下面的例子进行案例讲解吧！

假设有一个4列的dataframe：

id	country	hour	clicked
7	“US”	18	1.0
8	“CA”	12	0.0
9	“NZ”	15	0.0

如果使用RFormula，并且构建公式： clicked ~ country + hour，它表示通过country，hour这两个特征去预测clicked这个因变量。于是我们会得到以下dataframe:

id	country	hour	clicked	features	label
7	“US”	18	1.0	[0.0, 0.0, 18.0]	1.0
8	“CA”	12	0.0	[0.0, 1.0, 12.0]	0.0
9	“NZ”	15	0.0	[1.0, 0.0, 15.0]	0.0

features列为转换后的特征表示，因为country是字符串类型的类编变量，故进行one-hot编码变成了两列， hour是数值型的，故转换成double类型。label列是因变量click列，双精度的类型保持不变。

示例：

from pyspark.sql import SparkSession
from pyspark.ml.feature import RFormula
if __name__ == '__main__':
    spark = SparkSession.builder.appName("demo").master("local[*]").getOrCreate()
    df = spark.createDataFrame((
        (7, "US", 18, 1.0),
        (8, "CA", 12, 0.0),
        (9, "NZ", 15, 0.0)
    )).toDF("id", "country", "hour", "clicked")
    df.show()
    formula = RFormula(formula="clicked ~ country + hour")
    formula.fit(df).transform(df).show()

打印结果：

+---+-------+----+-------+--------------+-----+
|id |country|hour|clicked|feature       |label|
+---+-------+----+-------+--------------+-----+
|7  |US     |18  |1.0    |[0.0,0.0,18.0]|1.0  |
|8  |CA     |12  |0.0    |[1.0,0.0,12.0]|0.0  |
|9  |NZ     |15  |0.0    |[0.0,1.0,15.0]|0.0  |
+---+-------+----+-------+--------------+-----+
+--------------+-----+
|feature       |label|
+--------------+-----+
|[0.0,0.0,18.0]|1.0  |
|[1.0,0.0,12.0]|0.0  |
|[0.0,1.0,15.0]|0.0  |
+--------------+-----+

3：pandas 的 concat 函数

concat 函数是在pandas的方法，可以将数据根据不同的轴作简单的融合。

源码：

def concat(
    objs,
    axis = 0,
    join = "outer",
    join_axes = None,
    ignore_index = False,
    keys = None,
    levels = None,
    names = None,
    verify_integrity = False,
    sort = None,
    copy = True,
):# 省略

常用参数：

objs: series，dataframe 或者是 panel 构成的序列 lsit ；
axis：需要合并链接的轴，0是行，1是列；
join：连接的方式 inner，或者 outer。

示例：

import pandas
df1 = pandas.DataFrame([[1, 2, 3], [4, 5, 6], [7, 8, 9]])
df2 = pandas.DataFrame([["A", "B", "C"], ["D", "E", "F"], ["G", "H", "I"]])
df3 = pandas.concat([df1, df2], axis=1)
print(df3.head())
df4 = pandas.concat([df1, df2], axis=0)
print(df4.head())

打印结果：

   0  1  2  0  1  2
0  1  2  3  A  B  C
1  4  5  6  D  E  F
2  7  8  9  G  H  I
   0  1  2
0  1  2  3
1  4  5  6
2  7  8  9
0  A  B  C
1  D  E  F

编程要求

请仔细阅读右侧代码，结合相关知识，在Begin-End区域内进行代码补充，实现Iris 种类分类的模型训练，具体需求如下：

1：利用pandas将原始数组转换成DataFrame矩阵；

2：利用concat将特征与目标矩阵按行连接；

3：利用SparkSession将本地的pandas.DataFrame上传到Spark集群中，变成了Spark中的分布式DataFrame；

4：利用RFormula将DataFrame中属于特征的列提取到features字段中，将目标提取到labal字段；

5：将数据随机切分成8:2的训练集与测试集；

6：利用spark的逻辑回归分类器进行分布式训练fit；

7：然后返回元组，元组构造为（模型、测试集）。

代码实现

————————————————————————————————————————

# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
from sklearn.datasets import load_iris
import pandas
from pyspark.ml.classification import LogisticRegression
from pyspark.mllib.evaluation import BinaryClassificationMetrics
from pyspark.ml.feature import RFormula


# 训练模型
def trainingModel(spark):
    # ********** Begin ********** #
    
    # 1.加载sklearn的训练数据
    iris = load_iris()

    # 2.特征矩阵
    features = pandas.DataFrame(iris.data, columns = iris.feature_names)

    # 3.目标矩阵
    targets = pandas.DataFrame(iris.target, columns = ['Species'])

    # 4.合并矩阵
    merged = pandas.concat([features, targets], axis = 1)
    
    # 5.创建spark DataFrame
    raw_df = spark.createDataFrame(merged)

    # 6.提取特征与目标
    fomula = RFormula(formula = 'Species ~ .')
    raw_df = fomula.fit(raw_df).transform(raw_df)
    # 7.拆分训练集和测试集
    train_df, test_df = raw_df.randomSplit([0.8, 0.2])

    # 8.创建LR分类器
    lr = LogisticRegression(family = "multinomial")
    
    # 9.训练
    model = lr.fit(train_df)

    # 10.返回（模型、测试集）
    return (model, test_df)
    # ********** End ********** #

第2关：图片识别 - 坦克类型分类

任务描述

本关任务：使用 pyspark ml 的LogisticRegression分类器完成坦克类型分类任务。

相关知识

1：数据集介绍

三种坦克图片数据集，如下图所示：
在这里插入图片描述

以上有三种数据集，分别是以bmp-2开头的BMP-2步兵战车的图片、以btr-70开头的BTR-70装甲输送车的图片、以t-72开头的T-72主战坦克的图片。

2：加载图片数据集

我们可以使用opencv_python计算机视觉库来读取一幅图像。

在读取图像之前，我们需要对图像的构造有一定的理解，怎么理解呢？

简单来说图像就是一个矩阵，在OpenCV for Python中，图像就是NumPy中的数组！

我们知道图像就是一个矩阵了，之后我们通过代码来读取呢？

示例:

读入图像文件，返回numpy数组

img = cv2.imread('/root/1.jpg', cv2.IMREAD_GRAYSCALE)
rows, columns = img.shape

将二维数组转一维数组

img = img.reshape(rows*columns)

3：将一维数组转换成 Spark 中的向量

上一步我们将图片转换成一维的数组之后了，接下来我们将一维数组转换成 Spark 中的向量。

可能你会有疑问？为什么要将一维数组转换成 Spark 中的向量呢？

简单来说就是类型不匹配了。

示例：

import cv2
img = cv2.imread("testing2\\bmp-2-1.jpg", cv2.IMREAD_GRAYSCALE)
rows, columns = img.shape
img = img.reshape(rows * columns)
print(type(img))

结果如下：

<class 'numpy.ndarray'>

我们都知道numpy.ndarray是Python中的类型，Spark是由Scala编写的，它怎么可能识别的了numpy.ndarray类型呢？

那我们怎么将将一维数组转换成 Spark 中的向量呢？

很简单，直接使用from pyspark.ml.linalg import Vectors中的Vectors来进行转换即可！

在pyspark中，有两种向量，分别是稠密向量和稀疏向量。

DenseVector ：稠密向量其创建方式 Vector.dense(数据)

SparseVector ：稀疏向量其创建方式有两种：

方法一：Vector.sparse(向量长度，索引数组，与索引数组所对应的数值数组)

方法二：Vector.sparse(向量长度，（索引，数值），（索引，数值），（索引，数值），…(索引，数值))

比如向量（1,0,3,4）的创建有三种方法：

稠密向量：直接Vectors.dense(1,0,3,4)

稀疏向量：

方法一：Vector.sparse(4,(0,2,3),(1,3,4))

表示该向量的第0个，第2个，第3个位置，(1,3,4) 表示（0,2,3）位置对应的数值分别为1,3,4

方法二：Vector.sparse(4,(0,1),(2,3),(3,4))

（0,1）就是（索引，数值）的形式。位置0的数值为1, 位置2的数值为3,位置3的数值为4。

一般我们会直接将数组转换成稠密向量。

4：将向量与标签进行绑定并将其转换成Dataframe

上面的两步，我们已经加载了图片数据并将其转换成Spark中的向量了，但是这样还远远不够，为什么呢？

因为我将图片数据转换成向量之后，这样只能得出该图片的特征，但是我们还没有指定特征的标签。

我们回到最初加载图片数据那步，这里我们在读取图片数据的同时获取图片的文件名，我们通过文件名进行打标签，具体要求如下：

如果文件名包含 btr-70 ，label 设置为 0 ;

如果文件名包含 t-72 ，label 设置为 1;

其他的，label 设置为 2。

我们得到标签之后，我们将特征向量与该标签进行绑定，构建一个元组。

构建元组之后，我们使用pandas将元组数据加载成Dataframe并将特征向量列的列名设置为features，标签的列的列名设置为label。

为什么要这样做呢？

我们查看下LogisticRegression算法的源码

构造函数源码如下：

@keyword_only
    def __init__(self, featuresCol="features", labelCol="label", predictionCol="prediction",
                 maxIter=100, regParam=0.0, elasticNetParam=0.0, tol=1e-6, fitIntercept=True,
                 threshold=0.5, thresholds=None, probabilityCol="probability",
                 rawPredictionCol="rawPrediction", standardization=True, weightCol=None,
                 aggregationDepth=2, family="auto",
                 lowerBoundsOnCoefficients=None, upperBoundsOnCoefficients=None,
                 lowerBoundsOnIntercepts=None, upperBoundsOnIntercepts=None):
        super(LogisticRegression, self).__init__()
        self._java_obj = self._new_java_obj(
            "org.apache.spark.ml.classification.LogisticRegression", self.uid)
        self._setDefault(maxIter=100, regParam=0.0, tol=1E-6, threshold=0.5, family="auto")
        kwargs = self._input_kwargs
        self.setParams(**kwargs)
        self._checkThresholdConsistency()

通过__init__函数中的参数，我们可以看到featuresCol的默认值为features，labelCol的默认值为label。

通过pandas构建了 Dataframe之后指定了列名这样做的话，就可以不用设置featuresCol、labelCol了，同时增强了代码可读性。

5：Spark 加载数据集

上面我们已经把数据集处理完毕之后，然后我们可以通过Spark来加载数据集。

spark = SparkSession.builder.master("local[*]").appName("demo").getOrCreate()
sparkDF = spark.createDataFrame(df) # df是我们通过pandas构建的Dataframe

6：将数据集拆分训练集和测试集

上面我们把图片数据集加载到Spark中了，之后我们要把数据集划分成两部分，一部分为训练集，另一部分为测试集。

简单来说，训练集就是用来训练模型的，测试集就是来评测这样模型的好与坏的。

关于训练集与测试集的比例问题，一般来说训练集越多越好，我们一般将训练集与测试集的比例调整为8:2或者7:3。

train_df, test_df = sparkDF.randomSplit([0.7, 0.3])

7：创建LR分类器

逻辑回归是一种用于预测分类响应的流行方法。这是广义线性模型的一种特殊情况，可以预测结果的可能性。在spark.ml逻辑回归中，可以通过使用二项式逻辑回归来预测二进制结果，或者在使用多项式逻辑回归时可以将其预测为多类结果。使用该family 参数在这两种算法之间进行选择，或者将其保留为未设置状态，Spark会推断出正确的变体。

通过将family参数设置为“多项式”，可以将多项式逻辑回归用于二进制分类。它将产生两组系数和两个截距。

当对具有恒定非零列的数据集进行LogisticRegressionModel拟合而没有截距时，Spark MLlib为恒定非零列输出零系数。

创建LR分类器用来训练模型：

lr = LogisticRegression(family="multinomial")

8：训练模型

上一步，我们已经把LR分类器构建好了并且我们把数据集划分成训练集与测试集了，接下来我们就是训练模型了。

model = lr.fit(train_df)

编程要求

请仔细阅读右侧代码，结合相关知识，在Begin-End区域内进行代码补充，实现坦克种类分类的模型训练，具体需求如下：

1：补全 get_file_path函数，完成遍历目录获取图片路径列表功能；

2：补全img2vector函数，完成提取图片特征并将其转换成向量功能；

3：补全trainingModel函数，完成模型的训练，最后返回一个元组，元组构造为（测试集，模型）

测试说明

代码实现

平台会使用利用 Spark 的一个Metrics库对你训练的模型进行统计模型得分，如果模型的AUC值符合预期值，则通关。

import cv2
import os
import pandas as pd
from pyspark.mllib.evaluation import BinaryClassificationMetrics
from pyspark.sql import SparkSession
from pyspark.ml.linalg import Vectors
from pyspark.ml.classification import LogisticRegression, NaiveBayes, LinearSVC
from pyspark.ml.regression import LinearRegression


# 获取目录的所有文件，返回图片路径列表
def get_file_path(root_path):
    # ********** Begin ********** #
    file_list = []
    dir_or_files = os.listdir(root_path)
    for dir_file in dir_or_files:
        dir_file_path = os.path.join(root_path, dir_file)
        file_list.append(dir_file_path)
    return file_list
    # ********** End ********** #
    

# 提取图片特征并将其转换成向量，返回特征向量
def img2vector(imgfilename):
    # ********** Begin ********** #
    img = cv2.imread(imgfilename, cv2.IMREAD_GRAYSCALE)
    rows, columns = img.shape
    img = img.reshape(rows * columns)
    return Vectors.dense(img)
    # ********** End ********** #


# 训练模型，返回元组（测试集，模型）
def trainingModel(spark,root_path):
    datas = []
    # ********** Begin ********** #

    # 1. 获取 root_path 下的所有图片文件
    fileList = get_file_path(root_path)

    # 2.1 遍历图片文件列表，根据文件名给数据打标签
    """
        如果文件名包含 `btr-70` ，label 设置为 0 
        如果文件名包含 `t-72` ，label 设置为 1
        其他的，label 设置为 2
    """
    # 2.2 调用img2vector函数，提取图片特征并将其转换成向量
    # 2.3 将向量与标签（label）构建成元组并存储到 datas 列表中
    for x in fileList:
        vector = img2vector(x)
        if "btr-70" in x:
            label = 0
        elif "t-72" in x:
            label = 1
        else:
            label = 2
        datas.append((vector, label))
   
    # 3. 利用 pandas 将 datas 转换成 DataFrame 并将包含向量的列的列名设置为 features，包含标签（label） 的列的列名设置为 label
    df = pd.DataFrame(datas, columns=['features', 'label'])
    
    # 4. 使用 spark 加载 pandas 转换后的数据集 
    sparkDF = spark.createDataFrame(df)

    # 5. 将数据集拆分训练集和测试集
    train_df, test_df = sparkDF.randomSplit([0.7, 0.3])

    # 6. 创建LR分类器
    lr = LogisticRegression(family="multinomial")

    # 7. 训练模型
    model = lr.fit(train_df)

    # 8.返回元组（测试集，模型）
    return (test_df,model)
    # ********** End ********** #