PySpark将模型预测与未转换的数据对齐：最佳实践

如何解决PySpark将模型预测与未转换的数据对齐：最佳实践

使用PySpark的ML模块，通常会发生以下步骤（在数据清除等之后）：

1. 执行功能和目标转换管道
2. 创建模型
3. 根据模型生成预测
4. 将预测和原始数据集合并在一起，以供业务用户和模型验证之用

---

获取一段简短的代码段：

predictions = model.transform(test_df)

此predictions数据帧将仅包含预测（以及预测的概率和可能的变换）。但是它将 不 包含原始数据集。

如何将预测与原始PySpark DataFrame相结合？

对我来说，我不知道如何将原始数据集（甚至是转换后的test_df）与预测结合起来；没有要加入的共享列，还有adding an index column seems quite tricky for large datasets。

当前解决方案：

对于大型数据集（例如我正在使用的数据集），我尝试了the suggestion here：

test_df = test_df.repartition(predictions.rdd.getNumPartitions())
joined_schema = StructType(test_df.schema.fields + predictions.schema.fields)
interim_rdd = test_df.rdd.zip(predictions.rdd).map(lambda x: x[0] + x[1])
full_data = spark.createDataFrame(interim_rdd,joined_schema)
full_data.write.parquet(my_predictions_path,mode="overwrite")

---

但是我不喜欢这样做有两个原因：

1. 我不确定要保持秩序。该链接表明应该如此，但我不明白为什么。
2. 即使我如上所述强制进行重新分区，有时也会崩溃，并出现以下错误，当我尝试通过上方的最后一行写入数据时：

Caused by: org.apache.spark.SparkException: Can only zip RDDs with same number of elements in each partition

---

我不想使用有时给出的monotonically_increasing_id建议，因为我的数据集太大而无法满足要求。

---

似乎很基本：我如何报告任何模型质量而又无法将预测与原始目标进行比较。别人怎么做？

解决方法

调用model = <your ml-algorithm>.fit(df_train)时，火车数据集可以具有任意数量的附加列。只有包含功能和标签的列将用于训练模型（通常称为features和label，这是可配置的），但可以存在其他列。

下一步，当在训练好的模型上调用predictions = model.transform(df_test)时，将返回一个包含 additional 列prediction，probability和{{1 }}。

尤其是原始要素列和标签列仍是数据框的一部分。此外，属于rawPrediction一部分的 any 列仍在输出中可用，并可用于标识该行。

df_test

打印

prediction = model.transform(df_test)
prediction.printSchema()

如果root |-- feature1: double (nullable = true) |-- feature2: double (nullable = true) |-- feature3: double (nullable = true) |-- label: double (nullable = true) |-- additional_data: string (nullable = true) |-- features: vector (nullable = true) |-- rawPrediction: vector (nullable = true) |-- probability: vector (nullable = true) |-- prediction: double (nullable = false) 不仅包含必需的列df_test，还包含其他列，包括features。通过评估label和label，现在可以例如创建BinaryClassificationMetrics。

从技术上讲，呼叫prediction是Dataset.withColumn call。

---

一个基于Spark docs的ML管道示例的示例：Spark ML工作流通常从包含训练数据，特征和标签（=目标值）的数据帧开始。在此示例中，还存在与ml过程无关的附加列。

model.transform

然后使用变压器将特征组合到单个列中。最简单的转换器是VectorAssembler

training_original = spark.createDataFrame([
    (0.0,1.1,0.1,1.0,'any random value that is not used to train the model'),(2.0,-1.0,0.0,'another value'),1.3,'value 3'),(0.0,1.2,-0.5,'this value is also not used for training nor testing')],["feature1","feature2","feature3","label","additional_data"])

现在可以使用列from pyspark.ml.feature import VectorAssembler assembler = VectorAssembler( inputCols=["feature1","feature3"],outputCol="features") training_transformed = assembler.transform(training_original) #+--------+--------+--------+-----+--------------------+--------------+ #|feature1|feature2|feature3|label| additional_data| features| #+--------+--------+--------+-----+--------------------+--------------+ #| 0.0| 1.1| 0.1| 1.0|any random value ...| [0.0,0.1]| #| ... 和features在此数据帧上训练模型。存在其他列，但label方法将忽略它们。

fit

现在根据测试数据对模型进行测试。准备与训练数据相同：

lr = LogisticRegression(maxIter=10,regParam=0.01)
model = lr.fit(training_transformed)

运行ML魔法会产生

test_df = spark.createDataFrame([
    (-1.0,1.5,'test value 1'),(3.0,2.0,-0.1,'another test value'),2.2,-1.5,'this is not important')],"additional_data"])
test_df_transformed = assembler.transform(test_df)
#+--------+--------+--------+-----+--------------------+--------------+
#|feature1|feature2|feature3|label|     additional_data|      features|
#+--------+--------+--------+-----+--------------------+--------------+
#|    -1.0|     1.5|     1.3|  1.0|        test value 1|[-1.0,1.3]|
#| ...

此数据框现在包含原始输入数据（prediction = model.transform(test_df_transformed) #+--------+--------+--------+-----+--------------------+--------------+--------------------+--------------------+----------+ #|feature1|feature2|feature3|label| additional_data| features| rawPrediction| probability|prediction| #+--------+--------+--------+-----+--------------------+--------------+--------------------+--------------------+----------+ #| -1.0| 1.5| 1.3| 1.0| test value 1|[-1.0,1.3]|[-6.5872014439355...|[0.00137599470692...| 1.0| #| ... 至feature1和feature3），预期目标值（additional_data），变换后的特征（{{1 }}）和模型预测的结果（label）。这是一个数据集中所有输入值，目标值和预测都可用的地方。这里是评估模型并计算模型所需度量的地方。将模型应用于新数据将得到相同的结果（但当然没有features列）。