管道中的火花訪問估計器

Question

與Is it possible to access estimator attributes in spark.ml pipelines?類似我想訪問估計器，例如管道中的最後一個元素。

這裏提到的方法似乎不再適用於spark 2.0.1。它現在如何工作？

編輯

也許我應該解釋多一點點細節： 這裏是我的估計+矢量彙編：

val numRound = 20 
val numWorkers = 4 
val xgbBaseParams = Map(
    "max_depth" -> 10, 
    "eta" -> 0.1, 
    "seed" -> 50, 
    "silent" -> 1, 
    "objective" -> "binary:logistic" 
) 

val xgbEstimator = new XGBoostEstimator(xgbBaseParams) 
    .setFeaturesCol("features") 
    .setLabelCol("label") 

val vectorAssembler = new VectorAssembler() 
    .setInputCols(train.columns 
     .filter(!_.contains("label"))) 
    .setOutputCol("features") 

    val simplePipeParams = new ParamGridBuilder() 
    .addGrid(xgbEstimator.round, Array(numRound)) 
    .addGrid(xgbEstimator.nWorkers, Array(numWorkers)) 
    .build() 

    val simplPipe = new Pipeline() 
    .setStages(Array(vectorAssembler, xgbEstimator)) 

    val numberOfFolds = 2 
    val cv = new CrossValidator() 
    .setEstimator(simplPipe) 
    .setEvaluator(new BinaryClassificationEvaluator() 
     .setLabelCol("label") 
     .setRawPredictionCol("prediction")) 
    .setEstimatorParamMaps(simplePipeParams) 
    .setNumFolds(numberOfFolds) 
    .setSeed(gSeed) 

    val cvModel = cv.fit(train) 
    val trainPerformance = cvModel.transform(train) 
    val testPerformance = cvModel.transform(test) 

現在我想例如執行自定義得分!= 0.5截止點。這是可能的，如果我弄到模型：

val realModel = cvModel.bestModel.asInstanceOf[XGBoostClassificationModel] 

但這一步這裏不編譯。 感謝您的建議，我可以得到模型：

val pipelineModel: Option[PipelineModel] = cvModel.bestModel match { 
    case p: PipelineModel => Some(p) 
    case _ => None 
    } 

    val realModel: Option[XGBoostClassificationModel] = pipelineModel 
    .flatMap { 
     _.stages.collect { case t: XGBoostClassificationModel => t } 
     .headOption 
    } 
    // TODO write it nicer 
    val measureResults = realModel.map { 
    rm => 
     { 
     for (
      thresholds <- Array(Array(0.2, 0.8), Array(0.3, 0.7), Array(0.4, 0.6), 
      Array(0.6, 0.4), Array(0.7, 0.3), Array(0.8, 0.2)) 
     ) { 
      rm.setThresholds(thresholds) 

      val predResult = rm.transform(test) 
      .select("label", "probabilities", "prediction") 
      .as[LabelledEvaluation] 
      println("cutoff was ", thresholds) 
      calculateEvaluation(R, predResult) 
     } 
     } 
    } 

然而，問題是，

val predResult = rm.transform(test) 

會train失敗不包含vectorAssembler的功能列。 此列僅在完整管道運行時創建。

所以我決定創建第二個管道：

val scoringPipe = new Pipeline() 
      .setStages(Array(vectorAssembler, rm)) 
val predResult = scoringPipe.fit(train).transform(test) 

，但似乎有點笨拙。你有更好/更好的想法嗎？

Answer 1

Spark 2.0.0中沒有任何改變，並且相同的方法有效。 Example Pipeline：

import org.apache.spark.ml.{Pipeline, PipelineModel} 
import org.apache.spark.ml.classification.LogisticRegression 
import org.apache.spark.ml.feature.{HashingTF, Tokenizer} 
import org.apache.spark.ml.linalg.Vector 
import org.apache.spark.sql.Row 

// Prepare training documents from a list of (id, text, label) tuples. 
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") 

// 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.01) 
val pipeline = new Pipeline() 
    .setStages(Array(tokenizer, hashingTF, lr)) 

// Fit the pipeline to training documents. 
val model = pipeline.fit(training) 

和型號：

val logRegModel = model.stages.last 
    .asInstanceOf[org.apache.spark.ml.classification.LogisticRegressionModel]