重写 Apache Spark 管道以使用现有模型

Question

我有一个管道（参见 pipelineBefore 方法）：

• 预处理数据
• 训练模型
• 获取预测

然后我委托模型训练，现在只需要预处理数据并获得预测结果。查看管道之后

如何重构代码以通过 Pipeline API 使用现有模型，而不是手动调用转换器？

澄清。我需要集成一个普通模型，例如 org.apache.spark.ml.classification.LogisticRegression，而不是以前训练的 org.apache.spark.ml.PipelineModel

    private def pipelineBefore: org.apache.spark.sql.DataFrame = {
      val training = spark.createDataFrame(Seq(
        (0L, "a b c d e spark", 1.0),
        (1L, "b d", 0.0),
        (2L, "spark f g h", 1.0),
        (3L, "hadoop mapreduce", 0.0)
      )).toDF("id", "text", "label")
      println("Pipeline example. Training dataframe before preprocessing")
      training.show()
      // Configure an ML pipeline, which consists of three stages: tokenizer, hashingTF, and lr.
      val tokenizer = new Tokenizer()
        .setInputCol("text")
        .setOutputCol("words")
      val hashingTF = new HashingTF()
        .setNumFeatures(1000)
        .setInputCol(tokenizer.getOutputCol)
        .setOutputCol("features")
      val lr = new LogisticRegression()
        .setMaxIter(10)
        .setRegParam(0.001)
      val pipeline = new Pipeline()
        .setStages(Array(tokenizer, hashingTF, lr))
      // Fit the pipeline to training documents.
      val model = pipeline.fit(training)
      // Prepare test documents, which are unlabeled (id, text) tuples.
      val test = spark.createDataFrame(Seq(
        (4L, "spark i j k"),
        (5L, "l m n"),
        (6L, "spark hadoop spark"),
        (7L, "apache hadoop")
      )).toDF("id", "text")
      // Make predictions on test documents.
      val predictionResult = model.transform(test)
      println("Pipeline example. Prediction result")
      predictionResult.show()
      return predictionResult
    }

    private def pipelineAfter: org.apache.spark.sql.DataFrame = {
      // Given a valid model trained on a preprocessed DataFrame
      val trainedModel = getTrainedModel()
      // Preprocess a test dataset
      val test = spark.createDataFrame(Seq(
        (4L, "spark i j k"),
        (5L, "l m n"),
        (6L, "spark hadoop spark"),
        (7L, "apache hadoop")
      )).toDF("id", "text")
      //HOW TO ADOPT A PIPELINE API HERE ?
      val tokenizer = new Tokenizer()
        .setInputCol("text")
        .setOutputCol("words")
      val hashingTF = new HashingTF()
        .setNumFeatures(1000)
        .setInputCol(tokenizer.getOutputCol)
        .setOutputCol("features")
      val tokenizedTestData = tokenizer.transform(test)
      val hashedTestData = hashingTF.transform(tokenizedTestData)
      println("Preprocessed test data")
      hashedTestData.show()
      // Make predictions on the test dataset.
      val predictionResult = trainedModel.transform(hashedTestData)
      println("Prediction result")
      predictionResult.show()
      return predictionResult
    }

Answer 1

如果您想将后者与另一个模型一起使用，您需要序列化您的管道。在您的示例中：

private def pipelineBefore: org.apache.spark.sql.DataFrame = {
    val training = spark.createDataFrame(Seq(
      (0L, "a b c d e spark", 1.0),
      (1L, "b d", 0.0),
      (2L, "spark f g h", 1.0),
      (3L, "hadoop mapreduce", 0.0)
    )).toDF("id", "text", "label")
    println("Pipeline example. Training dataframe before preprocessing")
    training.show()
    // Configure an ML pipeline, which consists of three stages: tokenizer, hashingTF, and lr.
    val tokenizer = new Tokenizer()
      .setInputCol("text")
      .setOutputCol("words")
    val hashingTF = new HashingTF()
      .setNumFeatures(1000)
      .setInputCol(tokenizer.getOutputCol)
      .setOutputCol("features")
    val lr = new LogisticRegression()
      .setMaxIter(10)
      .setRegParam(0.001)
    val pipeline = new Pipeline()
      .setStages(Array(tokenizer, hashingTF, lr))
    // Fit the pipeline to training documents.


    // Save your pipeline transformations
    pipeline.write.overwrite().save("/tmp/path")

    // ....
}

那你需要加载：

  private def pipelineAfter: org.apache.spark.sql.DataFrame = {
    // Given a valid model trained, for example a LR model
    // You can use pipeline model to load your model too
    val trainedModel : LogisticRegressionModel = ???
    // val trainedModel = PipelineModel.load("path_to_your_model")

    // Preprocess a test dataset
    val test = spark.createDataFrame(Seq(
      (4L, "spark i j k"),
      (5L, "l m n"),
      (6L, "spark hadoop spark"),
      (7L, "apache hadoop")
    )).toDF("id", "text")
    //HOW TO ADOPT A PIPELINE API HERE ?

    // Path where you stored the transform pipeline
    val transformPipeline = PipelineModel.load("/tmp/path")
    val hashedTestData = transformPipeline.transform(test)

    // Make predictions on the test dataset.
    val predictionResult = trainedModel.transform(hashedTestData)
    println("Prediction result")
    predictionResult.show()
    return predictionResult
  }

查看 Spark doc 了解更多详情。