CCO 抛出 NegativeArraySizeException 的 Apache Mahout SimilarityAnalysis答案

【问题标题】：Apache Mahout SimilarityAnalysis for CCO throwing NegativeArraySizeExceptionCCO 抛出 NegativeArraySizeException 的 Apache Mahout SimilarityAnalysis
【发布时间】：2017-04-19 20:57:05
【问题描述】：

在为 CCO 调用 Apache Mahout 的 SimilarityAnalysis 时，我收到一个关于 NegativeArraySizeException 的致命异常。

我正在运行的代码如下所示：

val result = SimilarityAnalysis.cooccurrencesIDSs(myIndexedDataSet:Array[IndexedDataset],
      randomSeed = 1234,
      maxInterestingItemsPerThing = 3,
      maxNumInteractions = 4)

我看到以下错误和相应的堆栈跟踪：

17/04/19 20:49:09 ERROR Executor: Exception in task 0.0 in stage 11.0 (TID 20)
java.lang.NegativeArraySizeException
    at org.apache.mahout.math.DenseVector.<init>(DenseVector.java:57)
    at org.apache.mahout.sparkbindings.SparkEngine$$anonfun$5.apply(SparkEngine.scala:73)
    at org.apache.mahout.sparkbindings.SparkEngine$$anonfun$5.apply(SparkEngine.scala:72)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$20.apply(RDD.scala:710)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$20.apply(RDD.scala:710)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
17/04/19 20:49:09 ERROR Executor: Exception in task 1.0 in stage 11.0 (TID 21)
java.lang.NegativeArraySizeException
    at org.apache.mahout.math.DenseVector.<init>(DenseVector.java:57)
    at org.apache.mahout.sparkbindings.SparkEngine$$anonfun$5.apply(SparkEngine.scala:73)
    at org.apache.mahout.sparkbindings.SparkEngine$$anonfun$5.apply(SparkEngine.scala:72)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$20.apply(RDD.scala:710)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$20.apply(RDD.scala:710)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
17/04/19 20:49:09 WARN TaskSetManager: Lost task 0.0 in stage 11.0 (TID 20, localhost): java.lang.NegativeArraySizeException
    at org.apache.mahout.math.DenseVector.<init>(DenseVector.java:57)
    at org.apache.mahout.sparkbindings.SparkEngine$$anonfun$5.apply(SparkEngine.scala:73)
    at org.apache.mahout.sparkbindings.SparkEngine$$anonfun$5.apply(SparkEngine.scala:72)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$20.apply(RDD.scala:710)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$20.apply(RDD.scala:710)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

我使用的是 Apache Mahout 0.13.0 版

【问题讨论】：

标签： apache scala mahout mahout-recommender

【解决方案1】：

这总是意味着输入矩阵之一是空的。数组中有多少个矩阵？每个中的行数和列数是多少？ IndexedDatasetSpark 有一个配套对象，它提供了一个构造函数，在 Scala 中称为 apply，它采用 RDD[String, String]，因此如果您可以将数据放入 RDD，只需使用它构造 IndexedDatasetSpark。这里的字符串对是用户 ID，对于某些事件（如购买）的项目 ID。

在此处查看伴随对象：https://github.com/apache/mahout/blob/master/spark/src/main/scala/org/apache/mahout/sparkbindings/indexeddataset/IndexedDatasetSpark.scala#L75

稍加搜索就会找到代码，只需一行左右的代码即可将 csv 转换为 RDD[String, String]。它看起来像这样：

val rawPurchaseInteractions = sc.textFile("/path/in/hdfs").map { line =>
  (line.split("\,")(0), (line.split("\,")(1))
}

虽然这会拆分两次，但它需要在带有user-id,item-id 的文本文件中以逗号分隔的行列表用于某种类型的交互，例如“购买”。如果文件中还有其他字段，只需拆分得到user-id和item-id。 map 函数中的行返回一对字符串，因此生成的 RDD 将是正确的类型，即 RDD[String, String]。将其传递给 IndexedDatasetSpark：

val purchasesRdd = IndexedDatasetSpark(rawPurchaseInteractions)(sc)

其中 sc 是您的 Spark 上下文。这应该会给您一个非空的IndexedDatasetSpark，您可以通过查看包装的BiDictionarys 的大小或调用包装的 Mahout DRM 上的方法来检查它。

顺便说一句，这假设 csv 没有标题。这是文本分隔的而不是完整的规范 csv。在 Spark 中使用其他方法可以读取真正的 CSV，但可能不需要。

【讨论】：

感谢@pferrel 的回复，我发现了与 Mahout 无关的问题（见下文）。

【解决方案2】：

这个问题实际上与 Mahout 无关，只是前面的一行：

inputRDD.filter(_ (1) == primaryFilter).map(o => (o(0), o(2)))

范围已关闭，我有 1 到 3 而不是 0 到 2。鉴于错误，我确信它的制作地点在 Mahout 内，但事实证明这是真正的问题。

【讨论】：