[MAHOUT-1815] dsqDist(X,Y) and dsqDist(X) failing in flink tests.

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Epic Link: Samsara-framework
• Sprint: Nov/Dec-2015
• Assignee: Andrew Palumbo
• Reporter: Andrew Palumbo
• Created: 15/Mar/16 21:04
• Updated: 11/Apr/16 08:42
• Resolved: 17/Mar/16 22:35

Description

  test("dsqDist(X,Y)") {
    val m = 100
    val n = 300
    val d = 7
    val mxX = Matrices.symmetricUniformView(m, d, 12345).cloned -= 5
    val mxY = Matrices.symmetricUniformView(n, d, 1234).cloned += 10
    val (drmX, drmY) = (drmParallelize(mxX, 3), drmParallelize(mxY, 4))

    val mxDsq = dsqDist(drmX, drmY).collect
    val mxDsqControl = new DenseMatrix(m, n) := { (r, c, _) ⇒ (mxX(r, ::) - mxY(c, ::)) ^= 2 sum }
    (mxDsq - mxDsqControl).norm should be < 1e-7
  }

And

 test("dsqDist(X)") {
    val m = 100
    val d = 7
    val mxX = Matrices.symmetricUniformView(m, d, 12345).cloned -= 5
    val drmX = drmParallelize(mxX, 3)

    val mxDsq = dsqDist(drmX).collect
    val mxDsqControl = sqDist(drmX)
    (mxDsq - mxDsqControl).norm should be < 1e-7
  }

are both failing in flink tests with arrayOutOfBounds Exceptions:

03/15/2016 17:02:19	DataSink (org.apache.flink.api.java.Utils$CollectHelper@568b43ab)(5/10) switched to FINISHED 
1 [CHAIN GroupReduce (GroupReduce at org.apache.mahout.flinkbindings.blas.FlinkOpAtB$.notZippable(FlinkOpAtB.scala:78)) -> Map (Map at org.apache.mahout.flinkbindings.blas.FlinkOpMapBlock$.apply(FlinkOpMapBlock.scala:37)) -> FlatMap (FlatMap at org.apache.mahout.flinkbindings.drm.BlockifiedFlinkDrm.asRowWise(FlinkDrm.scala:93)) (8/10)] ERROR org.apache.flink.runtime.operators.BatchTask  - Error in task code:  CHAIN GroupReduce (GroupReduce at org.apache.mahout.flinkbindings.blas.FlinkOpAtB$.notZippable(FlinkOpAtB.scala:78)) -> Map (Map at org.apache.mahout.flinkbindings.blas.FlinkOpMapBlock$.apply(FlinkOpMapBlock.scala:37)) -> FlatMap (FlatMap at org.apache.mahout.flinkbindings.drm.BlockifiedFlinkDrm.asRowWise(FlinkDrm.scala:93)) (8/10)
java.lang.ArrayIndexOutOfBoundsException: 5
	at org.apache.mahout.math.drm.package$$anonfun$4$$anonfun$apply$3.apply(package.scala:317)
	at org.apache.mahout.math.drm.package$$anonfun$4$$anonfun$apply$3.apply(package.scala:317)
	at org.apache.mahout.math.scalabindings.MatrixOps$$anonfun$$colon$eq$3$$anonfun$apply$2.apply(MatrixOps.scala:164)
	at org.apache.mahout.math.scalabindings.MatrixOps$$anonfun$$colon$eq$3$$anonfun$apply$2.apply(MatrixOps.scala:164)
	at scala.collection.Iterator$class.foreach(Iterator.scala:727)
	at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
	at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
	at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
	at org.apache.mahout.math.scalabindings.MatrixOps$$anonfun$$colon$eq$3.apply(MatrixOps.scala:164)
	at org.apache.mahout.math.scalabindings.MatrixOps$$anonfun$$colon$eq$3.apply(MatrixOps.scala:164)
	at scala.collection.Iterator$class.foreach(Iterator.scala:727)
	at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
	at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
	at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
	at org.apache.mahout.math.scalabindings.MatrixOps.$colon$eq(MatrixOps.scala:164)
	at org.apache.mahout.math.drm.package$$anonfun$4.apply(package.scala:317)
	at org.apache.mahout.math.drm.package$$anonfun$4.apply(package.scala:311)
	at org.apache.mahout.flinkbindings.blas.FlinkOpMapBlock$$anonfun$1.apply(FlinkOpMapBlock.scala:39)
	at org.apache.mahout.flinkbindings.blas.FlinkOpMapBlock$$anonfun$1.apply(FlinkOpMapBlock.scala:38)
	at org.apache.flink.api.scala.DataSet$$anon$1.map(DataSet.scala:297)
	at org.apache.flink.runtime.operators.chaining.ChainedMapDriver.collect(ChainedMapDriver.java:78)
	at org.apache.mahout.flinkbindings.blas.FlinkOpAtB$$anon$6.reduce(FlinkOpAtB.scala:86)
	at org.apache.flink.runtime.operators.GroupReduceDriver.run(GroupReduceDriver.java:125)
	at org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:480)
	at org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:345)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
	at java.lang.Thread.run(Thread.java:745)

Issue Links

• is part of: MAHOUT-1570 Adding support for Apache Flink as a backend for the Mahout DSL