如何在 Apache Spark 中编码分类特征答案

【问题标题】：How to encode categorical features in Apache Spark如何在 Apache Spark 中编码分类特征
【发布时间】：2015-10-30 14:00:27
【问题描述】：

我有一组数据，我想根据这些数据创建分类模型。每一行的格式如下：

user1,class1,product1
user1,class1,product2
user1,class1,product5
user2,class1,product2
user2,class1,product5
user3,class2,product1

大约有 100 万用户、2 个类和 100 万种产品。我接下来想做的是创建稀疏向量（MLlib 已经支持的东西）但是为了应用该函数，我必须首先创建密集向量（带有 0）。换句话说，我必须对我的数据进行二值化。最简单（或最优雅）的方法是什么？

鉴于我是 MLlib 的新手，我可以请你提供一个具体的例子吗？我正在使用 MLlib 1.2。

编辑

我已经完成了以下代码，但结果确实很慢...是否提供了我只能使用 MLlib 1.2 的其他想法？

val data = test11.map(x=> ((x(0) , x(1)) , x(2))).groupByKey().map(x=> (x._1 , x._2.toArray)).map{x=>
  var lt : Array[Double] = new Array[Double](test12.size)
  val id = x._1._1
  val cl = x._1._2
  val dt = x._2
  var i = -1
  test12.foreach{y => i += 1; lt(i) = if(dt contains y) 1.0 else 0.0}
  val vs = Vectors.dense(lt)
  (id , cl , vs)
}

【问题讨论】：

能否举例说明您希望密集向量输出对于该输入的外观？
你到底想做什么分类？即如果userX 和classY 那么很可能是productZ 或其他什么？
并非如此。我将使用二进制分类，其中userX 是值的稀疏向量，classY 是相应的类。
@user706838 是 userX 一个实际的对象还是一个非常简单的字符串？您在分类过程中是否以任何方式考虑product？
我想我在这里提出的问题对于以前从事过机器学习工作的人来说是非常直截了当的。我只是想找出在 MLlib 中实现这一点的最佳方法。在这里查看 scikit-learn 中的类似示例：scikit-learn.org/stable/modules/…

标签： scala apache-spark apache-spark-mllib apache-spark-1.2

【解决方案1】：

您可以使用 spark.ml 的OneHotEncoder。

你第一次使用：

OneHotEncoder.categories(rdd, categoricalFields)

其中categoricalField 是您的RDD 包含分类数据的索引序列。 categories，给定一个数据集和作为分类变量的列的索引，返回一个结构，对于每个字段，该结构描述了数据集中存在的值。该映射旨在用作编码方法的输入：

OneHotEncoder.encode(rdd, categories)

返回矢量化的RDD[Array[T]]。

【讨论】：

在 MLlib 1.2 中不可用 :-)
是的，不是吗，很遗憾我无法更新...请查看我的编辑。
这在 1.4 中似乎不可用。

【解决方案2】：

如果使用内置的OneHotEncoder 不是一个选项，并且您只有一个变量来实现穷人的 one-hot 或多或少是直截了当的。首先让我们创建一个示例数据：

import org.apache.spark.mllib.linalg.{Vector, Vectors}

val rdd = sc.parallelize(List(
    Array("user1", "class1", "product1"),
    Array("user1", "class1", "product2"),
    Array("user1", "class1", "product5"),
    Array("user2", "class1", "product2"),
    Array("user2", "class1", "product5"),
    Array("user3", "class2", "product1")))

接下来我们必须创建一个从值到索引的映射：

val prodMap = sc.broadcast(rdd.map(_(2)).distinct.zipWithIndex.collectAsMap)

还有一个简单的编码函数：

def encodeProducts(products: Iterable[String]): Vector =  {
    Vectors.sparse(
        prodMap.value.size,
        products.map(product => (prodMap.value(product).toInt, 1.0)).toSeq
    )
}

最后我们可以将它应用到数据集上：

rdd.map(x => ((x(0), x(1)), x(2))).groupByKey.mapValues(encodeProducts)

上面扩展处理多个变量相对容易。

编辑：

如果产品数量太大而无法使用广播，则应该可以改用join。首先，我们可以创建从产品到索引的类似映射，但将其保留为 RDD：

import org.apache.spark.HashPartitioner

val nPartitions = ???

val prodMapRDD = rdd
     .map(_(2))
     .distinct
     .zipWithIndex
     .partitionBy(new HashPartitioner(nPartitions))
     .cache

val nProducts = prodMapRDD.count // Should be < Int.MaxValue

接下来我们重塑输入 RDD 以获取按产品索引的 PairRDD：

val pairs = rdd
    .map(rec => (rec(2), (rec(0), rec(1))))
    .partitionBy(new HashPartitioner(nPartitions))

我们终于可以join两者了

def indicesToVec(n: Int)(indices: Iterable[Long]): Vector = {
     Vectors.sparse(n, indices.map(x => (x.toInt, 1.0)).toSeq)
}

pairs.join(prodMapRDD)
   .values
   .groupByKey
   .mapValues(indicesToVec(nProducts.toInt))

【讨论】：

+1 表示通用解决方案。您还有其他不使用broadcast 的解决方案吗？我使用像你这样的解决方案，但有时这不起作用，因为 prodMap 太大而无法广播。
@emeth 这要贵得多，但应该可以使用连接。有关详细信息，请参阅编辑。

【解决方案3】：

原始问题要求以最简单的方式从非分类中指定分类特征。

在 Spark ML 中，您可以使用 VectorIndexer 的 setMaxCategories 方法，您不必指定字段 - 相反，它会将基数小于或等于给定数字（在本例中为 2）的那些字段理解为分类字段。

val indexer = new VectorIndexer()
.setInputCol("features")
.setOutputCol("indexed")
.setMaxCategories(10)

详情请见this reply。

【讨论】：