Spark、ML、StringIndexer：如何处理未知标签

Question

我的目标是构建一个多分类器。

我已经建立了一个特征提取的流程，并且第一步包括使用StringIndexer转换器来将每个类别名称映射到一个标签，该标签将在分类器训练阶段使用。

这个流程被用于对训练集进行拟合。

测试集必须经过拟合后的流程处理，以便提取相同的特征向量。

考虑到我的测试集文件与训练集具有相同的结构。可能的情况是在测试集中遇到未知的类别名称，在这种情况下，StringIndexer将无法找到标签，并引发异常。

是否有解决方案？或者我们如何避免这种情况发生？

- Rami

---

请重新接受@queise的答案。它比已添加为解决方案的那个好得多。 - lu5er

Answer 1

在Spark 2.2（于2017年发布）中，您可以在创建索引器时使用.setHandleInvalid("keep")选项。使用此选项，当索引器看到新的标签时，它会添加新的索引。

val categoryIndexerModel = new StringIndexer()
  .setInputCol("category")
  .setOutputCol("indexedCategory")
  .setHandleInvalid("keep") // options are "keep", "error" or "skip"

根据文档，在将StringIndexer拟合于一个数据集并用于转换另一个数据集时，有三种处理未知标签的策略:

• 'error'：抛出异常（这是默认值）
• 'skip'：完全跳过包含未知标签的行（从输出中删除这些行！）
• 'keep'：将未知标签放入特殊的附加桶中，索引为numLabels

请参见链接的文档，了解StringIndexer不同选项的输出示例。

- queise

---

您能否提供有关保留，错误，跳过选项之间确切差异的更多详细信息。 跳过是删除数据点吗？ 错误会中断模型吗？而保留则会添加新列？这是否适用于两种情况，即测试集中未在训练集中看到的值和训练集中未在测试集中看到的值？ - Chuck

Answer 2

在Spark 1.6中可以绕过此问题。

这是jira: https://issues.apache.org/jira/browse/SPARK-8764

以下是一个示例:

val categoryIndexerModel = new StringIndexer()
  .setInputCol("category")
  .setOutputCol("indexedCategory")
  .setHandleInvalid("skip") // new method.  values are "error" or "skip"

我开始使用这个，但最终回到了KrisP的第二个要点，即将特定的估算器拟合到完整数据集上。

在将IndexToString转换时，您稍后需要使用它作为一部分管道。

这是修改后的示例：

val categoryIndexerModel = new StringIndexer()
  .setInputCol("category")
  .setOutputCol("indexedCategory")
  .fit(itemsDF) // Fit the Estimator and create a Model (Transformer)

... do some kind of classification ...

val categoryReverseIndexer = new IndexToString()
  .setInputCol(classifier.getPredictionCol)
  .setOutputCol("predictedCategory")
  .setLabels(categoryIndexerModel.labels) // Use the labels from the Model

- Chris Fregly

---

当你尝试将模型应用于新数据时会发生什么？你可能会发现某些列中有新的值，这些值在原始测试或训练数据中并不存在。如果使用setHandleInvalid("skip")，整行数据会被丢弃，而你可能只想忽略先前未见过的值，但仍然使用行中的其他值。 - user1933178

这意味着新的未见过的数据不属于训练数据的同一统计分布。在大多数应用中，不给它一个标签是可以的，事实上这是可取的，因为它看起来不像模型以前见过的任何数据。在统计意义上，“看起来像”显然是我们试图进行学习而不是记忆。 - Kai

Spark 2.2 新增了一个 .setHandleInvalid("keep") 选项，当处理新数据时，它将添加新的索引。我认为这个功能非常有用，因为希望之后应用的预测模型将利用所有其他变量输出有效的预测结果（当然，新的索引没有任何预测能力）。 - queise

Answer 3

很抱歉，没有更好的方法。你可以选择：

• 在应用 StringIndexer 之前过滤掉具有未知标签的测试示例。
• 将 StringIndexer 拟合于训练和测试数据框的联合体中，以确保所有标签都存在。
• 将具有未知标签的测试样例转换为已知标签。

这里是一些执行上述操作的示例代码：

// get training labels from original train dataframe
val trainlabels = traindf.select(colname).distinct.map(_.getString(0)).collect  //Array[String]
// or get labels from a trained StringIndexer model
val trainlabels = simodel.labels 

// define an UDF on your dataframe that will be used for filtering
val filterudf = udf { label:String => trainlabels.contains(label)}

// filter out the bad examples 
val filteredTestdf = testdf.filter( filterudf(testdf(colname)))

// transform unknown value to some value, say "a"
val mapudf = udf { label:String => if (trainlabels.contains(label)) label else "a"}

// add a new column to testdf: 
val transformedTestdf = testdf.withColumn( "newcol", mapudf(testdf(colname)))

- KrisP

---

有没有一种方法可以提供没有任何标签的测试数据，以便算法可以从头开始预测。在我的情况下，我没有任何测试数据项的标签。请参见：https://stackoverflow.com/questions/44127634/providing-test-data-items-with-empty-labels-in-spark-random-forest-classifier在我的情况下，我是否必须为项目分配随机标签？ - suat

@queise使用Spark 2.2的答案现在是最佳答案。 - mrjrdnthms

Answer 4

在我的情况下，我在大数据集上运行spark ALS，并且数据并不完全在所有分区中可用，因此我必须适当地缓存数据(cache())，这样它就能够很好地工作了。

- Suresh Gorakala

Answer 5

对我来说，通过设置一个参数（https://issues.apache.org/jira/browse/SPARK-8764）完全忽略行是不太可行的解决方法。

最终，我创建了自己的CustomStringIndexer转换器，它将为所有在训练期间未遇到的新字符串分配一个新值。您也可以通过更改Spark特征代码的相关部分（只需删除明确检查此内容的if条件并使其返回数组的长度）来实现此目的，并重新编译jar。

这并不是一个简单的修复，但肯定是一个修复。

我记得在JIRA中看到了一个错误，也要将其纳入其中：https://issues.apache.org/jira/browse/SPARK-17498 虽然它被设置为与Spark 2.2一起发布，但只能等待了 :S

- Subramaniam Ramasubramanian