PySpark Mllib 预测 DataFrame 中的所有行答案

【问题标题】：PySpark Mllib Predict all rows in DataFramePySpark Mllib 预测 DataFrame 中的所有行
【发布时间】：2016-12-14 23:31:37
【问题描述】：

我正在使用 Spark Streaming 从 Kafka 获取批量 JSON 读数。生成的批次从 RDD 转换为数据帧。

我的目标是对此数据帧的每一行进行分类，因此我使用 VectorAssembler 创建将传递给模型的特征：

sqlContext = SQLContext(rdd.context)
rawReading = sqlContext.jsonRDD(rdd)
sensorReadings = rawReading.selectExpr("actual.y AS yaw","actual.p AS pitch", "actual.r AS roll")
assembler = VectorAssembler(
        inputCols=["yaw", "pitch", "roll"],
        outputCol="features")
sensorReadingsFinal = assembler.transform(sensorReadings)
sensorReadingsFinal.show()
+---+-----+----+-----------------+
|yaw|pitch|roll|         features|
+---+-----+----+-----------------+
| 18| 17.5| 120|[18.0,17.5,120.0]|
| 18| 17.5| 120|[18.0,17.5,120.0]|
| 18| 17.5| 120|[18.0,17.5,120.0]|
| 18| 17.5| 120|[18.0,17.5,120.0]|
| 18| 17.5| 120|[18.0,17.5,120.0]|
+---+-----+----+-----------------+

我有一个我之前训练过的随机森林模型。

loadedModel = RandomForestModel.load(sc, "MyRandomForest.model")

我的问题是，在将整个数据插入数据库之前，如何对数据框中的每一行进行预测？

我最初正在考虑做这样的事情......

prediction = loadedModel.predict(sensorReadings.features)

但我意识到，由于数据框有多行，我需要以某种方式添加一列并逐行进行预测。也许我在这一切都错了？

我想要的最终数据框是这样的：

+---+-----+----+-----------------+
|yaw|pitch|roll|       Prediction|
+---+-----+----+-----------------+
| 18| 17.5| 120|              1  |
| 18| 17.5| 120|              1  |
| 18| 17.5| 120|              1  |
| 18| 17.5| 120|              1  |
| 18| 17.5| 120|              1  |
+---+-----+----+-----------------+

此时我会将其保存到数据库中：

sensorReadingsFinal.write.jdbc("jdbc:mysql://localhost/testdb", "SensorReadings", properties=connectionProperties)

【问题讨论】：

标签： python-2.7 apache-spark spark-streaming spark-dataframe apache-spark-mllib

【解决方案1】：

在 Spark ML 中，从数据中获取预测的方法称为transform，所以我猜你正在寻找这个：

prediction = loadedModel.transform(sensorReadings)

【讨论】：

【解决方案2】：

这是我最终解决此问题的方法：

# Convert DStream RDD's to DataFrame and run SQL query
sqlContext = SQLContext(rdd.context)
if rdd.isEmpty() == False:
    rawReading = sqlContext.jsonRDD(rdd)
    sensorReadings = rawReading.selectExpr("actual.y AS yaw","actual.p AS pitch", "actual.r AS roll")
    assembler = VectorAssembler(
                inputCols=["yaw","pitch","roll"], # Must be in same order as what was used to train the model.  Testing using only pitch since model has limited dataset.
                outputCol="features")
    sensorReadings = assembler.transform(sensorReadings)

    # Create a new dataFrame of predictions and readingID
    predictions = loadedModel.predict(sensorReadings.map(lambda x: x.features))
    predictionsDF = sensorReadings.map(lambda x: x.readingID).zip(predictions).toDF(["readingID","positionID"])
    
    # Join the prediction dataFrame back to the sensorReadings dataFrame.  Drop the duplicate readingID column.
    combinedDF = sensorReadings.join(predictionsDF, sensorReadings.readingID == predictionsDF.readingID).drop(predictionsDF.readingID)
    
    # Drop the feature vector column
    combinedDF = combinedDF.drop("features")
    
    combinedDF.show()

我基本上创建了一个仅包含特征向量和 readingID 的新数据帧，然后将其加入到原始数据帧中。

这可能不是最优雅的解决方案，所以如果有人可以提出更好的建议，请这样做。

【讨论】：