Apache Spark MLLib - 使用 IDF-TF 向量运行 KMeans - Java 堆空间答案

【问题标题】：Apache Spark MLLib - Running KMeans with IDF-TF vectors - Java heap spaceApache Spark MLLib - 使用 IDF-TF 向量运行 KMeans - Java 堆空间
【发布时间】：2014-12-14 11:29:42
【问题描述】：

我正在尝试从（大型）文本文档集合（TF-IDF 向量）在 MLLib 上运行 KMeans。文档通过 Lucene 英语分析器发送，稀疏向量由 HashingTF.transform() 函数创建。无论我使用的并行度如何（通过 coalesce 函数），KMeans.train 总是在下面返回一个 OutOfMemory 异常。关于如何解决这个问题的任何想法？

Exception in thread "main" java.lang.OutOfMemoryError: Java heap space
at scala.reflect.ManifestFactory$$anon$12.newArray(Manifest.scala:138)
at scala.reflect.ManifestFactory$$anon$12.newArray(Manifest.scala:136)
at breeze.linalg.Vector$class.toArray(Vector.scala:80)
at breeze.linalg.SparseVector.toArray(SparseVector.scala:48)
at breeze.linalg.Vector$class.toDenseVector(Vector.scala:75)
at breeze.linalg.SparseVector.toDenseVector(SparseVector.scala:48)
at breeze.linalg.Vector$class.toDenseVector$mcD$sp(Vector.scala:74)
at breeze.linalg.SparseVector.toDenseVector$mcD$sp(SparseVector.scala:48)
at org.apache.spark.mllib.clustering.BreezeVectorWithNorm.toDense(KMeans.scala:422)
at org.apache.spark.mllib.clustering.KMeans$$anonfun$initKMeansParallel$1.apply(KMeans.scala:285)
at org.apache.spark.mllib.clustering.KMeans$$anonfun$initKMeansParallel$1.apply(KMeans.scala:284)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
at org.apache.spark.mllib.clustering.KMeans.initKMeansParallel(KMeans.scala:284)
at org.apache.spark.mllib.clustering.KMeans.runBreeze(KMeans.scala:143)
at org.apache.spark.mllib.clustering.KMeans.run(KMeans.scala:126)
at org.apache.spark.mllib.clustering.KMeans$.train(KMeans.scala:338)
at org.apache.spark.mllib.clustering.KMeans$.train(KMeans.scala:348)

【问题讨论】：

您能否检查问题是否来自没有更多内存或向量试图创建一个太大的数组（例如接近 Integer.MAX_VALUE）？这与 newArray 方法一致吗？
它总是来自同一个 newArray 方法。使用 HashingTF，向量非常大，但很稀疏。我想知道为什么 MLLib 试图将它们转换为 DenseVectors（这可能是问题所在）
据此：github.com/apache/spark/blob/master/mllib/src/main/scala/org/… 看来您的尺寸太大了。你以前试过降维吗？（虽然这可能需要更多内存，但我不确定。）
我也注意到了。 initKMeansParallel 和 initRandom 实现都创建了我的中心向量的密集副本，因此导致内存不足错误。我将研究可能的降维，例如 SVD。谢谢加博
介意分享您为 TF-IDF 构建向量的方法吗？在 KMeans 中使用 SparseVectors 时，我仍然遇到 OutOfMemory 的问题。 An error occurred while calling o379.trainKMeansModel. : java.lang.OutOfMemoryError: Java heap space at org.apache.spark.mllib.linalg.SparseVector.toArray(Vectors.scala:523) at org.apache.spark.mllib.clustering.KMeans$$anonfun$initRandom$1$$anonfun$apply$7.apply(KMeans.scala:267)

标签： scala apache-spark out-of-memory k-means apache-spark-mllib

【解决方案1】：

经过一番调查，发现这个问题与new HashingTF().transform(v)方法有关。尽管使用散列技巧创建稀疏向量确实很有帮助（尤其是在特征数量未知的情况下），但向量必须保持稀疏。 HashingTF 向量的默认大小为 2^20。给定 64 位双精度，理论上每个向量在转换为密集向量时需要 8MB - 无论我们可以应用降维。

遗憾的是，KMeans 使用 toDense 方法（至少对于集群中心而言），因此会导致 OutOfMemory 错误（假设 k = 1000）。

  private def initRandom(data: RDD[BreezeVectorWithNorm]) : Array[Array[BreezeVectorWithNorm]] = {
    val sample = data.takeSample(true, runs * k, new XORShiftRandom().nextInt()).toSeq
    Array.tabulate(runs)(r => sample.slice(r * k, (r + 1) * k).map { v =>
      new BreezeVectorWithNorm(v.vector.toDenseVector, v.norm)
    }.toArray)
  }

【讨论】：