【pyspark学习从入门到精通20】机器学习库_3

使用 ML 预测婴儿生存几率

在这一部分，我们将使用前一章中的数据集的一部分来介绍 PySpark ML 的概念。

在这一部分，我们将再次尝试预测婴儿的生存几率。

加载数据

首先，我们使用以下代码加载数据：

import pyspark.sql.types as typ
labels = [
 ('INFANT_ALIVE_AT_REPORT', typ.IntegerType()),
 ('BIRTH_PLACE', typ.StringType()),
 ('MOTHER_AGE_YEARS', typ.IntegerType()),
 ('FATHER_COMBINED_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('births_transformed.csv.gz', 
 header=True, 
 schema=schema)

我们指定 DataFrame 的 schema；我们现在严重限制的数据集仅有 17 列。

创建转换器

在我们能够使用数据集来估计模型之前，我们需要进行一些转换。由于统计模型只能操作数值数据，我们将不得不对 BIRTH_PLACE 变量进行编码。

在我们进行这些操作之前，由于我们将在本章后面使用许多不同的特征转换，让我们先导入它们：

import pyspark.ml.feature as ft

为了对 BIRTH_PLACE 列进行编码，我们将使用 OneHotEncoder 方法。然而，该方法不能接受 StringType 列；它只能处理数值类型，所以我们首先将列转换为 IntegerType：

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

完成这些后，我们现在可以创建我们的第一个转换器：

encoder = ft.OneHotEncoder(
 inputCol='BIRTH_PLACE_INT', 
 outputCol='BIRTH_PLACE_VEC')

现在，我们创建一个包含所有特征的单一列。我们将使用 VectorAssembler 方法：

featuresCreator = ft.VectorAssembler(
 inputCols=[
 col[0] 
 for col 
 in labels[2:]] + \
 [encoder.getOutputCol()], 
 outputCol='features'
)

向 VectorAssembler 对象传递的 inputCols 参数是要组合在一起形成 outputCol —— 'features' 的所有列的列表。注意，我们使用编码器对象的输出（通过调用 .getOutputCol() 方法），因此如果我们在编码器对象中更改输出列的名称，我们就不必记住更改此参数的值。

现在轮到创建我们的第一个估计器了。

创建估计器

在这个例子中，我们将（再次）使用逻辑回归模型。然而，在本章后面，我们将展示一些更复杂的 PySpark ML 模型集中的模型，所以我们加载整个部分：

import pyspark.ml.classification as cl

加载完成后，让我们使用以下代码创建模型：

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

如果我们的目标列的名称是 'label'，我们就不必指定 labelCol 参数。同样，如果我们的特征创建器的输出不叫 'features'，我们将不得不通过（最方便地）在 featuresCreator 对象上调用 getOutputCol() 方法来指定 featuresCol。

创建管道

现在剩下的就是创建一个管道并拟合模型。首先，让我们从 ML 包中加载 Pipeline：

from pyspark.ml import Pipeline

创建一个 Pipeline 非常容易。这是我们管道的概念视图：

将这个结构转换为 Pipeline 是小菜一碟：

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

就是这样！我们的管道现在已经创建好了，所以我们可以（终于！）估计模型了。

拟合模型

在拟合模型之前，我们需要将数据集分割为训练集和测试集。方便的是，DataFrame API 有 .randomSplit(...) 方法：

births_train, births_test = births \
 .randomSplit([0.7, 0.3], seed=666)

第一个参数是要分别进入 births_train 和 births_test 子集的数据集比例列表。seed 参数为随机数生成器提供了一个种子。

现在终于到了运行我们的管道并估计模型的时候了：

model = pipeline.fit(births_train)
test_model = model.transform(births_test)

pipeline 对象的 .fit(...) 方法将我们的训练数据集作为输入。在幕后，births_train 数据集首先传递给编码器对象。在编码器阶段创建的 DataFrame 然后传递给 featuresCreator，后者创建了 'features' 列。最后，这个阶段的输出传递给 logistic 对象，该对象估计最终模型。

.fit(...) 方法返回 PipelineModel 对象（前面的模型对象），然后可以用于预测；我们通过调用 .transform(...) 方法并传递之前创建的测试数据集来获得这个对象。以下是 test_model 在以下命令中的样子：

test_model.take(1)

它生成了以下输出：

如你所见，我们得到了转换器和估计器的所有列。逻辑回归模型输出了几个列：rawPrediction 是特征和 β 系数的线性组合的值，probability 是每个类别计算出的概率，最后，prediction 是我们的最终类别分配。