创建一个映射以为Spark Dataframe的每一行调用POJO

Question

我在R中构建了一个H2O模型并保存了POJO代码。我想使用POJO在hdfs中对镶木地板文件进行评分，但我不知道如何去做。我计划将镶木地板文件读入spark（scala / SparkR / PySpark）并在那里进行评分。以下是我在H2O's documentation page.上发现的摘录

“如何在Spark Cluster上运行POJO？

POJO提供了进行预测的数学逻辑，因此您不会在那里找到任何Spark（甚至H2O）特定代码。如果要使用POJO对Spark中的数据集进行预测，请创建一个映射以为每行调用POJO并将结果保存到新列中，逐行“

有没有人有一些示例代码，我怎么能这样做？我非常感谢任何帮助。我主要在R和SparkR中编码，我不知道如何将POJO“映射”到每一行。

提前致谢。

Answer 1

我刚发布了一个实际上使用DataFrame / Dataset的solution。该帖子使用星球大战数据集在R中构建模型，然后在Spark中的测试集上评分MOJO。我会在这里粘贴唯一的相关部分：

Scoring with Spark (and Scala)

你可以使用spark-submit或spark-shell。如果使用spark-submit，则需要将h2o-genmodel.jar放在spark应用程序根目录的lib文件夹下，以便在编译期间将其添加为依赖项。以下代码假定您正在运行spark-shell。为了使用h2o-genmodel.jar，你需要通过提供--jar标志来启动spark-shell时附加jar文件。例如：

/usr/lib/spark/bin/spark-shell \
--conf spark.serializer="org.apache.spark.serializer.KryoSerializer" \
--conf spark.driver.memory="3g" \
--conf spark.executor.memory="10g" \
--conf spark.executor.instances=10 \
--conf spark.executor.cores=4 \
--jars /path/to/h2o-genmodel.jar

现在在Spark shell中，导入依赖项

import _root_.hex.genmodel.easy.{EasyPredictModelWrapper, RowData}
import _root_.hex.genmodel.MojoModel

Using DataFrame

val modelPath = "/path/to/zip/file"
val dataPath = "/path/to/test/data"

// Import data
val dfStarWars = spark.read.option("header", "true").csv(dataPath)
// Import MOJO model
val mojo = MojoModel.load(modelPath)
val easyModel = new EasyPredictModelWrapper(mojo)

// score
val dfScore = dfStarWars.map {
  x =>
    val r = new RowData
    r.put("height", x.getAs[String](1))
    r.put("mass", x.getAs[String](2))
    val score = easyModel.predictBinomial(r).classProbabilities
    (x.getAs[String](0), score(1))
}.toDF("name", "isHumanScore")

可变分数是级别0和1的两个分数的列表。分数（1）是级别1的分数，其是“人”。默认情况下，map函数返回一个带有未指定列名“_1”，“_ 2”等的DataFrame。您可以通过调用toDF来重命名列。

Using Dataset

要使用数据集API，我们只需要创建两个案例类，一个用于输入数据，另一个用于输出。

case class StarWars (
  name: String,
  height: String,
  mass: String,
  is_human: String
)

case class Score (
  name: String,
  isHumanScore: Double
)


// Dataset
val dtStarWars = dfStarWars.as[StarWars]
val dtScore = dtStarWars.map {
  x =>
    val r = new RowData
    r.put("height", x.height)
    r.put("mass", x.mass)
    val score = easyModel.predictBinomial(r).classProbabilities
    Score(x.name, score(1))
}

使用数据集，您可以通过直接调用x.columnName来获取列的值。请注意，列值的类型必须是String，因此如果它们是case类中定义的其他类型，则可能需要手动转换它们。

Answer 2

如果你想在spark中使用POJO或MOJO进行评分，你应该使用在h2o-genmodel.jar类中提供的RowData作为逐行输入数据来调用easyPredict方法来生成分数。

您的解决方案是从HDFS读取镶木地板文件，然后为每一行，通过填充每个条目将其转换为RowData对象，然后将其传递给您的POJO评分函数。记住POJO和MOJO他们都使用完全相同的评分函数来评分，唯一的区别在于如何使用POJO类与使用MOJO资源zip包。由于MOJO是向后兼容的，并且可以与任何更新的h2o-genmodel.jar一起使用，因此最好使用MOJO而不是POJO。

以下是您可以在Spark上使用的完整Scala代码来加载MOJO模型然后进行评分：

import _root_.hex.genmodel.GenModel
import _root_.hex.genmodel.easy.{EasyPredictModelWrapper, RowData}
import _root_.hex.genmodel.easy.prediction
import _root_.hex.genmodel.MojoModel
import _root_.hex.genmodel.easy.RowData

// Load Mojo
val mojo = MojoModel.load("/Users/avkashchauhan/learn/customers/mojo_bin/gbm_model.zip")
val easyModel = new EasyPredictModelWrapper(mojo)

// Get Mojo Details
var features = mojo.getNames.toBuffer

// Creating the row
val r = new RowData
r.put("AGE", "68")
r.put("RACE", "2")
r.put("DCAPS", "2")
r.put("VOL", "0")
r.put("GLEASON", "6")

// Performing the Prediction
val prediction = easyModel.predictBinomial(r).classProbabilities 

Here是一个在Spark中读取镶木地板文件然后另存为CSV的示例。您可以使用相同的代码从HDFS读取镶木地板，然后将每行作为RowData传递给上面的示例。

Here是在spark中使用MOJO模型并使用RowData执行评分的详细示例。