pySpark 机器学习库ml入门

在spark中提供了两个机器学习库mllib和ml，mllib的操作是基于RDD的，而ml则是基于DataFrame，是主流机器学习库。

1、ml包的概述

ml包包括三个主要的抽象类：转换器（Transformer）、评估器（Estimator）和管道（Pipeline）。

1.1 转换器

转换器类通过将一个新列附加到DataFrame来转换数据。

从高层次上看，当从转换器的抽象类派生时，每个新的转换器类需要实现.transform()方法。该方法要求传递一个要被转换的DataFrame，该参数通常是第一个也是唯一的一个强制性参数。这在ml包的不同方法中也不尽相同：其他常见的参数有inputCol和outputCol；然而，这些参数通常用一些预定义的值作为默认值，例如inputCol参数的默认值为“features”。

spark.ml.feature中提供了许多转换器，下面做个简要介绍：

• Binarizer, 根据指定的阈值将连续变量转换为对应的二进制值；
• Bucketizer, 与Binarizer类似，该方法根据阈值列表（分割的参数），将连续变量转换为多项值（即将连续变量离散到指定的范围区间）；
• ChiSqSelector, 对于分类目标变量，此功能允许你选择预定义数量的特征（由numTopFeatures参数指定），以便更好地说明目标的变化。如名所示，该方法使用卡方检验（Chi-Square）完成选择。该方法需要两步：首先，需要fit()数据。调用fit方法返回一个ChipSelectorModel对象，然后使用该对象的transform方法来转换DataFrame；
• countVectorizer, 该方法对于标记文本（如[['learning', 'pyspark', 'with', 'us'], ['us', 'us', 'us', 'us']]）是有用的。这是一个需要两步的方法：首先，需要用fit从数据集中学习这些模式，然后才能使用fit方法返回的CountVectorizerModel对象的transform方法。对于如上所示的标记文本，该转换器的输出类似于`[(4, [0, 1, 2, 3], [1.0, 1.0, 1.0, 1.0]), (4, [3], [3.0])]。
• DCT, 离散余弦变换取实数值向量，并返回相同长度的向量，但余弦函数之和在不同频率下振荡。这种转换对于提取数据或数据压缩中的一些基本频率很有用。
• ElementwiseProduct, 该方法返回一个向量，其中的元素是传入该方法的向量和另一个传入参数scalingVec向量的乘积。例如，如果传入的向量是[10.0, 3.0, 15.0], 而传入的scalingVec为[0.99, 3.30, 0.66], 那么将获得如下所示的向量：[9.9, 9.9, 9.9, 9.9]；
• HashingTF, 一个哈希转换器，输入为标记文本的列表，返回一个带有技术的有预定长度的向量。摘自pyspark文档："由于使用简单的模数将散列函数转换为列索引，建议使用2的幂作为numFeatures参数；否则特征将不会均匀的映射到列"；
• IDF, 该方法计算文档列表的逆向文件频率。请注意，文档需要提前用向量表示（例如，使用HashingTF或CountVectorizer）；
• IndexToString, 与StringIndexer方法对应。它使用StringIndexerModel对象中的编码将字符串索引反转到原始值。另外请注意，如果有时不起作用，你需要指定StringIndexer中的值；
• MaxAbsScaler, 将数据调整到[-1,0,1]范围内（因此不会移动数据中心）；
• MinMaxScaler, 这与MaxAbsScaler相似，区别在于它将数据缩放到[0.0, 1.0]范围内；
• NGram, 此方法的输入为标记文本的列表，返回结果包含一系列n-gram：以两个词、三个词或更多的n个词记为一个n-gram。例如，如果你有一个['good', 'morning', 'Robin', 'Williams'], 你会得到以下输出：['good morning', 'morning Robin', 'Robin Williams'];
• Normlizer, 该方法使用p范数将数据缩放为单位范数(默认为L2);
• OneHotEncoder, 该方法将分类列编码为二进制向量列；
• PCA, 使用主成分分析执行数据降维；
• PolynomiaExpansion, 执行向量的多项式展开。例如，加入你有一个如[x,y,z]的向量，则该方法将产生一下扩展：[x, xx, y, xy, yy, z, xz, yz, zz];
• QuantileDiscretizer, 与Bucketizer方法类似，但不是传递分隔参数，而是传递一个numBuckets参数。然后，该方法通过计算数据的近似分位数来决定分隔应该是什么；
• RegexTokenizer, 这是一个使用正则表达式的字符串分词器；
• RFormula, 对于狂热的R用户，你可以传递一个公式，如vec ~ alpha*3 + beta (假设你的DataFrame具有alpha和beta列)，它将产生给定表达式的vec列；
• SQLTransformer, 与上面相似，但不是类似R的公式，你可以使用SQL语法（FROM 语句应该从__THIS中选择，表示你正在访问DataFrame，如SELECT alpha*3 + beta AS vec FROM THIS）;
• StandardScaler, 标准化列，使其拥有零均值和等于1的标准差；
• StopWordsRemover, 从标记文本中删除停用词（如'the', 'a');
• StringIndexer, 假设包含所有单词的列表都在一列，这将产生一个索引向量；
• Tokenizer (分词器)：该默认分词器将字符串转成小写，然后以空格为分隔符分词；
• VectorAssembler, 这是一个非常有用的转换器，他将多个数字（包括向量）列合并为一列向量；
• VectorIndexer, 该方法为类别列生成索引向量。它以逐列方式工作，从列中选择不同的值，排序并从映射中返回值的索引而不是原始值；
• VectorSlicer, 用于特征向量，不管是密集的还是稀疏的：给定一个索引列表，它从特征向量中提取值；
• Word2Vec, 该方法将一个句子（字符串）作为输入 ，并将其转换为{string, vector}格式的映射，这种表示在自然语言处理中非常有用。

1.2 评估器

评估器可以被视为需要评估的统计模型，对你的观测对象做预测或分类。

如果从抽象的评估器类派生，新模型必须实现fit方法，该方法给出的在DataFrame中找到的数据和某些默认或自定义的参数来拟合模型。在pyspark中有很多评估器可用，下面简要介绍下spark中提供的模型。

分类

ML包提供了七种分类（Classification）模型以供选择，从最简单的逻辑回归到一些更复杂的模型，下面作简要的描述：

• LogisticRegression, 分类的基准模型。逻辑回归使用一个对数函数来计算属于特定类别的观察对象的概率；
• DecisionTreeClassifier, 该分类器构建了一个决策树来预测一个观察对象的所属类别。指定maxDepth参数限制树的深度，minInstancePerNode确定需要进一步拆分的树节点的观察对象的最小数量，maxBins参数指定连续变量将被分割的Bin的最大数量，而impurity指定用于测量并计算来自分隔的信息的度量；
• GBTClassifier, 用于分类的梯度提升决策树模型。该模型属于集合模型家族：集合模型结合多个弱预测模型而形成一个强健的模型；
• RandomForestClassifier，该模型产生多个决策树，并使用模式输出的决策树来对观察对象进行分类；
• NaiveBayes, 基于贝叶斯定理，该模型使用条件概率理论对观测进行分类；
• MultilayerPerceptronClassfier, 多层感知器分类器。模仿人类大脑本质的分类器，深深植根于人造神经网络理论，该模型是一个黑盒模型，内部参数不易解释。该模型至少包含三个完全相连的人造神经元层：输入层（需要和数据集中特征的数量一样）、多个隐藏层（至少一个）以及一个输出层，其神经元数量等于标签中的类别数量。输入层和隐藏层中的所有神经元都有sigmoid激活函数，而输出神经元的激活函数则为softmax。
• OneVsRest，将多分类问题简化为二分类问题。例如，在多标签的情况下，模型可以训练成多个二元逻辑回归模型。如多标签情况下，模型可以训练成多个二元逻辑回归模型。所有模型分别计分，具有最高概率的模型获胜。

回归

pyspark ML软件包中有七种可用于回归（Regression）任务的模型。与分类一样，范围从一些基本的回归（如强制线性回归）到更复杂的回归：

• AFTSurvivalRegression，适合加速失效时间回归模型。它是一个参数化模型，假设其中一个特征的边际效应加速或减缓了预期寿命（或过程失败）。它非常适用于具有明确阶段的过程；
• DecisionTreeRegressor， 类似于分类模型，明显不同的是其标签是连续的而不是二元的；
• GBTRegressor, 与DecisionTressRegressor一样，区别在于标签的数据类型；
• GeneralizedLinearRegression，广义线性回归是具有不同内核功能（链接功能）的线性模型家族。与假设误差项的常态性的线性回归相反，GLM允许标签具有不同的误差分布项：pyspark ML包的generalizedRegression模型支持gaussian、binomial、gamma、和possion家族的误差分布，他们有许多不同的连接功能；
• IsotonicRegression, 这种回归拟合一个形式自由的、非递减的行到数据中。对于拟合有序的和递增的观测数据集是有用的；
• LinearRgression, 最简单的回归模型，他架设了特征与连续标签以及误差项的常态之间的线性关系；
• RandomForestRegressor, 与DecisionTreeRegressor或GBTRegressor类似，RandomForestRegressor适合连续的标签，而不是离散标签。

聚类

聚类是一系列无监督模型，用于查找数据中的隐含模式。pyspark ML包提供了四种当前最流行的模型：

• BisectingKMeans, 二分k均值算法，该算法结合了k均值聚类算法和层次聚类算法。最初该算法将所有观察点作为一个簇，然后将数据迭代的分解为k个簇；
• KMeans, K均值算法，将数据分成k个簇，迭代地搜索那些使每个观察点和它所属簇的质点之间距离平方和最小的那些质点；
• GaussianMixture, 混合高斯模型。该方法使用具有未知参数的k个高斯分布来剖析数据集。使用期望最大化算法，通过最大化对数似然函数找到高斯参数；
• LDA, 该模型用于自然语言处理应用程序中的主题生成；
• 除此之外，pyspark ML还提供了推荐模型。

1.3 管道

pyspark ML中管道的概念用来表示从转换到评估（具有一系列不同阶段）的端到端的过程，这个过程可以对输入的一些原始数据（以DataFrame形式）执行必要的数据加工（转换），最后评估模型。

一个管道可以被认为是由一系列不同阶段组成的。在Pipeline对象上执行fit方法时，所有阶段按照stage参数中指定的顺序执行；stage参数是转换器和评估器对象的列表。管道对象的fit方法执行每个转换器的transform方法和所有评估器的fit方法。

通常，前一阶段的输出会成为下一阶段的输入：当从转换器或评估器抽象类型派生时，需要实现getOutputCol()方法，该方法返回创建对象时指定的outputCol参数的值。

下面通过一些例子来详细介绍ml的用法。

2、例子：使用ML预测婴儿生存几率

在本节中我们将使用ml中的函数方法来预测婴儿生存率，数据可从http://www.tomdrabas.com/data/LearningPySpark/births_transformed.csv.gz下载。

2.1 加载数据

import pyspark.sql.types as typ
from pyspark.ml import Pipeline
import pyspark.ml.classification as cl
import pyspark.ml.evaluation as ev
import pandas as pd
import numpy as np
import os

labels = [('INFANT_ALIVE_AT_REPORT', typ.IntegerType()),
          ('BIRTH_PLACE', typ.StringType()),
          ('MOTHER_AGE_YEARS', typ.IntegerType()),
          ('FATHER_COMBINE_AGE', typ.IntegerType()),
          ('CIG_BEFORE', typ.IntegerType()),
          ('CIG_1_TRI', typ.IntegerType()),
          ('CIG_2_TRI', typ.IntegerType()),
          ('CIG_3_TRI', typ.IntegerType()),
          ('MOTHER_HEIGHT_IN', typ.IntegerType()),
          ('MOTHER_PRE_WEIGHT', typ.IntegerType()),
          ('MOTHER_DELIVERY_WEIGHT', typ.IntegerType()),
          ('MOTHER_WEIGHT_GAIN', typ.IntegerType()),
          ('DIABETES_PRE', typ.IntegerType()),
          ('DIABETES_GEST', typ.IntegerType()),
          ('HYP_TENS_PRE', typ.IntegerType()),
          ('HYP_TENS_GEST', typ.IntegerType()),
          ('PREV_BIRTH_PRETERM', typ.IntegerType())
          ]

schema = typ.StructType([
    typ.StructField(e[0], e[1], False) for e in labels
])

births = spark.read.csv(
    '/Users/shexuan/Downloads/births_transformed.csv.gz', header=True, schema=schema)

births.show(3)
+----------------------+-----------+----------------+------------------+----------+---------+---------+---------+----------------+-----------------+----------------------+------------------+------------+-------------+------------+-------------+------------------+
|INFANT_ALIVE_AT_REPORT|BIRTH_PLACE|MOTHER_AGE_YEARS|FATHER_COMBINE_AGE|CIG_BEFORE|CIG_1_TRI|CIG_2_TRI|CIG_3_TRI|MOTHER_HEIGHT_IN|MOTHER_PRE_WEIGHT|MOTHER_DELIVERY_WEIGHT|MOTHER_WEIGHT_GAIN|DIABETES_PRE|DIABETES_GEST|HYP_TENS_PRE|HYP_TENS_GEST|PREV_BIRTH_PRETERM|
+----------------------+-----------+----------------+------------------+----------+---------+---------+---------+----------------+-----------------+----------------------+------------------+------------+-------------+------------+-------------+------------------+
|                     0|          1|              29|                99|         0|        0|        0|        0|              99|              999|                   999|                99|           0|            0|           0|            0|                 0|
|                     0|          1|              22|                29|         0|        0|        0|        0|              65|              180|                   198|                18|           0|            0|           0|            0|                 0|
|                     0|          1|              38|                40|         0|        0|        0|        0|              63|              155|                   167|                12|           0|            0|           0|            0|                 0|
+----------------------+-----------+----------------+------------------+----------+---------+---------+---------+----------------+-----------------+----------------------+------------------+------------+-------------+------------+-------------+------------------+

在这里我们指定DataFrame的schema，限制数据集只有17列。

2.2 创建转换器

在使用模型对数据集进行评估预测前，先要对数据做一些特征转换。

# 创建转换器、评估器
import  pyspark.ml.feature as ft

births = births.withColumn('BIRTH_PLACE_INT', births['BIRTH_PLACE']\
    .cast(typ.IntegerType()))

# birth place使用one-hot编码
encoder = ft.OneHotEncoder(inputCol='BIRTH_PLACE_INT',
                           outputCol='BIRTH_PLACE_VEC')

# 创建单一的列将所有特征整合在一起
featuresCreator = ft.VectorAssembler(
    inputCols=[col[0] for col in labels[2:]] + [encoder.getOutputCol()],
    outputCol='features'
)

# 创建一个评估器
import pyspark.ml.classification as cl

logistic = cl.LogisticRegression(maxIter=10,
                                regParam=0.01,
                                featuresCol=featuresCreator.getOutputCol(),
                                labelCol='INFANT_ALIVE_AT_REPORT')

2.3 创建一个管道、拟合模型

在前面我们已经创建了数据转换器和评估器，现在我们可以通过管道将他们串联起来并方便的进行模型拟合了。

# 创建一个管道
from pyspark.ml import Pipeline

pipeline = Pipeline(stages=[encoder, featuresCreator, logistic])

# 拟合模型
birth_train, birth_test = births.randomSplit([0.7,0.3],seed=123)

model = pipeline.fit(birth_train)
test_model = model.transform(birth_test)

2.4 评估模型

在前面我们将数据分为了两部分并通过管道方便的对训练集进行了拟合以及对测试集进行了测试。现在我们可以通过测试集的结果来对模型拟合效果进行评估了。

# 评估模型性能
import pyspark.ml.evaluation as ev

evaluator = ev.BinaryClassificationEvaluator(
    rawPredictionCol='probability',
    labelCol='INFANT_ALIVE_AT_REPORT'
)

print(evaluator.evaluate(test_model, {evaluator.metricName:'areaUnderROC'}))
print(evaluator.evaluate(test_model, {evaluator.metricName:'areaUnderPR'}))

0.7187355793173213
0.6819691176245866

2.5 保存模型

PySpark不仅允许保存训练好的模型，还可以保存管道结构及所有转换器和评估器的定义。

# 保存模型pipeline
pipelinePath = './infant_oneHotEncoder_Logistic_Pipeline'
pipeline.write().overwrite().save(pipelinePath)

# 重载模型pipeline
loadedPipeline = Pipeline.load(pipelinePath)
loadedPipeline.fit(birth_train).transform(birth_test).take(1)

[Row(INFANT_ALIVE_AT_REPORT=0, BIRTH_PLACE='1', ...]

# 保存模型
from pyspark.ml import PipelineModel

modelPath = './infant_oneHotEncoder_LogisticPipelineModel'
model.write().overwrite().save(modelPath)

# 载入模型
loadedPipelineModel = PipelineModel.load(modelPath)
test_reloadedModel = loadedPipelineModel.transform(birth_test)
test_reloadedModel.take(1)

[Row(INFANT_ALIVE_AT_REPORT=0, BIRTH_PLACE='1', MOTHER_AGE_YEARS=12, ...]

2.6 超参调优

我们的第一个模型几乎不可能是最好的模型。利用超参调优能帮我们找到模型的最佳参数，如逻辑回归模型所需的最佳迭代次数或决策树的最大深度。

在超参调优时PySpark提供了两种验证方法：K-Fold交叉验证和train-validation（相当于1-Fold）交叉验证。

# 超参调优：grid search和train-validation splitting 

# 网格搜索
import pyspark.ml.tuning as tune

logistic = cl.LogisticRegression(labelCol='INFANT_ALIVE_AT_REPORT')
grid = tune.ParamGridBuilder()\
    .addGrid(logistic.maxIter, [5,10,50])\
    .addGrid(logistic.regParam, [0.01,0.05,0.3])\
    .build()

evaluator = ev.BinaryClassificationEvaluator(
    rawPredictionCol='probability',
    labelCol='INFANT_ALIVE_AT_REPORT'
)

# 使用K-Fold交叉验证评估各种参数的模型
cv = tune.CrossValidator(
    estimator=logistic,
    estimatorParamMaps=grid,
    evaluator=evaluator,
    numFolds=3
)

# 创建一个构建特征的pipeline
pipeline = Pipeline(stages=[encoder, featuresCreator])
birth_train, birth_test = births.randomSplit([0.7,0.3],seed=123) # 重新打开数据进行处理
data_transformer = pipeline.fit(birth_train)
data_test = data_transformer.transform(birth_test)


# cvModel 返回估计的最佳模型
cvModel = cv.fit(data_transformer.transform(birth_train))
results = cvModel.transform(data_test)

print(evaluator.evaluate(results, {evaluator.metricName:'areaUnderROC'}))
print(evaluator.evaluate(results, {evaluator.metricName:'areaUnderPR'}))

0.735848884034915
0.6959036715961695

使用下面的代码可以查看模型最佳参数：

# 查看最佳模型参数
param_maps = cvModel.getEstimatorParamMaps()
eval_metrics = cvModel.avgMetrics

param_res = []

for params, metric in zip(param_maps, eval_metrics):
    param_metric = {}
    for key, param_val in zip(params.keys(), params.values()):
        param_metric[key.name]=param_val
    param_res.append((param_metric, metric))

sorted(param_res, key=lambda x:x[1], reverse=True)

[({'maxIter': 50, 'regParam': 0.01}, 0.7406291618177623),
 ({'maxIter': 10, 'regParam': 0.01}, 0.735580969909943),
 ({'maxIter': 50, 'regParam': 0.05}, 0.7355100622938429),
 ({'maxIter': 10, 'regParam': 0.05}, 0.7351586303619441),
 ({'maxIter': 10, 'regParam': 0.3}, 0.7248698034708339),
 ({'maxIter': 50, 'regParam': 0.3}, 0.7214679272915997),
 ({'maxIter': 5, 'regParam': 0.3}, 0.7180255703028883),
 ({'maxIter': 5, 'regParam': 0.01}, 0.7179304617840288),
 ({'maxIter': 5, 'regParam': 0.05}, 0.7173397593133481)]

上面使用的使用K-Fold来进行超参调优，K-Fold交叉验证往往非常耗时，使用1-Fold的交叉验证（即将数据集按比例分为训练集合验证集）能大大缩短时间。

# Train-validation划分

# 使用卡方检验选择特征
selector = ft.ChiSqSelector(
    numTopFeatures=5,
    featuresCol=featuresCreator.getOutputCol(),
    outputCol='selectedFeatures',
    labelCol='INFANT_ALIVE_AT_REPORT'
)

logistic = cl.LogisticRegression(labelCol='INFANT_ALIVE_AT_REPORT',
                                featuresCol='selectedFeatures')

pipeline = Pipeline(stages=[encoder, featuresCreator, selector])
data_transformer = pipeline.fit(birth_train)

tvs = tune.TrainValidationSplit(estimator=logistic,
                               estimatorParamMaps=grid,
                               evaluator=evaluator,
                                trainRatio=0.75
                               )

tvsModel = tvs.fit(data_transformer.transform(birth_train))
data_test = data_transformer.transform(birth_test)
results = tvsModel.transform(data_test)

print(evaluator.evaluate(results, {evaluator.metricName:'areaUnderROC'}))
print(evaluator.evaluate(results, {evaluator.metricName:'areaUnderPR'}))

0.6111344483529891
0.5735913338089571

3、使用PySpark ML的其他功能

在上面我们完整的介绍了利用pyspark ml库来进行建模的过程。下面我们介绍一些其他常用的功能。

3.1 特征提取

3.1.1 NLP相关特征提取

如第一部分所述，NGram模型采用标记文本的列表，并生成单词对（或n-gram）。
本例中，我们从pyspark的文档中摘录一段，并介绍如何在将文本传递给NGram模型之前进行清理。

# NLP相关特征提取（NGram模型采用标记文本的列表，并生成单词对或n-gram）

text_data = spark.createDataFrame([
    ['''K-fold cross validation performs model selection by splitting the dataset into a set of non-overlapping 
    randomly partitioned folds which are used as separate training and test datasets e.g., with k=3 folds, 
    K-fold cross validation will generate 3 (training, test) dataset pairs, each of which uses 2/3 of the data 
    for training and 1/3 for testing. Each fold is used as the test set exactly once.'''],
    ['''CrossValidatorModel contains the model with the highest average cross-validation metric across folds and
    uses this model to transform input data. CrossValidatorModel also tracks the metrics for each param map 
    evaluated.'''],
    ['''Creates a copy of this instance with a randomly generated uid and some extra params. This copies the 
    underlying bestModel, creates a deep copy of the embedded paramMap, and copies the embedded and extra 
    parameters over.''']
], ['input'])

# 将文本拆分成单词
tokenizer = ft.RegexTokenizer(inputCol='input',
                              outputCol='input_arr',
                              pattern='\s+|[,.\"]')

# 删掉停用词
stopwords = ft.StopWordsRemover(inputCol=tokenizer.getOutputCol(),
                               outputCol='input_stop')
# 生成ngram词对
ngram = ft.NGram(n=2,
                inputCol=stopwords.getOutputCol(),
                outputCol='nGrams')

# 构建特征pipeline
pipeline = Pipeline(stages=[tokenizer, stopwords, ngram])

data_ngram = pipeline\
    .fit(text_data)\
    .transform(text_data)

data_ngram.show()

+--------------------+--------------------+--------------------+--------------------+
|               input|           input_arr|          input_stop|              nGrams|
+--------------------+--------------------+--------------------+--------------------+
|K-fold cross vali...|[k-fold, cross, v...|[k-fold, cross, v...|[k-fold cross, cr...|
|CrossValidatorMod...|[crossvalidatormo...|[crossvalidatormo...|[crossvalidatormo...|
|Creates a copy of...|[creates, a, copy...|[creates, copy, i...|[creates copy, co...|
+--------------------+--------------------+--------------------+--------------------+

3.1.2 离散连续变量

我们常常需要处理高度非线性连续特征，很难只用一个系数来供给模型。这种情况下，可能难以用一个系数来解释这样的特征与目标之间的关系。有时候，将值划分成分类级别是很有用的。

# 离散连续变量
x = np.arange(0, 100)
x = (x/100.0)*np.pi*4
y = x*np.sin(x/1.764)+20.1234

schema = typ.StructType([typ.StructField('continuous_var', typ.DoubleType(), nullable=False)])
data = spark.createDataFrame([[float(e)] for e in y], schema=schema)
data.show(4)
+------------------+
|    continuous_var|
+------------------+
|           20.1234|
|20.132344452369832|
|20.159087064491775|
|20.203356291885854|
+------------------+


# 使用QuantileDiscretizer模型将连续变量分为五个分类级别
discretizer = ft.QuantileDiscretizer(
    numBuckets=5, 
    inputCol='continuous_var',
    outputCol='discritized'
)

data_discretized = discretizer.fit(data).transform(data)
data_discretized.show(5,truncate=False)
+------------------+-----------+
|continuous_var    |discritized|
+------------------+-----------+
|20.1234           |2.0        |
|20.132344452369832|2.0        |
|20.159087064491775|2.0        |
|20.203356291885854|2.0        |
|20.26470185735763 |2.0        |
+------------------+-----------+

3.1.3 标准化连续变量

标准化连续变量不仅有助于更好地理解特征之间的关系，而且还有助于计算效率，并防止运行到某些数字陷阱。

# 标准化连续变量
# 首先，要创建一个向量代表连续变量（因为它只是一个float）
vectorizer = ft.VectorAssembler(inputCols=['continuous_var'],
                               outputCol='continuous_vec')

normlizer = ft.StandardScaler(inputCol=vectorizer.getOutputCol(),
                             outputCol='normlized',
                             withMean=True,
                             withStd=True)

pipeline = Pipeline(stages=[vectorizer, normlizer])
data_standardized = pipeline.fit(data).transform(data)
data_standardized.show(4)
+------------------+--------------------+--------------------+
|    continuous_var|      continuous_vec|           normlized|
+------------------+--------------------+--------------------+
|           20.1234|           [20.1234]|[0.23429139554502...|
|20.132344452369832|[20.132344452369832]|[0.23630959828688...|
|20.159087064491775|[20.159087064491775]|[0.24234373105179...|
|20.203356291885854|[20.203356291885854]|[0.25233252325644...|
+------------------+--------------------+--------------------+

3.2 聚类

在前面的例子中我们介绍了如何使用pyspark ml库来拟合分类模型，在本节我们将简单介绍pyspark ml库中的聚类模型。

聚类是机器学习中的另一个重要组成部分：通常在现实世界中，我们没有那么幸运具有目标特征，所以需要回到一个无监督的学习范例，来试图从中发掘数据内的模式。

# 聚类
# 使用Kmeans模型在出生数据中查找相似性
import pyspark.ml.clustering as clus

kmeans = clus.KMeans(k=5, featuresCol='features')
pipeline = Pipeline(stages=[encoder, featuresCreator, kmeans])
model = pipeline.fit(birth_train)

test = model.transform(birth_test)
test.groupby('prediction')\
    .agg({'*':'count',
         'MOTHER_HEIGHT_IN':'avg'})\
    .collect()

[Row(prediction=1, avg(MOTHER_HEIGHT_IN)=67.43708609271523, count(1)=453),
 Row(prediction=3, avg(MOTHER_HEIGHT_IN)=67.65714285714286, count(1)=245),
 Row(prediction=4, avg(MOTHER_HEIGHT_IN)=63.92423385728825, count(1)=8843),
 Row(prediction=2, avg(MOTHER_HEIGHT_IN)=84.97315436241611, count(1)=447),
 Row(prediction=0, avg(MOTHER_HEIGHT_IN)=65.45034965034965, count(1)=3575)]

3.3 回归

上面已经介绍过了分类和聚类模型，最后我们再来简单介绍一下回归模型。
在本节中，我们将尝试用给定的一些特征来预测MOTHER_WEIGHT_GAIN。

# 回归
# 使用梯度提升决策树来预测增加的体重
import pyspark.ml.regression as reg

features = ['MOTHER_AGE_YEARS', 'MOTHER_HEIGHT_IN', 'MOTHER_PRE_WEIGHT',
            'DIABETES_PRE', 'DIABETES_GEST', 'HYP_TENS_PRE', 'HYP_TENS_GEST',
            'PREV_BIRTH_PRETERM', 'CIG_BEFORE', 'CIG_1_TRI', 'CIG_2_TRI',
            'CIG_3_TRI']

featuresCreator = ft.VectorAssembler(
    inputCols=[col for col in features[1:]],
    outputCol='features'
)

# 这里使用卡方检验选择前六个最重要的特征
selector = ft.ChiSqSelector(numTopFeatures=6,
                            outputCol='selectedFeatures',
                            labelCol='MOTHER_WEIGHT_GAIN')

regressor = reg.GBTRegressor(maxIter=15,
                            maxDepth=3,
                            labelCol='MOTHER_WEIGHT_GAIN')

pipeline = Pipeline(stages=[featuresCreator, selector, regressor])
weight_gain = pipeline.fit(birth_train)

# 测试集评估
evaluator = ev.RegressionEvaluator(predictionCol='prediction',labelCol='MOTHER_WEIGHT_GAIN')
print(evaluator.evaluate(weight_gain.transform(birth_test),
                        {evaluator.metricName:'r2'}))

0.49363823556949404

虽然模型结果不太好，但是我们的重点不在这里，我们只是为了了解pyspark ml库中回归模型的用法。

参考：

《pyspark 实战指南：利用python和spark构建数据密集型应用并规模化部署》