Spark MLib 每周一算法 —— OneHotEncoder

2019-03-23

OneHotEncoder

OneHotEncoder是一个Estimator，用来将某个类别index属性(整数值)的值转化成向量值，该向量值的特点是每个分量不是0就是1，并且最多只有1个分量是1，这个分量是1即代表这条数据的这个特性取得特定的属性值，所以被称作One-Hot(独热)。这样的向量可以将离散分类数据变成可用于回归类算法的连续数据，特征的index属性值通常可以由之前介绍的StringIndexer从文本label值来生成。该算法通常被用在自然语言处理(nlp)任务中，对词典(vocabulary)数据中的每个单词(word)进行编码。

编码规则

Spark的OneHotEncoder的有两种编码规则如下，包含原点(0向量)和不包含原点：

包含原点(0向量)——向量长度为训练数据中最大的index值，该最大index值对应的编码为全0向量，其他index值对应的编码为——在index对应的分量上值为1，其他分量值为0的向量。
不包含原点——向量长度为训练数据中最大的index值+1，index值对应的编码为——在index对应的分量上值为1，其他分量值为0的向量。

参数

inputCols: 需要进行编码的一组属性名称，数组类型。
outputCols: 编码后的属性名称，数组类型，次序和上面的inputCols一一对应。
dropLast: 表示结果编码中是否包含原点(0向量)，具体规则见上面编码规则，true表示包含。(布尔类型，默认值: true)
handleInvalid: 表示在transform期间如何处理在fit训练集时没有碰到过的属性值或者非index(>=0的整数)值。 (默认值: error)
可选项1. error-抛出异常org.apache.spark.SparkException: Failed to execute user defined function；
可选项2. keep-一旦需要transform的数据集里出现非法index值，所有的最终向量的长度都会+1，非法index值的处理也同样取决于dropLast参数，非法index值会被转化为全0向量或者最后位为1的one-hot向量。

例子

val training = spark.createDataFrame(
  Seq((2.0, 1.0),
    (1.0, 3.0),
    (1.0, 1.0),
    (2.0, 2.0),
    (2.0, 1.0),
    (1.0, 100.0))
).toDF("categoryIndex1", "categoryIndex2")
training.show()
//    +--------------+--------------+
//    |categoryIndex1|categoryIndex2|
//    +--------------+--------------+
//    |           2.0|           1.0|
//    |           1.0|           3.0|
//    |           1.0|           1.0|
//    |           2.0|           2.0|
//    |           2.0|           1.0|
//    |           1.0|         100.0|
//    +--------------+--------------+


val encoder = new OneHotEncoderEstimator()
  .setInputCols(Array("categoryIndex1", "categoryIndex2"))
  .setOutputCols(Array("categoryVec1", "categoryVec2"))

val model = encoder.fit(training)

model.transform(training).show()
//    +--------------+--------------+-------------+---------------+
//    |categoryIndex1|categoryIndex2| categoryVec1|   categoryVec2|
//    +--------------+--------------+-------------+---------------+
//    |           2.0|           1.0|    (2,[],[])|(100,[1],[1.0])|
//    |           1.0|           3.0|(2,[1],[1.0])|(100,[3],[1.0])|
//    |           1.0|           1.0|(2,[1],[1.0])|(100,[1],[1.0])|
//    |           2.0|           2.0|    (2,[],[])|(100,[2],[1.0])|
//    |           2.0|           1.0|    (2,[],[])|(100,[1],[1.0])|
//    |           1.0|         100.0|(2,[1],[1.0])|    (100,[],[])|
//    +--------------+--------------+-------------+---------------+

val test = spark.createDataFrame(
  Seq((0, 104), (-1, 0), (2, -100), (1, 101))
).toDF("categoryIndex1", "categoryIndex2")

try {
  model.transform(test).show()
} catch {  //error from -1 in categoryIndex1 & -100/101/104 in categoryIndex2.
  case ex: SparkException => println(ex)
}

model.setHandleInvalid("keep")
model.transform(test).show()
//    +--------------+--------------+-------------+---------------+
//    |categoryIndex1|categoryIndex2| categoryVec1|   categoryVec2|
//    +--------------+--------------+-------------+---------------+
//    |             0|           104|(3,[0],[1.0])|    (101,[],[])|
//    |            -1|             0|    (3,[],[])|(101,[0],[1.0])|
//    |             2|          -100|(3,[2],[1.0])|    (101,[],[])|
//    |             1|           101|(3,[1],[1.0])|    (101,[],[])|
//    +--------------+--------------+-------------+---------------+