目录
使用 ML 预测婴儿生存几率
加载数据
创建转换器
创建估计器
创建管道
拟合模型
使用 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 是我们的最终类别分配。
