Apache Mahout CCO를 던지기위한 SimilarityAnalysis NegativeArraySizeException

Question

Cfa에 대한 Apache Mahout의 SimilarityAnalysis를 호출 할 때 NegativeArraySizeException에 대한 치명적인 예외가 발생합니다.

코드 나는이 같은 모습을하고 있는데 :

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

나는 다음과 같은 오류가 표시 및 스택 트레이스에 대응하고있다 : 나는 아파치 두싯 버전 0.13.0

을 사용하고

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) 

을

Answer 1

이것은 항상 입력 행렬 중 하나가 비어 있음을 의미합니다. 배열에 몇 개의 행렬이 있습니까? 각각의 행과 열의 수는 얼마입니까? 스칼라에서 apply이라는 생성자를 제공하는 IndexedDatasetSpark와 관련된 객체가 있습니다. 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

약간의 검색이 코드 정도의 한 라인으로 RDD [문자열, 문자열]에 CSV를 설정하는 코드를 찾을 수 있습니다. 이 두 번 분할 있지만

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을 얻을 수 있습니다.이 값은 포장 된 BiDictionary의 크기를 확인하거나 포장 된 Mahout DRM의 메소드를 호출하여 확인할 수 있습니다.

BTW 여기에는 csv에 헤더가 없다고 가정합니다. 텍스트로 구분 된 전체 사양이 아닌 CSV입니다. Spark에서 다른 방법을 사용하면 실제 CSV를 읽을 수 있지만 필요는 없습니다.

Answer 2

이 문제는 실제로 두싯와는 아무하지만 이전 라인이 없었 :

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

내가이 만들어진 확인 장소를 생각 오프, 내가 2 3 대신에 1 대 0이었다 범위 Mahout 내에서 오류가 발생했으나 실제로 문제가 발생했습니다.