为什么我的 DecisionTreeClassifier 模型在预测时抱怨 labelCol 不存在？

Question

我开始编写用于对一系列文档中的段落进行分类的 ML 模型。我写了我的模型，结果看起来很棒！但是，当我尝试提供不包含 labelCol 的 CSV（即标记列，我要预测的列）时，它会引发错误！ '字段 tagIndexed 不存在。'

所以这很奇怪。我要预测的是“tag”列，那么当我调用model.transform(df)（在Predict.scala 中）时，为什么它会期望“tagIndexed”列？我对 ML 没有经验，但所有 DecisionTreeClassifiers 往往在测试数据中不存在 labelCol。我在这里错过了什么？

我创建了模型，使用测试数据对其进行了验证，并将其保存到磁盘。然后，在另一个 Scala 对象中，我加载模型并将我的 csv 传递给它。

//Train.scala    
package com.secret.classifier
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.DecisionTreeClassifier
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.sql.Column
import org.apache.spark.ml.feature.{HashingTF, IDF, StringIndexer, Tokenizer, VectorAssembler, Word2Vec}
import org.apache.spark.ml.regression.LinearRegression
import org.apache.spark.ml.tuning.{ParamGridBuilder, TrainValidationSplit}
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.types
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

...

val colSeq = Seq("font", "tag")
val indexSeq = colSeq.map(col => new StringIndexer().setInputCol(col).setOutputCol(col+"Indexed").fit(dfNoNan))

val tokenizer = new Tokenizer().setInputCol("soup").setOutputCol("words")
//val wordsData = tokenizer.transform(dfNoNan)

val hashingTF = new HashingTF()
.setInputCol(tokenizer.getOutputCol)
.setOutputCol("rawFeatures")
.setNumFeatures(20)

val featuresCol = "features"
val assembler = new VectorAssembler()
.setInputCols((numericCols ++ colSeq.map(_+"Indexed")).toArray)
.setOutputCol(featuresCol)

val labelCol = "tagIndexed"
val decisionTree = new DecisionTreeClassifier()
.setLabelCol(labelCol)
.setFeaturesCol(featuresCol)

val pipeline = new Pipeline().setStages((indexSeq :+ tokenizer :+ hashingTF :+ assembler :+ decisionTree).toArray)

val Array(training, test) = dfNoNan.randomSplit(Array(0.8, 0.2), seed=420420)

val model = pipeline.fit(training)


model.write.overwrite().save("tmp/spark-model")

//Predict.scala
package com.secret.classifier
import org.apache.spark.sql.functions._
import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.DecisionTreeClassifier
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.sql.Column
import org.apache.spark.ml.feature.{HashingTF, IDF, StringIndexer, Tokenizer, VectorAssembler, Word2Vec}
import org.apache.spark.ml.regression.LinearRegression
import org.apache.spark.ml.tuning.{ParamGridBuilder, TrainValidationSplit}
import org.apache.spark.sql.types
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

...

  val dfImport = spark.read
  .format("csv")
  .option("header", "true")
  //.option("mode", "DROPMALFORMED")
  .schema(customSchema)
  .load(csvLocation)

val df = dfImport.drop("_c0", "doc_name")
df.show(20)

val model = PipelineModel.load("tmp/spark-model")

val predictions = model.transform(df)

predictions.show(20)


//pom.xml -> Spark/Scala specific dependencies
<properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <encoding>UTF-8</encoding>
    <scala.version>2.11.12</scala.version>
    <scala.compat.version>2.11</scala.compat.version>
    <spec2.version>4.2.0</spec2.version>
</properties>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.3.1</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/com.databricks/spark-csv -->
    <dependency>
        <groupId>com.databricks</groupId>
        <artifactId>spark-csv_2.11</artifactId>
        <version>1.5.0</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
        <version>2.3.1</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.3.1</version>
    </dependency>

    <dependency>
        <groupId>com.univocity</groupId>
        <artifactId>univocity-parsers</artifactId>
        <version>2.8.0</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-mllib -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-mllib_2.11</artifactId>
        <version>2.3.1</version>
    </dependency>
</dependencies>

预期结果是预测模型不会引发错误。相反，它会抛出错误“字段“tagIndexed”不存在。”

Answer 1

看起来您已经在特征中包含了标签字段，因为它位于 colSeq 列输出中。在这一步中，您只想包含特征列：

.setInputCols((numericCols ++ colSeq.map(_+"Indexed")).toArray)

我发现使用 .filterNot() 函数很有帮助。