[jira] [Comment Edited] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15197724#comment-15197724
]
Andrew Palumbo edited comment on MAHOUT-1815 at 3/16/16 5:27 PM:
-
The exception is thrown here:
in {{dsqDist(X)}}
{code}
block := { (r, c, x) ⇒ s(keys(r)) + s(c) - 2 * x}
{code}
and similarilly in {{dsqDrist(X,Y)}}.
The offending call is {{keys(r)}} where {{r}} > {{keys.size}}
this can be seen in the following trace from within {{dsqDist(X)}}:
{code}
Keys.size: 5 block.rrow: 10
{code}
As these tests pass in H2O and Spark, it seems that this is likely due to some
partitioning problems in Flink Bindings.
ie. (Key, block) tuples are somehow being shuffled/mangled.
was (Author: andrew_palumbo):
The exception is thrown here:
in {{dsqDrist(X)}}
{code}
block := { (r, c, x) ⇒ s(keys(r)) + s(c) - 2 * x}
{code}
and similarilly in {{dsqDrist(X,Y)}}.
The offending call is {{keys(r)}} where {{r}} > {{keys.size}}
this can be seen in the following trace from within {{dsqDist(X)}}:
{code}
Keys.size: 5 block.rrow: 10
{code}
As these tests pass in H2O and Spark, it seems that this is likely due to some
partitioning problems in Flink Bindings.
ie. (Key, block) tuples are somehow being shuffled/mangled.
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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.scal
[jira] [Comment Edited] (MAHOUT-1810) Failing test in flink-bindings: A %*% B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200153#comment-15200153
]
Suneel Marthi edited comment on MAHOUT-1810 at 3/17/16 6:49 PM:
There's a DataSetUtils which has methods for zipping Datasets.
was (Author: smarthi):
There's a DataSetUtils which has methhods for zipping Datasets.
> Failing test in flink-bindings: A %*% B Identically partitioned
> ---
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1794) Support alternate temporary directories in example scripts.
[ https://issues.apache.org/jira/browse/MAHOUT-1794?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15203077#comment-15203077 ] Hudson commented on MAHOUT-1794: SUCCESS: Integrated in Mahout-Quality #3322 (See [https://builds.apache.org/job/Mahout-Quality/3322/]) Adding instructions for MAHOUT-1794 to the readme. (akm: rev b25a70a1bc6b9f8cb6c89947e0eaba5588463652) * examples/bin/README.txt > Support alternate temporary directories in example scripts. > --- > > Key: MAHOUT-1794 > URL: https://issues.apache.org/jira/browse/MAHOUT-1794 > Project: Mahout > Issue Type: Improvement > Components: Examples >Affects Versions: 0.11.1 >Reporter: Albert Chu >Assignee: Andrew Musselman >Priority: Minor > Fix For: 0.12.0 > > > In many of the example scripts, a directory in /tmp (e.g. > /tmp/mahout-work-dir) is created for use as temporary scratch space, such as > to store data files that are downloaded. > In a number of HPC environments, /tmp may not exist or /tmp may be very small > b/c local disk drives don't exist. It'd be convenient to be able to specify > an alternate directory to use as scratch space. > Pull request via Github to be sent shortly. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Resolved] (MAHOUT-1689) Create a doc on how to write an app that uses Mahout as a lib
[ https://issues.apache.org/jira/browse/MAHOUT-1689?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Pat Ferrel resolved MAHOUT-1689. Resolution: Fixed > Create a doc on how to write an app that uses Mahout as a lib > - > > Key: MAHOUT-1689 > URL: https://issues.apache.org/jira/browse/MAHOUT-1689 > Project: Mahout > Issue Type: Documentation >Affects Versions: 0.10.0 >Reporter: Andrew Palumbo >Assignee: Pat Ferrel > Fix For: 1.0.0 > > > Create a doc on how to write an app that uses Mahout as a lib -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Issue Comment Deleted] (MAHOUT-1762) Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
[
https://issues.apache.org/jira/browse/MAHOUT-1762?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Suneel Marthi updated MAHOUT-1762:
--
Comment: was deleted
(was: [~pferrel] I don't think we would have time to address this for 0.12.0
and should be punned to a subsequent release. But we really need this fixed in
the minor release following 0.12.0. )
> Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
> ---
>
> Key: MAHOUT-1762
> URL: https://issues.apache.org/jira/browse/MAHOUT-1762
> Project: Mahout
> Issue Type: Improvement
> Components: spark
>Reporter: Sergey Tryuber
>Assignee: Pat Ferrel
> Fix For: 1.0.0
>
>
> [spark-defaults.conf|http://spark.apache.org/docs/latest/configuration.html#dynamically-loading-spark-properties]
> is aimed to contain global configuration for Spark cluster. For example, in
> our HDP2.2 environment it contains:
> {noformat}
> spark.driver.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> spark.yarn.am.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> {noformat}
> and there are many other good things. Actually it is expected that when a
> user starts Spark Shell, it will be working fine. Unfortunately this does not
> happens with Mahout Spark Shell, because it ignores spark configuration and
> user has to copy-past lots of options into _MAHOUT_OPTS_.
> This happens because
> [org.apache.mahout.sparkbindings.shell.Main|https://github.com/apache/mahout/blob/master/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala]
> is executed directly in [initialization
> script|https://github.com/apache/mahout/blob/master/bin/mahout]:
> {code}
> "$JAVA" $JAVA_HEAP_MAX $MAHOUT_OPTS -classpath "$CLASSPATH"
> "org.apache.mahout.sparkbindings.shell.Main" $@
> {code}
> In contrast, in Spark shell is indirectly invoked through spark-submit in
> [spark-shell|https://github.com/apache/spark/blob/master/bin/spark-shell]
> script:
> {code}
> "$FWDIR"/bin/spark-submit --class org.apache.spark.repl.Main "$@"
> {code}
> [SparkSubmit|https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala]
> contains an additional initialization layer for loading properties file (see
> SparkSubmitArguments#mergeDefaultSparkProperties method).
> So there are two possible solutions:
> * use proper Spark-like initialization logic
> * use thin envelope like it is in H2O Sparkling Water
> ([sparkling-shell|https://github.com/h2oai/sparkling-water/blob/master/bin/sparkling-shell])
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Resolved] (MAHOUT-1794) Support alternate temporary directories in example scripts.
[ https://issues.apache.org/jira/browse/MAHOUT-1794?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Andrew Musselman resolved MAHOUT-1794. -- Resolution: Fixed Closed with c00b96ecc909fe6bcd4dce5e365a7fcea8a2bfdf and 1c1abbf3d946c6a8082f089b3f49900450553578 > Support alternate temporary directories in example scripts. > --- > > Key: MAHOUT-1794 > URL: https://issues.apache.org/jira/browse/MAHOUT-1794 > Project: Mahout > Issue Type: Improvement > Components: Examples >Affects Versions: 0.11.1 >Reporter: Albert Chu >Assignee: Andrew Musselman >Priority: Minor > Fix For: 0.12.0 > > > In many of the example scripts, a directory in /tmp (e.g. > /tmp/mahout-work-dir) is created for use as temporary scratch space, such as > to store data files that are downloaded. > In a number of HPC environments, /tmp may not exist or /tmp may be very small > b/c local disk drives don't exist. It'd be convenient to be able to specify > an alternate directory to use as scratch space. > Pull request via Github to be sent shortly. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1794) Support alternate temporary directories in example scripts.
[ https://issues.apache.org/jira/browse/MAHOUT-1794?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15203065#comment-15203065 ] ASF GitHub Bot commented on MAHOUT-1794: Github user asfgit closed the pull request at: https://github.com/apache/mahout/pull/178 > Support alternate temporary directories in example scripts. > --- > > Key: MAHOUT-1794 > URL: https://issues.apache.org/jira/browse/MAHOUT-1794 > Project: Mahout > Issue Type: Improvement > Components: Examples >Affects Versions: 0.11.1 >Reporter: Albert Chu >Assignee: Andrew Musselman >Priority: Minor > Fix For: 0.12.0 > > > In many of the example scripts, a directory in /tmp (e.g. > /tmp/mahout-work-dir) is created for use as temporary scratch space, such as > to store data files that are downloaded. > In a number of HPC environments, /tmp may not exist or /tmp may be very small > b/c local disk drives don't exist. It'd be convenient to be able to specify > an alternate directory to use as scratch space. > Pull request via Github to be sent shortly. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15197824#comment-15197824
]
Andrew Palumbo commented on MAHOUT-1815:
error is likely due to the implementation of {{FlinkOpAtB}} which sets matrix
block rows with an arbitrary factor of 10:
{code}
val blockHeight = 10
{code}
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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)
> a
[jira] [Assigned] (MAHOUT-1700) OutOfMemory Problem in ABtDenseOutJob in Distributed SSVD
[
https://issues.apache.org/jira/browse/MAHOUT-1700?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Suneel Marthi reassigned MAHOUT-1700:
-
Assignee: Suneel Marthi (was: Dmitriy Lyubimov)
> OutOfMemory Problem in ABtDenseOutJob in Distributed SSVD
> -
>
> Key: MAHOUT-1700
> URL: https://issues.apache.org/jira/browse/MAHOUT-1700
> Project: Mahout
> Issue Type: Bug
> Components: Math
>Affects Versions: 0.9, 0.10.0
>Reporter: Ethan Yi
>Assignee: Suneel Marthi
> Labels: patch
> Fix For: 0.12.0
>
>
> Recently, I tried mahout's hadoop ssvd(mahout-0.9 or mahout-1.0) job.
> There's a java heap space out of memory problem in ABtDenseOutJob. I found
> the reason, the ABtDenseOutJob map code is as below:
> protected void map(Writable key, VectorWritable value, Context context)
> throws IOException, InterruptedException {
> Vector vec = value.get();
> int vecSize = vec.size();
> if (aCols == null) {
> aCols = new Vector[vecSize];
> } else if (aCols.length < vecSize) {
> aCols = Arrays.copyOf(aCols, vecSize);
> }
> if (vec.isDense()) {
> for (int i = 0; i < vecSize; i++) {
> extendAColIfNeeded(i, aRowCount + 1);
> aCols[i].setQuick(aRowCount, vec.getQuick(i));
> }
> } else if (vec.size() > 0) {
> for (Vector.Element vecEl : vec.nonZeroes()) {
> int i = vecEl.index();
> extendAColIfNeeded(i, aRowCount + 1);
> aCols[i].setQuick(aRowCount, vecEl.get());
> }
> }
> aRowCount++;
> }
> If the input is RandomAccessSparseVector, usually with big data, it's
> vec.size() is Integer.MAX_VALUE, which is 2^31, then aCols = new
> Vector[vecSize] will introduce the OutOfMemory problem. The settlement of
> course should be enlarge every tasktracker's maximum memory:
>
> mapred.child.java.opts
> -Xmx1024m
>
> However, if you are NOT hadoop administrator or ops, you have no permission
> to modify the config. So, I try to modify ABtDenseOutJob map code to support
> RandomAccessSparseVector situation, I use hashmap to represent aCols instead
> of the original Vector[] aCols array, the modified code is as below:
> private Map aColsMap = new HashMap();
> protected void map(Writable key, VectorWritable value, Context context)
> throws IOException, InterruptedException {
> Vector vec = value.get();
> if (vec.isDense()) {
> for (int i = 0; i < vecSize; i++) {
> //extendAColIfNeeded(i, aRowCount + 1);
> if (aColsMap.get(i) == null) {
> aColsMap.put(i, new
> RandomAccessSparseVector(Integer.MAX_VALUE, 100));
> }
> aColsMap.get(i).setQuick(aRowCount, vec.getQuick(i));
> //aCols[i].setQuick(aRowCount, vec.getQuick(i));
> }
> } else if (vec.size() > 0) {
> for (Vector.Element vecEl : vec.nonZeroes()) {
> int i = vecEl.index();
> //extendAColIfNeeded(i, aRowCount + 1);
> if (aColsMap.get(i) == null) {
> aColsMap.put(i, new
> RandomAccessSparseVector(Integer.MAX_VALUE, 100));
> }
> aColsMap.get(i).setQuick(aRowCount, vecEl.get());
> //aCols[i].setQuick(aRowCount, vecEl.get());
> }
> }
> aRowCount++;
> }
> Then the OutofMemory problem is dismissed.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200524#comment-15200524
]
ASF GitHub Bot commented on MAHOUT-1815:
Github user andrewpalumbo commented on the pull request:
https://github.com/apache/mahout/pull/197#issuecomment-198109765
thx
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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.runti
[jira] [Commented] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200530#comment-15200530
]
ASF GitHub Bot commented on MAHOUT-1815:
Github user andrewpalumbo closed the pull request at:
https://github.com/apache/mahout/pull/197
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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
[jira] [Updated] (MAHOUT-1799) Read null row vectors from file in TextDelimeterReaderWriter driver
[ https://issues.apache.org/jira/browse/MAHOUT-1799?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Pat Ferrel updated MAHOUT-1799: --- Fix Version/s: (was: 0.12.0) 1.0.0 > Read null row vectors from file in TextDelimeterReaderWriter driver > --- > > Key: MAHOUT-1799 > URL: https://issues.apache.org/jira/browse/MAHOUT-1799 > Project: Mahout > Issue Type: Improvement > Components: spark >Reporter: Jussi Jousimo >Assignee: Pat Ferrel >Priority: Minor > Fix For: 1.0.0 > > > Since some row vectors in a sparse matrix can be null, Mahout writes them out > to a file with the row label only. However, Mahout cannot read these files, > but throws an exception when it encounters a label-only row. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Palumbo updated MAHOUT-1810:
---
Summary: Failing test in flink-bindings: A + B Identically partitioned
(was: Failing test in flink-bindings: A %*% B Identically partitioned)
> Failing test in flink-bindings: A + B Identically partitioned
> -
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200523#comment-15200523
]
ASF GitHub Bot commented on MAHOUT-1815:
Github user smarthi commented on the pull request:
https://github.com/apache/mahout/pull/197#issuecomment-198109713
LGTM
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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.ta
[jira] [Commented] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15197724#comment-15197724
]
Andrew Palumbo commented on MAHOUT-1815:
The exception is thrown here:
in {{dsqDrist(X)}}
{code}
block := { (r, c, x) ⇒ s(keys(r)) + s(c) - 2 * x}
{code}
and similarilly in {{dsqDrist(X,Y)}}.
The offending call is {{keys(r)}} where {{r}} > {{keys.size}}
this can be seen in the following trace:
{code}
Keys.size: 5 Block rows: 10
{code}
As these tests pass in H2O and Spark, it seems that this is likely due to some
partitioning problems in Flink Bindings.
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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)
>
[jira] [Updated] (MAHOUT-1679) example script run-item-sim should work on hdfs as well as local
[ https://issues.apache.org/jira/browse/MAHOUT-1679?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Pat Ferrel updated MAHOUT-1679: --- Fix Version/s: (was: 0.12.0) 1.0.0 Issue Type: Improvement (was: Bug) > example script run-item-sim should work on hdfs as well as local > > > Key: MAHOUT-1679 > URL: https://issues.apache.org/jira/browse/MAHOUT-1679 > Project: Mahout > Issue Type: Improvement > Components: Examples >Affects Versions: 0.10.0 >Reporter: Pat Ferrel >Assignee: Pat Ferrel >Priority: Minor > Fix For: 1.0.0 > > > mahout/examples/bin/run-item-sim does not run on a cluster or pseudo-cluster > Spark + HDFS > It prints a warning and how to run in cluster but should just work in either > mode -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A %*% B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199864#comment-15199864
]
Dmitriy Lyubimov commented on MAHOUT-1810:
--
Assumption of identical partitioning depends on the engine. Maybe it doesn't
hold in case of flink at all?
In this case (checkpoint or not) the assumption is that collection.map(x=>x)
doesn't change neither data allocation to splits nor its ordering inside every
split (aka partition). If this holds, then input and output are identically
partitioned.
Therefore, if B = A.map(x=> x...) then A and B are identically partitioned, and
then A + B can be optimized as A.zip(B).map (_._1 + _._2). If A and B are not
identically partitioned, then elementwise binary functions would require
pre-join, which is much more expensive than zip.
This test simply provokes this optimization (in spark), but if engine doesn't
support zips or assumption of identical partitioning does not hold, then engine
optimizer should rectify the situation by always executing join() after
mapblocks. Check back with me for more info where to hack it if it is indeed
the case..
> Failing test in flink-bindings: A %*% B Identically partitioned
> ---
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1809) Failing tests in Flink-bindings: dals and dspca
[
https://issues.apache.org/jira/browse/MAHOUT-1809?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15201863#comment-15201863
]
Andrew Palumbo commented on MAHOUT-1809:
{code}
test("dspca") {
val rnd = RandomUtils.getRandom
// Number of points
val m = 500
// Length of actual spectrum
val spectrumLen = 40
val spectrum = dvec((0 until spectrumLen).map(x => 300.0 * exp(-x) max
1e-3))
printf("spectrum:%s\n", spectrum)
val (u, _) = qr(new SparseRowMatrix(m, spectrumLen) :=
((r, c, v) => if (rnd.nextDouble() < 0.2) 0 else rnd.nextDouble() + 5.0))
// PCA Rotation matrix -- should also be orthonormal.
val (tr, _) = qr(Matrices.symmetricUniformView(spectrumLen, spectrumLen,
rnd.nextInt) - 10.0)
val input = (u %*%: diagv(spectrum)) %*% tr.t
val drmInput = drmParallelize(m = input, numPartitions = 2)
// Calculate just first 10 principal factors and reduce dimensionality.
// Since we assert just validity of the s-pca, not stochastic error, we
bump p parameter to
// ensure to zero stochastic error and assert only functional correctness
of the method's pca-
// specific additions.
val k = 10
// Calculate just first 10 principal factors and reduce dimensionality.
var (drmPCA, _, s) = dspca(drmA = drmInput, k = 10, p = spectrumLen, q = 1)
// Un-normalized pca data:
drmPCA = drmPCA %*% diagv(s)
val pca = drmPCA.checkpoint(CacheHint.NONE).collect
// Of course, once we calculated the pca, the spectrum is going to be
different since our originally
// generated input was not centered. So here, we'd just brute-solve pca to
verify
val xi = input.colMeans()
for (r <- 0 until input.nrow) input(r, ::) -= xi
var (pcaControl, _, sControl) = svd(m = input)
pcaControl = (pcaControl %*%: diagv(sControl))(::, 0 until k)
printf("pca:\n%s\n", pca(0 until 10, 0 until 10))
printf("pcaControl:\n%s\n", pcaControl(0 until 10, 0 until 10))
(pca(0 until 10, 0 until 10).norm - pcaControl(0 until 10, 0 until
10).norm).abs should be < 1E-5
}
{code}
This is the base test which is currently passing on Spark and H20 and failing
in Flink.
By checkpointing {{drmInout}} immediately upon parallelization, the test will
pass:
{code}
val drmInput = drmParallelize(m = input, numPartitions = 2).checkpoint()
{code}
> Failing tests in Flink-bindings: dals and dspca
> ---
>
> Key: MAHOUT-1809
> URL: https://issues.apache.org/jira/browse/MAHOUT-1809
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> {{dals}} and {{dspca}} are failing in the flinnk distributed decomposition
> suite with numerical and oom errors respectively:
> {dals}} Failure and stach trace:
> {code}
> 54.69239412917543 was not less than 1.0E-5
> ScalaTestFailureLocation:
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4 at
> (FailingTestsSuite.scala:230)
> org.scalatest.exceptions.TestFailedException: 54.69239412917543 was not less
> than 1.0E-5
> at
> org.scalatest.MatchersHelper$.newTestFailedException(MatchersHelper.scala:160)
> at
> org.scalatest.Matchers$ShouldMethodHelper$.shouldMatcher(Matchers.scala:6231)
> at org.scalatest.Matchers$AnyShouldWrapper.should(Matchers.scala:6265)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply$mcV$sp(FailingTestsSuite.scala:230)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
> at
> org.scalatest.Transformer$$anonfun$apply$1.apply$mcV$sp(Transformer.scala:22)
> at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
> at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
> at org.scalatest.Transformer.apply(Transformer.scala:22)
> at org.scalatest.Transformer.apply(Transformer.scala:20)
> at org.scalatest.FunSuiteLike$$anon$1.apply(FunSuiteLike.scala:166)
> at org.scalatest.Suite$class.withFixture(Suite.scala:1122)
> at org.scalatest.FunSuite.withFixture(FunSuite.scala:1555)
> at
> org.scalatest.FunSuiteLike$class.invokeWithFixture$1(FunSuiteLike.scala:163)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
> at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
> at org.scalatest.FunSuiteLike$class.runTest(FunSuiteLike.scala:175)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(Fai
[jira] [Commented] (MAHOUT-1794) Support alternate temporary directories in example scripts.
[ https://issues.apache.org/jira/browse/MAHOUT-1794?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15202978#comment-15202978 ] ASF GitHub Bot commented on MAHOUT-1794: Github user andrewmusselman commented on the pull request: https://github.com/apache/mahout/pull/178#issuecomment-198787422 False alarm, checked with a few scripts and this works fine. > Support alternate temporary directories in example scripts. > --- > > Key: MAHOUT-1794 > URL: https://issues.apache.org/jira/browse/MAHOUT-1794 > Project: Mahout > Issue Type: Improvement > Components: Examples >Affects Versions: 0.11.1 >Reporter: Albert Chu >Assignee: Andrew Musselman >Priority: Minor > Fix For: 0.12.0 > > > In many of the example scripts, a directory in /tmp (e.g. > /tmp/mahout-work-dir) is created for use as temporary scratch space, such as > to store data files that are downloaded. > In a number of HPC environments, /tmp may not exist or /tmp may be very small > b/c local disk drives don't exist. It'd be convenient to be able to specify > an alternate directory to use as scratch space. > Pull request via Github to be sent shortly. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1773) Fix cluster-syntheticcontrol.sh for HDFS synthax
[
https://issues.apache.org/jira/browse/MAHOUT-1773?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200589#comment-15200589
]
ASF GitHub Bot commented on MAHOUT-1773:
Github user andrewmusselman commented on a diff in the pull request:
https://github.com/apache/mahout/pull/157#discussion_r56592605
--- Diff: examples/bin/cluster-syntheticcontrol.sh ---
@@ -66,13 +66,13 @@ if [ ! -f ${WORK_DIR}/synthetic_control.data ]; then
fi
if [ "$HADOOP_HOME" != "" ] && [ "$MAHOUT_LOCAL" == "" ]; then
echo "Checking the health of DFS..."
- $DFS -ls
+ $DFS -ls /
if [ $? -eq 0 ];then
echo "DFS is healthy... "
echo "Uploading Synthetic control data to HDFS"
$DFSRM testdata
--- End diff --
Shouldn't this have a slash in it too?
> Fix cluster-syntheticcontrol.sh for HDFS synthax
>
>
> Key: MAHOUT-1773
> URL: https://issues.apache.org/jira/browse/MAHOUT-1773
> Project: Mahout
> Issue Type: Bug
> Components: Clustering
>Affects Versions: 0.11.0
>Reporter: Eduardo Niemeyer
>Priority: Trivial
> Fix For: 0.11.1
>
> Original Estimate: 5m
> Remaining Estimate: 5m
>
> Added '/' to some lines, so it can check the root of the HDFS, where it's
> possible to use -mkdir, -put and others. Otherwise the example keeps failing
> (Tested on Hadoop 2.4.1).
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1629) Mahout cvb on AWS EMR: p(topic|docId) doesn't make sense when using s3 folder as --input
[
https://issues.apache.org/jira/browse/MAHOUT-1629?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199781#comment-15199781
]
Suneel Marthi commented on MAHOUT-1629:
---
Have not heard back from the original poster in a while, and more over this is
an issue with the legacy MapReduce CVB code. Resolving this as 'Cannot
Reproduce', please feel free to open another jira if the issue recurs.
> Mahout cvb on AWS EMR: p(topic|docId) doesn't make sense when using s3 folder
> as --input
>
>
> Key: MAHOUT-1629
> URL: https://issues.apache.org/jira/browse/MAHOUT-1629
> Project: Mahout
> Issue Type: Bug
> Components: Clustering
>Affects Versions: 0.9
> Environment: AWS EMR with AMI 3.2.3
>Reporter: Markus Paaso
>Assignee: Andrew Musselman
> Labels: legacy
> Fix For: 0.12.0
>
>
> When running 'mahout cvb' command on AWS EMR having option --input with value
> like s3://mybucket/input/ or s3://mybucket/input/* (7 input files in my case)
> the content of doc-topic output is really non-sense. It seems like the docIds
> in doc-topic output are shuffled. But the topic model output (p(term|topic)
> for each topic) looks still fine.
> The workaround is to first copy input files from s3 to cluster's hdfs with
> command:
> {code:none}hadoop fs -cp s3://mybucket/input /input{code}
> and then running mahout cvb with option --input /input .
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1809) Failing tests in Flink-bindings: dals and dspca
[
https://issues.apache.org/jira/browse/MAHOUT-1809?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15201855#comment-15201855
]
Andrew Palumbo commented on MAHOUT-1809:
DSPCA failing in this issue is related to relates to MAHOUT-1810. The reason
for the test failure is that some calclations are happening before the
checkpoint is reached.
> Failing tests in Flink-bindings: dals and dspca
> ---
>
> Key: MAHOUT-1809
> URL: https://issues.apache.org/jira/browse/MAHOUT-1809
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> {{dals}} and {{dspca}} are failing in the flinnk distributed decomposition
> suite with numerical and oom errors respectively:
> {dals}} Failure and stach trace:
> {code}
> 54.69239412917543 was not less than 1.0E-5
> ScalaTestFailureLocation:
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4 at
> (FailingTestsSuite.scala:230)
> org.scalatest.exceptions.TestFailedException: 54.69239412917543 was not less
> than 1.0E-5
> at
> org.scalatest.MatchersHelper$.newTestFailedException(MatchersHelper.scala:160)
> at
> org.scalatest.Matchers$ShouldMethodHelper$.shouldMatcher(Matchers.scala:6231)
> at org.scalatest.Matchers$AnyShouldWrapper.should(Matchers.scala:6265)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply$mcV$sp(FailingTestsSuite.scala:230)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
> at
> org.scalatest.Transformer$$anonfun$apply$1.apply$mcV$sp(Transformer.scala:22)
> at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
> at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
> at org.scalatest.Transformer.apply(Transformer.scala:22)
> at org.scalatest.Transformer.apply(Transformer.scala:20)
> at org.scalatest.FunSuiteLike$$anon$1.apply(FunSuiteLike.scala:166)
> at org.scalatest.Suite$class.withFixture(Suite.scala:1122)
> at org.scalatest.FunSuite.withFixture(FunSuite.scala:1555)
> at
> org.scalatest.FunSuiteLike$class.invokeWithFixture$1(FunSuiteLike.scala:163)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
> at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
> at org.scalatest.FunSuiteLike$class.runTest(FunSuiteLike.scala:175)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(FailingTestsSuite.scala:48)
> at
> org.scalatest.BeforeAndAfterEach$class.runTest(BeforeAndAfterEach.scala:255)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.runTest(FailingTestsSuite.scala:48)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
> at
> org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:413)
> at
> org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:401)
> at scala.collection.immutable.List.foreach(List.scala:318)
> at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
> at
> org.scalatest.SuperEngine.org$scalatest$SuperEngine$$runTestsInBranch(Engine.scala:396)
> at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:483)
> at org.scalatest.FunSuiteLike$class.runTests(FunSuiteLike.scala:208)
> at org.scalatest.FunSuite.runTests(FunSuite.scala:1555)
> at org.scalatest.Suite$class.run(Suite.scala:1424)
> at
> org.scalatest.FunSuite.org$scalatest$FunSuiteLike$$super$run(FunSuite.scala:1555)
> at
> org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
> at
> org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
> at org.scalatest.SuperEngine.runImpl(Engine.scala:545)
> at org.scalatest.FunSuiteLike$class.run(FunSuiteLike.scala:212)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.org$scalatest$BeforeAndAfterAllConfigMap$$super$run(FailingTestsSuite.scala:48)
> at
> org.scalatest.BeforeAndAfterAllConfigMap$class.liftedTree1$1(BeforeAndAfterAllConfigMap.scala:248)
> at
> org.scalatest.BeforeAndAfterAllConfigMap$class.run(BeforeAndAfterAllConfigMap.scala:247)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.run(FailingTestsSuite.scala:48)
> at org.scalatest.tools.SuiteRunner.run(SuiteRunner
[jira] [Commented] (MAHOUT-1762) Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
[
https://issues.apache.org/jira/browse/MAHOUT-1762?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199694#comment-15199694
]
Pat Ferrel commented on MAHOUT-1762:
Do you know of something that is blocked by this? Not sure what is being asked
for.
> Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
> ---
>
> Key: MAHOUT-1762
> URL: https://issues.apache.org/jira/browse/MAHOUT-1762
> Project: Mahout
> Issue Type: Improvement
> Components: spark
>Reporter: Sergey Tryuber
>Assignee: Pat Ferrel
> Fix For: 1.0.0
>
>
> [spark-defaults.conf|http://spark.apache.org/docs/latest/configuration.html#dynamically-loading-spark-properties]
> is aimed to contain global configuration for Spark cluster. For example, in
> our HDP2.2 environment it contains:
> {noformat}
> spark.driver.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> spark.yarn.am.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> {noformat}
> and there are many other good things. Actually it is expected that when a
> user starts Spark Shell, it will be working fine. Unfortunately this does not
> happens with Mahout Spark Shell, because it ignores spark configuration and
> user has to copy-past lots of options into _MAHOUT_OPTS_.
> This happens because
> [org.apache.mahout.sparkbindings.shell.Main|https://github.com/apache/mahout/blob/master/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala]
> is executed directly in [initialization
> script|https://github.com/apache/mahout/blob/master/bin/mahout]:
> {code}
> "$JAVA" $JAVA_HEAP_MAX $MAHOUT_OPTS -classpath "$CLASSPATH"
> "org.apache.mahout.sparkbindings.shell.Main" $@
> {code}
> In contrast, in Spark shell is indirectly invoked through spark-submit in
> [spark-shell|https://github.com/apache/spark/blob/master/bin/spark-shell]
> script:
> {code}
> "$FWDIR"/bin/spark-submit --class org.apache.spark.repl.Main "$@"
> {code}
> [SparkSubmit|https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala]
> contains an additional initialization layer for loading properties file (see
> SparkSubmitArguments#mergeDefaultSparkProperties method).
> So there are two possible solutions:
> * use proper Spark-like initialization logic
> * use thin envelope like it is in H2O Sparkling Water
> ([sparkling-shell|https://github.com/h2oai/sparkling-water/blob/master/bin/sparkling-shell])
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Comment Edited] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200163#comment-15200163
]
Andrew Palumbo edited comment on MAHOUT-1810 at 3/17/16 7:57 PM:
-
>From the above tests though we can see that in the case of{{ B %*% A.t}} where
>B is a checkpointed non-deterministic matrix, Checkpointing is creating a
>barrier. I should have illustrated this better by Using A+B and also
>providing failure case. (I was also using it as a sanity test for FlinkOpAtB)
>But it does in fact fail without the checkpoint of B.
It seems that checkpointing the mapBlock result is the culprit. Maybe this
speaks to your point about intermediate caching.
was (Author: andrew_palumbo):
>From the above tests though we can see that in the case of A%*%B where B is a
>checkpointed non-deterministic matrix, Checkpointing is creating a barrier. I
>should have illustrated this better by Using A+B and also providing failure
>case. (I was also using it as a sanity test for FlinkOpAtB) But it does in
>fact fail without the checkpoint of B.
It seems that checkpointing the mapBlock result is the culprit. Maybe this
speaks to your point about intermediate caching.
> Failing test in flink-bindings: A + B Identically partitioned
> -
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1762) Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
[
https://issues.apache.org/jira/browse/MAHOUT-1762?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199696#comment-15199696
]
Jonathan Kelly commented on MAHOUT-1762:
Why was using spark-submit voted down? (And where? On a JIRA or on the mailing
list?) Was it only voted down for now (e.g., due to a time constraint), or are
you not planning on switching ever?
I think using spark-submit is Spark's recommended way of invoking Spark, even
for something like Mahout on Spark. Zeppelin and spark-jobserver used to do
something similar to what Mahout on Spark is doing now but have long since
switched to using spark-submit. I'm not too familiar with Hive on Spark, but
it looks from a quick glance at the source that it is also using spark-submit.
In short, I'd really suggest using spark-submit for Mahout as well, at least in
order to match what most other apps are doing and in order to follow best
practices.
> Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
> ---
>
> Key: MAHOUT-1762
> URL: https://issues.apache.org/jira/browse/MAHOUT-1762
> Project: Mahout
> Issue Type: Improvement
> Components: spark
>Reporter: Sergey Tryuber
>Assignee: Pat Ferrel
> Fix For: 1.0.0
>
>
> [spark-defaults.conf|http://spark.apache.org/docs/latest/configuration.html#dynamically-loading-spark-properties]
> is aimed to contain global configuration for Spark cluster. For example, in
> our HDP2.2 environment it contains:
> {noformat}
> spark.driver.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> spark.yarn.am.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> {noformat}
> and there are many other good things. Actually it is expected that when a
> user starts Spark Shell, it will be working fine. Unfortunately this does not
> happens with Mahout Spark Shell, because it ignores spark configuration and
> user has to copy-past lots of options into _MAHOUT_OPTS_.
> This happens because
> [org.apache.mahout.sparkbindings.shell.Main|https://github.com/apache/mahout/blob/master/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala]
> is executed directly in [initialization
> script|https://github.com/apache/mahout/blob/master/bin/mahout]:
> {code}
> "$JAVA" $JAVA_HEAP_MAX $MAHOUT_OPTS -classpath "$CLASSPATH"
> "org.apache.mahout.sparkbindings.shell.Main" $@
> {code}
> In contrast, in Spark shell is indirectly invoked through spark-submit in
> [spark-shell|https://github.com/apache/spark/blob/master/bin/spark-shell]
> script:
> {code}
> "$FWDIR"/bin/spark-submit --class org.apache.spark.repl.Main "$@"
> {code}
> [SparkSubmit|https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala]
> contains an additional initialization layer for loading properties file (see
> SparkSubmitArguments#mergeDefaultSparkProperties method).
> So there are two possible solutions:
> * use proper Spark-like initialization logic
> * use thin envelope like it is in H2O Sparkling Water
> ([sparkling-shell|https://github.com/h2oai/sparkling-water/blob/master/bin/sparkling-shell])
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Comment Edited] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200531#comment-15200531
]
Andrew Palumbo edited comment on MAHOUT-1815 at 3/17/16 10:35 PM:
--
merged to apache/flink-binding
was (Author: andrew_palumbo):
merged to flinl-binding
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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.runtim
[jira] [Assigned] (MAHOUT-1773) Fix cluster-syntheticcontrol.sh for HDFS synthax
[ https://issues.apache.org/jira/browse/MAHOUT-1773?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Andrew Musselman reassigned MAHOUT-1773: Assignee: Andrew Musselman > Fix cluster-syntheticcontrol.sh for HDFS synthax > > > Key: MAHOUT-1773 > URL: https://issues.apache.org/jira/browse/MAHOUT-1773 > Project: Mahout > Issue Type: Bug > Components: Clustering >Affects Versions: 0.11.0 >Reporter: Eduardo Niemeyer >Assignee: Andrew Musselman >Priority: Trivial > Fix For: 0.11.1 > > Original Estimate: 5m > Remaining Estimate: 5m > > Added '/' to some lines, so it can check the root of the HDFS, where it's > possible to use -mkdir, -put and others. Otherwise the example keeps failing > (Tested on Hadoop 2.4.1). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Comment Edited] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200163#comment-15200163
]
Andrew Palumbo edited comment on MAHOUT-1810 at 3/18/16 4:21 PM:
-
-From the above tests though we can see that in the case of {{B %*% A.t}} where
{{B}} is a checkpointed non-deterministic matrix, Checkpointing is creating a
barrier. I should have illustrated this better by Using A+B and also providing
failure case. (I was also using it as a sanity test for FlinkOpAtB) But it
does in fact fail without the checkpoint of {{B}}-.
*Edit Note*: Same as above.
It seems that checkpointing the mapBlock result is the culprit. Maybe this
speaks to your point about intermediate caching.
was (Author: andrew_palumbo):
>From the above tests though we can see that in the case of {{B %*% A.t}} where
>{{B}} is a checkpointed non-deterministic matrix, Checkpointing is creating a
>barrier. I should have illustrated this better by Using A+B and also
>providing failure case. (I was also using it as a sanity test for FlinkOpAtB)
>But it does in fact fail without the checkpoint of {{B}}.
It seems that checkpointing the mapBlock result is the culprit. Maybe this
speaks to your point about intermediate caching.
> Failing test in flink-bindings: A + B Identically partitioned
> -
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A + B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A %*% B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200153#comment-15200153
]
Suneel Marthi commented on MAHOUT-1810:
---
There's a DataSetUtils which has methhods for zipping Datasets.
> Failing test in flink-bindings: A %*% B Identically partitioned
> ---
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1762) Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
[
https://issues.apache.org/jira/browse/MAHOUT-1762?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199672#comment-15199672
]
Pat Ferrel commented on MAHOUT-1762:
I agree with the reasoning for this but the drivers have a pass-through to
Spark for arbitrary key=value pairs and switching to sparksubmit was voted down
so it was never done. If you are using Mahout as a lib you can set anything in
the SparkConf that you want so not sure what is remaining here but a more than
reasonable complaint about how the launcher scripts are structured.
> Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
> ---
>
> Key: MAHOUT-1762
> URL: https://issues.apache.org/jira/browse/MAHOUT-1762
> Project: Mahout
> Issue Type: Improvement
> Components: spark
>Reporter: Sergey Tryuber
>Assignee: Pat Ferrel
> Fix For: 1.0.0
>
>
> [spark-defaults.conf|http://spark.apache.org/docs/latest/configuration.html#dynamically-loading-spark-properties]
> is aimed to contain global configuration for Spark cluster. For example, in
> our HDP2.2 environment it contains:
> {noformat}
> spark.driver.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> spark.yarn.am.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> {noformat}
> and there are many other good things. Actually it is expected that when a
> user starts Spark Shell, it will be working fine. Unfortunately this does not
> happens with Mahout Spark Shell, because it ignores spark configuration and
> user has to copy-past lots of options into _MAHOUT_OPTS_.
> This happens because
> [org.apache.mahout.sparkbindings.shell.Main|https://github.com/apache/mahout/blob/master/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala]
> is executed directly in [initialization
> script|https://github.com/apache/mahout/blob/master/bin/mahout]:
> {code}
> "$JAVA" $JAVA_HEAP_MAX $MAHOUT_OPTS -classpath "$CLASSPATH"
> "org.apache.mahout.sparkbindings.shell.Main" $@
> {code}
> In contrast, in Spark shell is indirectly invoked through spark-submit in
> [spark-shell|https://github.com/apache/spark/blob/master/bin/spark-shell]
> script:
> {code}
> "$FWDIR"/bin/spark-submit --class org.apache.spark.repl.Main "$@"
> {code}
> [SparkSubmit|https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala]
> contains an additional initialization layer for loading properties file (see
> SparkSubmitArguments#mergeDefaultSparkProperties method).
> So there are two possible solutions:
> * use proper Spark-like initialization logic
> * use thin envelope like it is in H2O Sparkling Water
> ([sparkling-shell|https://github.com/h2oai/sparkling-water/blob/master/bin/sparkling-shell])
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1689) Create a doc on how to write an app that uses Mahout as a lib
[ https://issues.apache.org/jira/browse/MAHOUT-1689?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199754#comment-15199754 ] Pat Ferrel commented on MAHOUT-1689: Done, many times over by several people. Mine is here: http://mahout.apache.org/users/environment/how-to-build-an-app.html > Create a doc on how to write an app that uses Mahout as a lib > - > > Key: MAHOUT-1689 > URL: https://issues.apache.org/jira/browse/MAHOUT-1689 > Project: Mahout > Issue Type: Documentation >Affects Versions: 0.10.0 >Reporter: Andrew Palumbo >Assignee: Pat Ferrel > Fix For: 1.0.0 > > > Create a doc on how to write an app that uses Mahout as a lib -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (MAHOUT-1788) spark-itemsimilarity integration test script cleanup
[ https://issues.apache.org/jira/browse/MAHOUT-1788?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Pat Ferrel updated MAHOUT-1788: --- Fix Version/s: (was: 0.12.0) 1.0.0 Issue Type: Improvement (was: Bug) work on this as time is available, not blocking anything IMO > spark-itemsimilarity integration test script cleanup > > > Key: MAHOUT-1788 > URL: https://issues.apache.org/jira/browse/MAHOUT-1788 > Project: Mahout > Issue Type: Improvement > Components: cooccurrence >Affects Versions: 0.11.0 >Reporter: Pat Ferrel >Assignee: Pat Ferrel >Priority: Trivial > Fix For: 1.0.0 > > > binary release does not contain data for itemsimilarity tests, neith binary > nor source versions will run on a cluster unless data is hand copied to hdfs. > Clean this up so it copies data if needed and the data is in both versions. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Assigned] (MAHOUT-1809) Failing tests in Flink-bindings: dals and dspca
[
https://issues.apache.org/jira/browse/MAHOUT-1809?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Palumbo reassigned MAHOUT-1809:
--
Assignee: Andrew Palumbo
> Failing tests in Flink-bindings: dals and dspca
> ---
>
> Key: MAHOUT-1809
> URL: https://issues.apache.org/jira/browse/MAHOUT-1809
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> {{dals}} and {{dspca}} are failing in the flinnk distributed decomposition
> suite with numerical and oom errors respectively:
> {dals}} Failure and stach trace:
> {code}
> 54.69239412917543 was not less than 1.0E-5
> ScalaTestFailureLocation:
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4 at
> (FailingTestsSuite.scala:230)
> org.scalatest.exceptions.TestFailedException: 54.69239412917543 was not less
> than 1.0E-5
> at
> org.scalatest.MatchersHelper$.newTestFailedException(MatchersHelper.scala:160)
> at
> org.scalatest.Matchers$ShouldMethodHelper$.shouldMatcher(Matchers.scala:6231)
> at org.scalatest.Matchers$AnyShouldWrapper.should(Matchers.scala:6265)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply$mcV$sp(FailingTestsSuite.scala:230)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
> at
> org.scalatest.Transformer$$anonfun$apply$1.apply$mcV$sp(Transformer.scala:22)
> at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
> at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
> at org.scalatest.Transformer.apply(Transformer.scala:22)
> at org.scalatest.Transformer.apply(Transformer.scala:20)
> at org.scalatest.FunSuiteLike$$anon$1.apply(FunSuiteLike.scala:166)
> at org.scalatest.Suite$class.withFixture(Suite.scala:1122)
> at org.scalatest.FunSuite.withFixture(FunSuite.scala:1555)
> at
> org.scalatest.FunSuiteLike$class.invokeWithFixture$1(FunSuiteLike.scala:163)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
> at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
> at org.scalatest.FunSuiteLike$class.runTest(FunSuiteLike.scala:175)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(FailingTestsSuite.scala:48)
> at
> org.scalatest.BeforeAndAfterEach$class.runTest(BeforeAndAfterEach.scala:255)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.runTest(FailingTestsSuite.scala:48)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
> at
> org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:413)
> at
> org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:401)
> at scala.collection.immutable.List.foreach(List.scala:318)
> at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
> at
> org.scalatest.SuperEngine.org$scalatest$SuperEngine$$runTestsInBranch(Engine.scala:396)
> at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:483)
> at org.scalatest.FunSuiteLike$class.runTests(FunSuiteLike.scala:208)
> at org.scalatest.FunSuite.runTests(FunSuite.scala:1555)
> at org.scalatest.Suite$class.run(Suite.scala:1424)
> at
> org.scalatest.FunSuite.org$scalatest$FunSuiteLike$$super$run(FunSuite.scala:1555)
> at
> org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
> at
> org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
> at org.scalatest.SuperEngine.runImpl(Engine.scala:545)
> at org.scalatest.FunSuiteLike$class.run(FunSuiteLike.scala:212)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.org$scalatest$BeforeAndAfterAllConfigMap$$super$run(FailingTestsSuite.scala:48)
> at
> org.scalatest.BeforeAndAfterAllConfigMap$class.liftedTree1$1(BeforeAndAfterAllConfigMap.scala:248)
> at
> org.scalatest.BeforeAndAfterAllConfigMap$class.run(BeforeAndAfterAllConfigMap.scala:247)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.run(FailingTestsSuite.scala:48)
> at org.scalatest.tools.SuiteRunner.run(SuiteRunner.scala:55)
> at
> org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$3.apply(Runner.scala:2563)
> at
> org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$3.apply(Runner.s
[jira] [Work started] (MAHOUT-1809) Failing tests in Flink-bindings: dals and dspca
[
https://issues.apache.org/jira/browse/MAHOUT-1809?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on MAHOUT-1809 started by Andrew Palumbo.
--
> Failing tests in Flink-bindings: dals and dspca
> ---
>
> Key: MAHOUT-1809
> URL: https://issues.apache.org/jira/browse/MAHOUT-1809
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> {{dals}} and {{dspca}} are failing in the flinnk distributed decomposition
> suite with numerical and oom errors respectively:
> {dals}} Failure and stach trace:
> {code}
> 54.69239412917543 was not less than 1.0E-5
> ScalaTestFailureLocation:
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4 at
> (FailingTestsSuite.scala:230)
> org.scalatest.exceptions.TestFailedException: 54.69239412917543 was not less
> than 1.0E-5
> at
> org.scalatest.MatchersHelper$.newTestFailedException(MatchersHelper.scala:160)
> at
> org.scalatest.Matchers$ShouldMethodHelper$.shouldMatcher(Matchers.scala:6231)
> at org.scalatest.Matchers$AnyShouldWrapper.should(Matchers.scala:6265)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply$mcV$sp(FailingTestsSuite.scala:230)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
> at
> org.scalatest.Transformer$$anonfun$apply$1.apply$mcV$sp(Transformer.scala:22)
> at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
> at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
> at org.scalatest.Transformer.apply(Transformer.scala:22)
> at org.scalatest.Transformer.apply(Transformer.scala:20)
> at org.scalatest.FunSuiteLike$$anon$1.apply(FunSuiteLike.scala:166)
> at org.scalatest.Suite$class.withFixture(Suite.scala:1122)
> at org.scalatest.FunSuite.withFixture(FunSuite.scala:1555)
> at
> org.scalatest.FunSuiteLike$class.invokeWithFixture$1(FunSuiteLike.scala:163)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
> at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
> at org.scalatest.FunSuiteLike$class.runTest(FunSuiteLike.scala:175)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(FailingTestsSuite.scala:48)
> at
> org.scalatest.BeforeAndAfterEach$class.runTest(BeforeAndAfterEach.scala:255)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.runTest(FailingTestsSuite.scala:48)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
> at
> org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
> at
> org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:413)
> at
> org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:401)
> at scala.collection.immutable.List.foreach(List.scala:318)
> at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
> at
> org.scalatest.SuperEngine.org$scalatest$SuperEngine$$runTestsInBranch(Engine.scala:396)
> at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:483)
> at org.scalatest.FunSuiteLike$class.runTests(FunSuiteLike.scala:208)
> at org.scalatest.FunSuite.runTests(FunSuite.scala:1555)
> at org.scalatest.Suite$class.run(Suite.scala:1424)
> at
> org.scalatest.FunSuite.org$scalatest$FunSuiteLike$$super$run(FunSuite.scala:1555)
> at
> org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
> at
> org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
> at org.scalatest.SuperEngine.runImpl(Engine.scala:545)
> at org.scalatest.FunSuiteLike$class.run(FunSuiteLike.scala:212)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.org$scalatest$BeforeAndAfterAllConfigMap$$super$run(FailingTestsSuite.scala:48)
> at
> org.scalatest.BeforeAndAfterAllConfigMap$class.liftedTree1$1(BeforeAndAfterAllConfigMap.scala:248)
> at
> org.scalatest.BeforeAndAfterAllConfigMap$class.run(BeforeAndAfterAllConfigMap.scala:247)
> at
> org.apache.mahout.flinkbindings.FailingTestsSuite.run(FailingTestsSuite.scala:48)
> at org.scalatest.tools.SuiteRunner.run(SuiteRunner.scala:55)
> at
> org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$3.apply(Runner.scala:2563)
> at
> org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$3.apply(Runner.scala:2557)
>
[jira] [Updated] (MAHOUT-1689) Create a doc on how to write an app that uses Mahout as a lib
[ https://issues.apache.org/jira/browse/MAHOUT-1689?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Suneel Marthi updated MAHOUT-1689: -- Fix Version/s: (was: 1.0.0) 0.12.0 > Create a doc on how to write an app that uses Mahout as a lib > - > > Key: MAHOUT-1689 > URL: https://issues.apache.org/jira/browse/MAHOUT-1689 > Project: Mahout > Issue Type: Documentation >Affects Versions: 0.10.0 >Reporter: Andrew Palumbo >Assignee: Pat Ferrel > Fix For: 0.12.0 > > > Create a doc on how to write an app that uses Mahout as a lib -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200502#comment-15200502
]
ASF GitHub Bot commented on MAHOUT-1815:
GitHub user andrewpalumbo opened a pull request:
https://github.com/apache/mahout/pull/197
MAHOUT-1815: dsqDist(X,Y) and dsqDist(X) failing in flink tests.
After taking the Very long way around trying to repartition, etc., it turns
out that the row just needed to be properly re-keyed.
Tests pass now.
Though we may want to re-examine the implementation of FlinkOpAtB, as it
seems pretty inefficient.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/andrewpalumbo/mahout MAHOUT-1815
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/mahout/pull/197.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #197
commit 731b87c88f65d7ecc0d098405f27cded5f094fa2
Author: Andrew Palumbo
Date: 2016-03-17T22:11:26Z
properly re-key rows in FlinkOpAtB
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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.ma
[jira] [Updated] (MAHOUT-1742) non-legacy framework related issues
[ https://issues.apache.org/jira/browse/MAHOUT-1742?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Suneel Marthi updated MAHOUT-1742: -- Fix Version/s: 0.12.0 > non-legacy framework related issues > --- > > Key: MAHOUT-1742 > URL: https://issues.apache.org/jira/browse/MAHOUT-1742 > Project: Mahout > Issue Type: Epic >Reporter: Dmitriy Lyubimov >Assignee: Suneel Marthi > Fix For: 0.12.0 > > -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (MAHOUT-1705) Verify dependencies in job jar for mahout-examples
[ https://issues.apache.org/jira/browse/MAHOUT-1705?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Suneel Marthi updated MAHOUT-1705: -- Assignee: Andrew Musselman (was: Andrew Palumbo) > Verify dependencies in job jar for mahout-examples > -- > > Key: MAHOUT-1705 > URL: https://issues.apache.org/jira/browse/MAHOUT-1705 > Project: Mahout > Issue Type: Improvement >Affects Versions: 0.10.0 >Reporter: Andrew Palumbo >Assignee: Andrew Musselman > Fix For: 0.12.0 > > > mahout-example-*-job.jar is around ~56M, and may package unused runtime > libraries. We need to go through this and make sure that there is nothing > unneeded or redundant. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Assigned] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Palumbo reassigned MAHOUT-1815:
--
Assignee: Andrew Palumbo
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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)
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Work started] (MAHOUT-1742) non-legacy framework related issues
[ https://issues.apache.org/jira/browse/MAHOUT-1742?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Work on MAHOUT-1742 started by Suneel Marthi. - > non-legacy framework related issues > --- > > Key: MAHOUT-1742 > URL: https://issues.apache.org/jira/browse/MAHOUT-1742 > Project: Mahout > Issue Type: Epic >Reporter: Dmitriy Lyubimov >Assignee: Suneel Marthi > Fix For: 0.12.0 > > -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15201922#comment-15201922
]
Andrew Palumbo commented on MAHOUT-1810:
[~dlyubimov]I am almost positive that the failure is due to checkpointing
failing on mapBlock. I believe that this causing the test failures on dspca
(MAHOUT-1809), ie, the internal mapBlock call in dspca. Could you let me know
what your thoughts are on the join fix when you get a chance?
> Failing test in flink-bindings: A + B Identically partitioned
> -
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A + B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1799) Read null row vectors from file in TextDelimeterReaderWriter driver
[ https://issues.apache.org/jira/browse/MAHOUT-1799?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200707#comment-15200707 ] ASF GitHub Bot commented on MAHOUT-1799: Github user statguy commented on the pull request: https://github.com/apache/mahout/pull/182#issuecomment-198145198 @pferrel you are asking why null rows should not give an error? For example, when I find user-item recommendations, but not all users get recommendations. I write the matrix to an intermediate file and read it back to process it further to fill those null rows with some sort of average recommendations. I could write a unit test for this too, but I would like get an approval first. > Read null row vectors from file in TextDelimeterReaderWriter driver > --- > > Key: MAHOUT-1799 > URL: https://issues.apache.org/jira/browse/MAHOUT-1799 > Project: Mahout > Issue Type: Improvement > Components: spark >Reporter: Jussi Jousimo >Assignee: Pat Ferrel >Priority: Minor > Fix For: 1.0.0 > > > Since some row vectors in a sparse matrix can be null, Mahout writes them out > to a file with the row label only. However, Mahout cannot read these files, > but throws an exception when it encounters a label-only row. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (MAHOUT-1814) Implement drm2intKeyed in flink bindings
[ https://issues.apache.org/jira/browse/MAHOUT-1814?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Suneel Marthi updated MAHOUT-1814: -- Summary: Implement drm2intKeyed in flink bindings (was: Implement drn2intKeyed in flink bindings) > Implement drm2intKeyed in flink bindings > > > Key: MAHOUT-1814 > URL: https://issues.apache.org/jira/browse/MAHOUT-1814 > Project: Mahout > Issue Type: New Feature >Reporter: Andrew Palumbo >Assignee: Suneel Marthi > Fix For: 0.12.0 > > -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (MAHOUT-1705) Verify dependencies in job jar for mahout-examples
[ https://issues.apache.org/jira/browse/MAHOUT-1705?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Suneel Marthi updated MAHOUT-1705: -- Fix Version/s: (was: 1.0.0) 0.12.0 > Verify dependencies in job jar for mahout-examples > -- > > Key: MAHOUT-1705 > URL: https://issues.apache.org/jira/browse/MAHOUT-1705 > Project: Mahout > Issue Type: Improvement >Affects Versions: 0.10.0 >Reporter: Andrew Palumbo >Assignee: Andrew Musselman > Fix For: 0.12.0 > > > mahout-example-*-job.jar is around ~56M, and may package unused runtime > libraries. We need to go through this and make sure that there is nothing > unneeded or redundant. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Comment Edited] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15197724#comment-15197724
]
Andrew Palumbo edited comment on MAHOUT-1815 at 3/16/16 5:25 PM:
-
The exception is thrown here:
in {{dsqDrist(X)}}
{code}
block := { (r, c, x) ⇒ s(keys(r)) + s(c) - 2 * x}
{code}
and similarilly in {{dsqDrist(X,Y)}}.
The offending call is {{keys(r)}} where {{r}} > {{keys.size}}
this can be seen in the following trace from within {{dsqDist(X)}}:
{code}
Keys.size: 5 block.rrow: 10
{code}
As these tests pass in H2O and Spark, it seems that this is likely due to some
partitioning problems in Flink Bindings.
ie. (Key, block) tuples are somehow being shuffled/mangled.
was (Author: andrew_palumbo):
The exception is thrown here:
in {{dsqDrist(X)}}
{code}
block := { (r, c, x) ⇒ s(keys(r)) + s(c) - 2 * x}
{code}
and similarilly in {{dsqDrist(X,Y)}}.
The offending call is {{keys(r)}} where {{r}} > {{keys.size}}
this can be seen in the following trace:
{code}
Keys.size: 5 Block rows: 10
{code}
As these tests pass in H2O and Spark, it seems that this is likely due to some
partitioning problems in Flink Bindings.
ie. (Key, block) tuples are somehow being shuffled/mangled.
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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.ap
[jira] [Work started] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on MAHOUT-1815 started by Andrew Palumbo.
--
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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)
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Updated] (MAHOUT-1809) Failing tests in Flink-bindings: dals and dspca
[
https://issues.apache.org/jira/browse/MAHOUT-1809?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Palumbo updated MAHOUT-1809:
---
Description:
{{dspca}} and {{dals}} are failing in the flink distributed decomposition suite
with numerical and oom errors respectively:
{{dspca}} Failure and stack trace:
{code}
54.69239412917543 was not less than 1.0E-5
ScalaTestFailureLocation:
org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4 at
(FailingTestsSuite.scala:230)
org.scalatest.exceptions.TestFailedException: 54.69239412917543 was not less
than 1.0E-5
at
org.scalatest.MatchersHelper$.newTestFailedException(MatchersHelper.scala:160)
at
org.scalatest.Matchers$ShouldMethodHelper$.shouldMatcher(Matchers.scala:6231)
at org.scalatest.Matchers$AnyShouldWrapper.should(Matchers.scala:6265)
at
org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply$mcV$sp(FailingTestsSuite.scala:230)
at
org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
at
org.apache.mahout.flinkbindings.FailingTestsSuite$$anonfun$4.apply(FailingTestsSuite.scala:186)
at
org.scalatest.Transformer$$anonfun$apply$1.apply$mcV$sp(Transformer.scala:22)
at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
at org.scalatest.Transformer.apply(Transformer.scala:22)
at org.scalatest.Transformer.apply(Transformer.scala:20)
at org.scalatest.FunSuiteLike$$anon$1.apply(FunSuiteLike.scala:166)
at org.scalatest.Suite$class.withFixture(Suite.scala:1122)
at org.scalatest.FunSuite.withFixture(FunSuite.scala:1555)
at
org.scalatest.FunSuiteLike$class.invokeWithFixture$1(FunSuiteLike.scala:163)
at
org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
at
org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:175)
at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
at org.scalatest.FunSuiteLike$class.runTest(FunSuiteLike.scala:175)
at
org.apache.mahout.flinkbindings.FailingTestsSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(FailingTestsSuite.scala:48)
at
org.scalatest.BeforeAndAfterEach$class.runTest(BeforeAndAfterEach.scala:255)
at
org.apache.mahout.flinkbindings.FailingTestsSuite.runTest(FailingTestsSuite.scala:48)
at
org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
at
org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:208)
at
org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:413)
at
org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:401)
at scala.collection.immutable.List.foreach(List.scala:318)
at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
at
org.scalatest.SuperEngine.org$scalatest$SuperEngine$$runTestsInBranch(Engine.scala:396)
at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:483)
at org.scalatest.FunSuiteLike$class.runTests(FunSuiteLike.scala:208)
at org.scalatest.FunSuite.runTests(FunSuite.scala:1555)
at org.scalatest.Suite$class.run(Suite.scala:1424)
at
org.scalatest.FunSuite.org$scalatest$FunSuiteLike$$super$run(FunSuite.scala:1555)
at
org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
at
org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:212)
at org.scalatest.SuperEngine.runImpl(Engine.scala:545)
at org.scalatest.FunSuiteLike$class.run(FunSuiteLike.scala:212)
at
org.apache.mahout.flinkbindings.FailingTestsSuite.org$scalatest$BeforeAndAfterAllConfigMap$$super$run(FailingTestsSuite.scala:48)
at
org.scalatest.BeforeAndAfterAllConfigMap$class.liftedTree1$1(BeforeAndAfterAllConfigMap.scala:248)
at
org.scalatest.BeforeAndAfterAllConfigMap$class.run(BeforeAndAfterAllConfigMap.scala:247)
at
org.apache.mahout.flinkbindings.FailingTestsSuite.run(FailingTestsSuite.scala:48)
at org.scalatest.tools.SuiteRunner.run(SuiteRunner.scala:55)
at
org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$3.apply(Runner.scala:2563)
at
org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$3.apply(Runner.scala:2557)
at scala.collection.immutable.List.foreach(List.scala:318)
at org.scalatest.tools.Runner$.doRunRunRunDaDoRunRun(Runner.scala:2557)
at
org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:1044)
at
org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:1043)
at
org.scalatest.tools.Runner$.withClassLoaderAndDispatchReporter(Runner.scala:2722)
at
org.scalatest.t
[jira] [Work started] (MAHOUT-1814) Implement drn2intKeyed in flink bindings
[ https://issues.apache.org/jira/browse/MAHOUT-1814?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Work on MAHOUT-1814 started by Suneel Marthi. - > Implement drn2intKeyed in flink bindings > > > Key: MAHOUT-1814 > URL: https://issues.apache.org/jira/browse/MAHOUT-1814 > Project: Mahout > Issue Type: New Feature >Reporter: Andrew Palumbo >Assignee: Suneel Marthi > Fix For: 0.12.0 > > -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A %*% B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15198693#comment-15198693
]
Andrew Palumbo commented on MAHOUT-1810:
Flink Checkpointing seems to be working fine on other computational paths
(other then MapBlock).
{code}
test("Checkpoint test") {
val inCoreA = dense((1, 2, 3), (3, 4, 5), (4, 5, 6), (5, 6, 7))
val inCoreB = inCoreA.like := { (r, c, v) => util.Random.nextDouble()}
val A = drmParallelize(inCoreA, numPartitions = 2)
val B = drmParallelize(inCoreB, numPartitions = 2).checkpoint()
val C = (B %*% A.t)
val D = (B %*% A.t)
val inCoreC = C.collect
val inCoreD = D.collect
println(inCoreC)
(inCoreC - inCoreD).norm should be < 1E-10
}
{code}
passes.
> Failing test in flink-bindings: A %*% B Identically partitioned
> ---
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200519#comment-15200519
]
ASF GitHub Bot commented on MAHOUT-1815:
Github user andrewpalumbo commented on the pull request:
https://github.com/apache/mahout/pull/197#issuecomment-198109270
This is a small fix so if there are no objections I'm just gonna push it.
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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.
[jira] [Commented] (MAHOUT-1766) Increase default PermGen size for spark-shell
[
https://issues.apache.org/jira/browse/MAHOUT-1766?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200095#comment-15200095
]
Andrew Palumbo commented on MAHOUT-1766:
I've run into this with different spark versions and different jvms. It's
easily overcome by stetting the JVM options in line when launching the
spark-shell.
eg.:
{code}
$MAHOUT_OPTS="-Xmx6g -XX:MaxPermSize=512m" mahout spark-shell
{code}
It does seem to pop up mow with Spatk 1.5.2 regularly, so maybe we should add
this as the default into the {{/bin/mahout}} launcher, though that might be a
bit restrictive.
> Increase default PermGen size for spark-shell
> -
>
> Key: MAHOUT-1766
> URL: https://issues.apache.org/jira/browse/MAHOUT-1766
> Project: Mahout
> Issue Type: Improvement
> Components: Mahout spark shell
>Affects Versions: 0.11.0
>Reporter: Sergey Tryuber
>Assignee: Andrew Palumbo
> Fix For: 0.12.0
>
>
> Mahout spark-shell is run with default perm gen size (64MB). Taking into
> account that it depends on lots of external jars and the whole count of used
> Java classes is very large, we constantly observe spontaneous corresponding
> OOM exceptions.
> A hot fix from our side is to modify envelope bash script (added
> -XX:PermSize=512m):
> {code}
> "$JAVA" $JAVA_HEAP_MAX -XX:PermSize=512m $MAHOUT_OPTS -classpath "$CLASSPATH"
> "org.apache.mahout.sparkbindings.shell.Main" $@
> {code}
> Of course, more elegant solution is needed. After the applied fix, the errors
> had gone.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A %*% B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200163#comment-15200163
]
Andrew Palumbo commented on MAHOUT-1810:
>From the above tests though we can see that in the case of A%*%B where B is a
>checkpointed non-deterministic matrix, Checkpointing is creating a barrier. I
>should have illustrated this better by Using A+B and also providing failure
>case. (I was also using it as a sanity test for FlinkOpAtB) But it does in
>fact fail without the checkpoint of B.
It seems that checkpointing the mapBlock result is the culprit. Maybe this
speaks to your point about intermediate caching.
> Failing test in flink-bindings: A %*% B Identically partitioned
> ---
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1794) Support alternate temporary directories in example scripts.
[ https://issues.apache.org/jira/browse/MAHOUT-1794?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15201004#comment-15201004 ] ASF GitHub Bot commented on MAHOUT-1794: Github user andrewmusselman commented on the pull request: https://github.com/apache/mahout/pull/178#issuecomment-198204388 Thanks for the addition, Al. The way this is written you will have to create a similar path on your local file system that will mirror the path you have in HDFS. So for example, if you create a path in HDFS for your data, say '/user/al/mahout-tmp', and export it into MAHOUT_WORK_DIR, you will need to make the same path on the local file system at '/user/al/mahout-tmp'. Does that work for people do you think? > Support alternate temporary directories in example scripts. > --- > > Key: MAHOUT-1794 > URL: https://issues.apache.org/jira/browse/MAHOUT-1794 > Project: Mahout > Issue Type: Improvement > Components: Examples >Affects Versions: 0.11.1 >Reporter: Albert Chu >Assignee: Andrew Musselman >Priority: Minor > Fix For: 0.12.0 > > > In many of the example scripts, a directory in /tmp (e.g. > /tmp/mahout-work-dir) is created for use as temporary scratch space, such as > to store data files that are downloaded. > In a number of HPC environments, /tmp may not exist or /tmp may be very small > b/c local disk drives don't exist. It'd be convenient to be able to specify > an alternate directory to use as scratch space. > Pull request via Github to be sent shortly. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1762) Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
[
https://issues.apache.org/jira/browse/MAHOUT-1762?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199583#comment-15199583
]
Suneel Marthi commented on MAHOUT-1762:
---
[~pferrel] I don't think we would have time to address this for 0.12.0 and
should be punned to a subsequent release. But we really need this fixed in the
minor release following 0.12.0.
> Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
> ---
>
> Key: MAHOUT-1762
> URL: https://issues.apache.org/jira/browse/MAHOUT-1762
> Project: Mahout
> Issue Type: Improvement
> Components: spark
>Reporter: Sergey Tryuber
>Assignee: Pat Ferrel
> Fix For: 1.0.0
>
>
> [spark-defaults.conf|http://spark.apache.org/docs/latest/configuration.html#dynamically-loading-spark-properties]
> is aimed to contain global configuration for Spark cluster. For example, in
> our HDP2.2 environment it contains:
> {noformat}
> spark.driver.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> spark.yarn.am.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> {noformat}
> and there are many other good things. Actually it is expected that when a
> user starts Spark Shell, it will be working fine. Unfortunately this does not
> happens with Mahout Spark Shell, because it ignores spark configuration and
> user has to copy-past lots of options into _MAHOUT_OPTS_.
> This happens because
> [org.apache.mahout.sparkbindings.shell.Main|https://github.com/apache/mahout/blob/master/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala]
> is executed directly in [initialization
> script|https://github.com/apache/mahout/blob/master/bin/mahout]:
> {code}
> "$JAVA" $JAVA_HEAP_MAX $MAHOUT_OPTS -classpath "$CLASSPATH"
> "org.apache.mahout.sparkbindings.shell.Main" $@
> {code}
> In contrast, in Spark shell is indirectly invoked through spark-submit in
> [spark-shell|https://github.com/apache/spark/blob/master/bin/spark-shell]
> script:
> {code}
> "$FWDIR"/bin/spark-submit --class org.apache.spark.repl.Main "$@"
> {code}
> [SparkSubmit|https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala]
> contains an additional initialization layer for loading properties file (see
> SparkSubmitArguments#mergeDefaultSparkProperties method).
> So there are two possible solutions:
> * use proper Spark-like initialization logic
> * use thin envelope like it is in H2O Sparkling Water
> ([sparkling-shell|https://github.com/h2oai/sparkling-water/blob/master/bin/sparkling-shell])
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1799) Read null row vectors from file in TextDelimeterReaderWriter driver
[ https://issues.apache.org/jira/browse/MAHOUT-1799?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199770#comment-15199770 ] Pat Ferrel commented on MAHOUT-1799: Can't test this or even merge it right now so if someone else can merge, great otherwise is doesn't seem like a requirement for release and so unless someone speaks up I'll push to 1.0 > Read null row vectors from file in TextDelimeterReaderWriter driver > --- > > Key: MAHOUT-1799 > URL: https://issues.apache.org/jira/browse/MAHOUT-1799 > Project: Mahout > Issue Type: Improvement > Components: spark >Reporter: Jussi Jousimo >Assignee: Pat Ferrel >Priority: Minor > Fix For: 1.0.0 > > > Since some row vectors in a sparse matrix can be null, Mahout writes them out > to a file with the row label only. However, Mahout cannot read these files, > but throws an exception when it encounters a label-only row. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Resolved] (MAHOUT-1629) Mahout cvb on AWS EMR: p(topic|docId) doesn't make sense when using s3 folder as --input
[
https://issues.apache.org/jira/browse/MAHOUT-1629?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Suneel Marthi resolved MAHOUT-1629.
---
Resolution: Cannot Reproduce
Fix Version/s: 0.12.0
> Mahout cvb on AWS EMR: p(topic|docId) doesn't make sense when using s3 folder
> as --input
>
>
> Key: MAHOUT-1629
> URL: https://issues.apache.org/jira/browse/MAHOUT-1629
> Project: Mahout
> Issue Type: Bug
> Components: Clustering
>Affects Versions: 0.9
> Environment: AWS EMR with AMI 3.2.3
>Reporter: Markus Paaso
>Assignee: Andrew Musselman
> Labels: legacy
> Fix For: 0.12.0
>
>
> When running 'mahout cvb' command on AWS EMR having option --input with value
> like s3://mybucket/input/ or s3://mybucket/input/* (7 input files in my case)
> the content of doc-topic output is really non-sense. It seems like the docIds
> in doc-topic output are shuffled. But the topic model output (p(term|topic)
> for each topic) looks still fine.
> The workaround is to first copy input files from s3 to cluster's hdfs with
> command:
> {code:none}hadoop fs -cp s3://mybucket/input /input{code}
> and then running mahout cvb with option --input /input .
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned (mapBlock Checkpointing issue)
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15202159#comment-15202159
]
Andrew Palumbo commented on MAHOUT-1810:
As far as I know there is no currently no Caching in the same sense as is done
in Spark. I've been told this is to come in future releases.
I am just getting familiar with the Flink engine. I wonder if we can use the
{{DistributedFileCache}}?
https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/api/common/cache/DistributedCache.DistributedCacheEntry.html
> Failing test in flink-bindings: A + B Identically partitioned (mapBlock
> Checkpointing issue)
>
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A + B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Comment Edited] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200163#comment-15200163
]
Andrew Palumbo edited comment on MAHOUT-1810 at 3/17/16 7:57 PM:
-
>From the above tests though we can see that in the case of {{B %*% A.t}} where
>{{B}} is a checkpointed non-deterministic matrix, Checkpointing is creating a
>barrier. I should have illustrated this better by Using A+B and also
>providing failure case. (I was also using it as a sanity test for FlinkOpAtB)
>But it does in fact fail without the checkpoint of {{B}}.
It seems that checkpointing the mapBlock result is the culprit. Maybe this
speaks to your point about intermediate caching.
was (Author: andrew_palumbo):
>From the above tests though we can see that in the case of{{ B %*% A.t}} where
>B is a checkpointed non-deterministic matrix, Checkpointing is creating a
>barrier. I should have illustrated this better by Using A+B and also
>providing failure case. (I was also using it as a sanity test for FlinkOpAtB)
>But it does in fact fail without the checkpoint of B.
It seems that checkpointing the mapBlock result is the culprit. Maybe this
speaks to your point about intermediate caching.
> Failing test in flink-bindings: A + B Identically partitioned
> -
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Assigned] (MAHOUT-1810) Failing test in flink-bindings: A %*% B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Palumbo reassigned MAHOUT-1810:
--
Assignee: Andrew Palumbo
> Failing test in flink-bindings: A %*% B Identically partitioned
> ---
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Updated] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned (mapBlock Checkpointing issue)
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Palumbo updated MAHOUT-1810:
---
Summary: Failing test in flink-bindings: A + B Identically partitioned
(mapBlock Checkpointing issue) (was: Failing test in flink-bindings: A + B
Identically partitioned)
> Failing test in flink-bindings: A + B Identically partitioned (mapBlock
> Checkpointing issue)
>
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A + B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Comment Edited] (MAHOUT-1809) Failing tests in Flink-bindings: dals and dspca
[
https://issues.apache.org/jira/browse/MAHOUT-1809?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15201863#comment-15201863
]
Andrew Palumbo edited comment on MAHOUT-1809 at 3/18/16 6:09 PM:
-
{code}
test("dspca") {
val rnd = RandomUtils.getRandom
// Number of points
val m = 500
// Length of actual spectrum
val spectrumLen = 40
val spectrum = dvec((0 until spectrumLen).map(x => 300.0 * exp(-x) max
1e-3))
printf("spectrum:%s\n", spectrum)
val (u, _) = qr(new SparseRowMatrix(m, spectrumLen) :=
((r, c, v) => if (rnd.nextDouble() < 0.2) 0 else rnd.nextDouble() + 5.0))
// PCA Rotation matrix -- should also be orthonormal.
val (tr, _) = qr(Matrices.symmetricUniformView(spectrumLen, spectrumLen,
rnd.nextInt) - 10.0)
val input = (u %*%: diagv(spectrum)) %*% tr.t
val drmInput = drmParallelize(m = input, numPartitions = 2)
// Calculate just first 10 principal factors and reduce dimensionality.
// Since we assert just validity of the s-pca, not stochastic error, we
bump p parameter to
// ensure to zero stochastic error and assert only functional correctness
of the method's pca-
// specific additions.
val k = 10
// Calculate just first 10 principal factors and reduce dimensionality.
var (drmPCA, _, s) = dspca(drmA = drmInput, k = 10, p = spectrumLen, q = 1)
// Un-normalized pca data:
drmPCA = drmPCA %*% diagv(s)
val pca = drmPCA.checkpoint(CacheHint.NONE).collect
// Of course, once we calculated the pca, the spectrum is going to be
different since our originally
// generated input was not centered. So here, we'd just brute-solve pca to
verify
val xi = input.colMeans()
for (r <- 0 until input.nrow) input(r, ::) -= xi
var (pcaControl, _, sControl) = svd(m = input)
pcaControl = (pcaControl %*%: diagv(sControl))(::, 0 until k)
printf("pca:\n%s\n", pca(0 until 10, 0 until 10))
printf("pcaControl:\n%s\n", pcaControl(0 until 10, 0 until 10))
(pca(0 until 10, 0 until 10).norm - pcaControl(0 until 10, 0 until
10).norm).abs should be < 1E-5
}
{code}
This is the base test which is currently passing on Spark and H20 and failing
in Flink.
By checkpointing {{drmInput}} immediately upon parallelization, the test will
pass:
{code}
val drmInput = drmParallelize(m = input, numPartitions = 2).checkpoint()
{code}
was (Author: andrew_palumbo):
{code}
test("dspca") {
val rnd = RandomUtils.getRandom
// Number of points
val m = 500
// Length of actual spectrum
val spectrumLen = 40
val spectrum = dvec((0 until spectrumLen).map(x => 300.0 * exp(-x) max
1e-3))
printf("spectrum:%s\n", spectrum)
val (u, _) = qr(new SparseRowMatrix(m, spectrumLen) :=
((r, c, v) => if (rnd.nextDouble() < 0.2) 0 else rnd.nextDouble() + 5.0))
// PCA Rotation matrix -- should also be orthonormal.
val (tr, _) = qr(Matrices.symmetricUniformView(spectrumLen, spectrumLen,
rnd.nextInt) - 10.0)
val input = (u %*%: diagv(spectrum)) %*% tr.t
val drmInput = drmParallelize(m = input, numPartitions = 2)
// Calculate just first 10 principal factors and reduce dimensionality.
// Since we assert just validity of the s-pca, not stochastic error, we
bump p parameter to
// ensure to zero stochastic error and assert only functional correctness
of the method's pca-
// specific additions.
val k = 10
// Calculate just first 10 principal factors and reduce dimensionality.
var (drmPCA, _, s) = dspca(drmA = drmInput, k = 10, p = spectrumLen, q = 1)
// Un-normalized pca data:
drmPCA = drmPCA %*% diagv(s)
val pca = drmPCA.checkpoint(CacheHint.NONE).collect
// Of course, once we calculated the pca, the spectrum is going to be
different since our originally
// generated input was not centered. So here, we'd just brute-solve pca to
verify
val xi = input.colMeans()
for (r <- 0 until input.nrow) input(r, ::) -= xi
var (pcaControl, _, sControl) = svd(m = input)
pcaControl = (pcaControl %*%: diagv(sControl))(::, 0 until k)
printf("pca:\n%s\n", pca(0 until 10, 0 until 10))
printf("pcaControl:\n%s\n", pcaControl(0 until 10, 0 until 10))
(pca(0 until 10, 0 until 10).norm - pcaControl(0 until 10, 0 until
10).norm).abs should be < 1E-5
}
{code}
This is the base test which is currently passing on Spark and H20 and failing
in Flink.
By checkpointing {{drmInout}} immediately upon parallelization, the test will
pass:
{code}
val drmInput = drmParallelize(m = input, numPartitions = 2).checkpoint()
{code}
> Failing tests in Flink-bindings: dals and dspca
> ---
>
> Key: MAHOUT-1809
> URL: https://issues.apache.org/jira/browse/MAHOUT-1809
> Project: Mahout
> Issue Type: Bug
>
[jira] [Commented] (MAHOUT-1668) Automate release process
[ https://issues.apache.org/jira/browse/MAHOUT-1668?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200043#comment-15200043 ] Suneel Marthi commented on MAHOUT-1668: --- No progress on this issue, resolving this for now as 'Won't Fix'. Feel free to create a new jira when you are ready to work on this. > Automate release process > > > Key: MAHOUT-1668 > URL: https://issues.apache.org/jira/browse/MAHOUT-1668 > Project: Mahout > Issue Type: Task >Affects Versions: 0.9, 0.10.0 >Reporter: Stevo Slavic >Assignee: Stevo Slavic > Fix For: 0.12.0 > > > 0.10.0 will be first release since project switched to git. Some changes have > to be made in build scripts to support the release process, the Apache way. > As consequence, how-to-make-release docs will likely need to be updated as > well. Also, it would be nice to automate release process as much as possible, > e.g. via dedicated Jenkins build job(s), so it's easy for any committer to > cut out a release for vote, and after vote either finalize release or easily > make a new RC - this will enable us to release faster and more often. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Resolved] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Palumbo resolved MAHOUT-1815.
Resolution: Fixed
merged to flinl-binding
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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)
> {code}
--
This message was sent by Atlassian JIRA
(v6.3
[jira] [Resolved] (MAHOUT-1668) Automate release process
[ https://issues.apache.org/jira/browse/MAHOUT-1668?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Suneel Marthi resolved MAHOUT-1668. --- Resolution: Won't Fix Fix Version/s: (was: 1.0.0) 0.12.0 > Automate release process > > > Key: MAHOUT-1668 > URL: https://issues.apache.org/jira/browse/MAHOUT-1668 > Project: Mahout > Issue Type: Task >Affects Versions: 0.9, 0.10.0 >Reporter: Stevo Slavic >Assignee: Stevo Slavic > Fix For: 0.12.0 > > > 0.10.0 will be first release since project switched to git. Some changes have > to be made in build scripts to support the release process, the Apache way. > As consequence, how-to-make-release docs will likely need to be updated as > well. Also, it would be nice to automate release process as much as possible, > e.g. via dedicated Jenkins build job(s), so it's easy for any committer to > cut out a release for vote, and after vote either finalize release or easily > make a new RC - this will enable us to release faster and more often. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Palumbo updated MAHOUT-1810:
---
Description:
the {{A + B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
fails. This test failure likely indicates an issue with Flink's Checkpointing
or mapBlock operator:
{code}
test("C = A + B, identically partitioned") {
val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
val A = drmParallelize(inCoreA, numPartitions = 2)
// Create B which would be identically partitioned to A. mapBlock() by
default will do the trick.
val B = A.mapBlock() {
case (keys, block) =>
val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
keys -> bBlock
}
// Prevent repeated computation non-determinism
// flink problem is here... checkpoint is not doing what it should
.checkpoint()
val inCoreB = B.collect
printf("A=\n%s\n", inCoreA)
printf("B=\n%s\n", inCoreB)
val C = A + B
val inCoreC = C.collect
printf("C=\n%s\n", inCoreC)
// Actual
val inCoreCControl = inCoreA + inCoreB
(inCoreC - inCoreCControl).norm should be < 1E-10
}
{code}
The output shous clearly that the line:
{code}
val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
{code}
in the {{mapBlock}} closure is being calculated more than once.
Output:
{code}
A=
{
0 => {0:1.0,1:2.0,2:3.0}
1 => {0:3.0,1:4.0,2:5.0}
2 => {0:5.0,1:6.0,2:7.0}
}
B=
{
0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
}
C=
{
0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
}
{code}
was:
the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
fails. This test failure likely indicates an issue with Flink's Checkpointing
or mapBlock operator:
{code}
test("C = A + B, identically partitioned") {
val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
val A = drmParallelize(inCoreA, numPartitions = 2)
// Create B which would be identically partitioned to A. mapBlock() by
default will do the trick.
val B = A.mapBlock() {
case (keys, block) =>
val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
keys -> bBlock
}
// Prevent repeated computation non-determinism
// flink problem is here... checkpoint is not doing what it should
.checkpoint()
val inCoreB = B.collect
printf("A=\n%s\n", inCoreA)
printf("B=\n%s\n", inCoreB)
val C = A + B
val inCoreC = C.collect
printf("C=\n%s\n", inCoreC)
// Actual
val inCoreCControl = inCoreA + inCoreB
(inCoreC - inCoreCControl).norm should be < 1E-10
}
{code}
The output shous clearly that the line:
{code}
val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
{code}
in the {{mapBlock}} closure is being calculated more than once.
Output:
{code}
A=
{
0 => {0:1.0,1:2.0,2:3.0}
1 => {0:3.0,1:4.0,2:5.0}
2 => {0:5.0,1:6.0,2:7.0}
}
B=
{
0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
}
C=
{
0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
}
{code}
> Failing test in flink-bindings: A + B Identically partitioned
> -
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A + B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
>
[jira] [Updated] (MAHOUT-1705) Verify dependencies in job jar for mahout-examples
[ https://issues.apache.org/jira/browse/MAHOUT-1705?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Andrew Musselman updated MAHOUT-1705: - Fix Version/s: (was: 0.12.0) 0.12.1 > Verify dependencies in job jar for mahout-examples > -- > > Key: MAHOUT-1705 > URL: https://issues.apache.org/jira/browse/MAHOUT-1705 > Project: Mahout > Issue Type: Improvement >Affects Versions: 0.10.0 >Reporter: Andrew Palumbo >Assignee: Andrew Musselman > Fix For: 0.12.1 > > > mahout-example-*-job.jar is around ~56M, and may package unused runtime > libraries. We need to go through this and make sure that there is nothing > unneeded or redundant. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1773) Fix cluster-syntheticcontrol.sh for HDFS synthax
[ https://issues.apache.org/jira/browse/MAHOUT-1773?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200737#comment-15200737 ] ASF GitHub Bot commented on MAHOUT-1773: Github user asfgit closed the pull request at: https://github.com/apache/mahout/pull/157 > Fix cluster-syntheticcontrol.sh for HDFS synthax > > > Key: MAHOUT-1773 > URL: https://issues.apache.org/jira/browse/MAHOUT-1773 > Project: Mahout > Issue Type: Bug > Components: Clustering >Affects Versions: 0.11.0 >Reporter: Eduardo Niemeyer >Assignee: Andrew Musselman >Priority: Trivial > Fix For: 0.11.1 > > Original Estimate: 5m > Remaining Estimate: 5m > > Added '/' to some lines, so it can check the root of the HDFS, where it's > possible to use -mkdir, -put and others. Otherwise the example keeps failing > (Tested on Hadoop 2.4.1). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A %*% B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200134#comment-15200134
]
Andrew Palumbo commented on MAHOUT-1810:
Thanks, I will check back in with you on the mapBlock join fix when I'm more
focused on this issue. I have MAHOUT-1815 more in my head right now, but I am
suspecting that Flink partitioning has something to do with all of our major
bugs.
I don't believe that the Flink engine supports zips for DataSets. In the case
of Flink AewB a join is always done via {{DataSet.coGroup}}:
{code}
val rowsA = A.asRowWise.ds
val rowsB = B.asRowWise.ds
implicit val kTag = op.keyClassTag
val res: DataSet[(K, Vector)] =
rowsA
.coGroup(rowsB)
.where(0)
.equalTo(0) {
(left, right, out: Collector[(K, Vector)]) =>
(left.toIterable.headOption, right.toIterable.headOption) match {
case (Some((idx, a)), Some((_, b))) => out.collect((idx,
function(a, b)))
case (None, Some(b)) => out.collect(b)
case (Some(a), None) => out.collect(a)
case (None, None) => throw new RuntimeException("At least one side
of the co group " +
"must be non-empty.")
}
}
{code}
> Failing test in flink-bindings: A %*% B Identically partitioned
> ---
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A %*% B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199914#comment-15199914
]
Dmitriy Lyubimov commented on MAHOUT-1810:
--
Another note is that checkpoint() (at least in case of spark) would not prevent
computation non-determinism in case a partition is lost and subsequently
recomputed.
it _may_ have effect on double executions if engine indeed recomputes the input
A again as part of A+B so maybe yes checkpoint is not doing what it is supposed
to do for Flink, i.e., does not create optimization barrier here or/and does
not cache intermediate result by default.
> Failing test in flink-bindings: A %*% B Identically partitioned
> ---
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A %*% B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned (mapBlock Checkpointing issue)
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15202129#comment-15202129
]
Dmitriy Lyubimov commented on MAHOUT-1810:
--
so if checkpoint doesn't cache, (which is the intent to get rid of determinism
in this test), it is formal non-adherence to the contract of checkpoint and
checkpoint caching capabilities (parameters CacheHint).
So you are saying there's no way to cope with this?
I think, in the worst case, the solution should seek dumping intermediate
checkpoint (for cache hints other than None) to dfs or in-memory file system.
various people are telling me that dfs can now have a pretty sizeable local
cache configured too, so persistence is not so bad (but not as good as keeping
object trees in the same jvm, of course).
> Failing test in flink-bindings: A + B Identically partitioned (mapBlock
> Checkpointing issue)
>
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A + B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Resolved] (MAHOUT-1762) Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
[
https://issues.apache.org/jira/browse/MAHOUT-1762?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Pat Ferrel resolved MAHOUT-1762.
Resolution: Won't Fix
We don't know of anything this blocks and moving to using sparksubmit was voted
down, which only applies to Mahout CLI drivers anyway. All CLI drivers support
passthrough of arbitrary key=value pairs, which go into the SparkConf and when
using Mahout as a Lib you can create any arbitrary SparkConf.
Will not fix unless someone can explain the need.
> Pick up $SPARK_HOME/conf/spark-defaults.conf on startup
> ---
>
> Key: MAHOUT-1762
> URL: https://issues.apache.org/jira/browse/MAHOUT-1762
> Project: Mahout
> Issue Type: Improvement
> Components: spark
>Reporter: Sergey Tryuber
>Assignee: Pat Ferrel
> Fix For: 1.0.0
>
>
> [spark-defaults.conf|http://spark.apache.org/docs/latest/configuration.html#dynamically-loading-spark-properties]
> is aimed to contain global configuration for Spark cluster. For example, in
> our HDP2.2 environment it contains:
> {noformat}
> spark.driver.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> spark.yarn.am.extraJavaOptions -Dhdp.version=2.2.0.0–2041
> {noformat}
> and there are many other good things. Actually it is expected that when a
> user starts Spark Shell, it will be working fine. Unfortunately this does not
> happens with Mahout Spark Shell, because it ignores spark configuration and
> user has to copy-past lots of options into _MAHOUT_OPTS_.
> This happens because
> [org.apache.mahout.sparkbindings.shell.Main|https://github.com/apache/mahout/blob/master/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala]
> is executed directly in [initialization
> script|https://github.com/apache/mahout/blob/master/bin/mahout]:
> {code}
> "$JAVA" $JAVA_HEAP_MAX $MAHOUT_OPTS -classpath "$CLASSPATH"
> "org.apache.mahout.sparkbindings.shell.Main" $@
> {code}
> In contrast, in Spark shell is indirectly invoked through spark-submit in
> [spark-shell|https://github.com/apache/spark/blob/master/bin/spark-shell]
> script:
> {code}
> "$FWDIR"/bin/spark-submit --class org.apache.spark.repl.Main "$@"
> {code}
> [SparkSubmit|https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala]
> contains an additional initialization layer for loading properties file (see
> SparkSubmitArguments#mergeDefaultSparkProperties method).
> So there are two possible solutions:
> * use proper Spark-like initialization logic
> * use thin envelope like it is in H2O Sparkling Water
> ([sparkling-shell|https://github.com/h2oai/sparkling-water/blob/master/bin/sparkling-shell])
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1810) Failing test in flink-bindings: A + B Identically partitioned
[
https://issues.apache.org/jira/browse/MAHOUT-1810?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15201713#comment-15201713
]
Andrew Palumbo commented on MAHOUT-1810:
Flink does cache intermediate results of a task by default:
https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/config.html
However, caching of a DRM to memory is not possible and the call to
{{CheckpointedFlinkDrm.cache(...)}} is only stubbed out.
However this is the same situation in H2O and this test passes without issue
there.
> Failing test in flink-bindings: A + B Identically partitioned
> -
>
> Key: MAHOUT-1810
> URL: https://issues.apache.org/jira/browse/MAHOUT-1810
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
> Fix For: 0.12.0
>
>
> the {{A + B, Identically Partitioned}} test in the Flink RLikeDrmOpsSuite
> fails. This test failure likely indicates an issue with Flink's
> Checkpointing or mapBlock operator:
> {code}
> test("C = A + B, identically partitioned") {
> val inCoreA = dense((1, 2, 3), (3, 4, 5), (5, 6, 7))
> val A = drmParallelize(inCoreA, numPartitions = 2)
> // Create B which would be identically partitioned to A. mapBlock() by
> default will do the trick.
> val B = A.mapBlock() {
> case (keys, block) =>
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> keys -> bBlock
> }
> // Prevent repeated computation non-determinism
> // flink problem is here... checkpoint is not doing what it should
> .checkpoint()
> val inCoreB = B.collect
> printf("A=\n%s\n", inCoreA)
> printf("B=\n%s\n", inCoreB)
> val C = A + B
> val inCoreC = C.collect
> printf("C=\n%s\n", inCoreC)
> // Actual
> val inCoreCControl = inCoreA + inCoreB
> (inCoreC - inCoreCControl).norm should be < 1E-10
> }
> {code}
> The output shous clearly that the line:
> {code}
> val bBlock = block.like() := { (r, c, v) => util.Random.nextDouble()}
> {code}
> in the {{mapBlock}} closure is being calculated more than once.
> Output:
> {code}
> A=
> {
> 0 => {0:1.0,1:2.0,2:3.0}
> 1 => {0:3.0,1:4.0,2:5.0}
> 2 => {0:5.0,1:6.0,2:7.0}
> }
> B=
> {
> 0 => {0:0.26203398262809574,1:0.22561543461472167,2:0.23229669514522655}
> 1 => {0:0.1638068194515867,1:0.18751822418846575,2:0.20586366231381614}
> 2 => {0:0.9279465706239354,1:0.2963513448240057,2:0.8866928923235948}
> }
> C=
> {
> 0 => {0:1.7883652623225594,1:2.6401297718606216,2:3.0023341959374195}
> 1 => {0:3.641411452208408,1:4.941233165480053,2:5.381282338548803}
> 2 => {0:5.707434148862531,1:6.022780876943659,2:7.149772825494352}
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (MAHOUT-1815) dsqDist(X,Y) and dsqDist(X) failing in flink tests.
[
https://issues.apache.org/jira/browse/MAHOUT-1815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15200529#comment-15200529
]
ASF GitHub Bot commented on MAHOUT-1815:
Github user andrewpalumbo commented on the pull request:
https://github.com/apache/mahout/pull/197#issuecomment-198110346
merged to flink-binding
> dsqDist(X,Y) and dsqDist(X) failing in flink tests.
> ---
>
> Key: MAHOUT-1815
> URL: https://issues.apache.org/jira/browse/MAHOUT-1815
> Project: Mahout
> Issue Type: Bug
>Reporter: Andrew Palumbo
>Assignee: Andrew Palumbo
>Priority: Blocker
>
> {code}
> 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
> }
> {code}
> And
> {code}
> 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
> }
> {code}
> are both failing in flink tests with {{arrayOutOfBounds}} Exceptions:
> {code}
> 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 or
[jira] [Commented] (MAHOUT-1799) Read null row vectors from file in TextDelimeterReaderWriter driver
[ https://issues.apache.org/jira/browse/MAHOUT-1799?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15199763#comment-15199763 ] ASF GitHub Bot commented on MAHOUT-1799: Github user pferrel commented on the pull request: https://github.com/apache/mahout/pull/182#issuecomment-197944335 Looks fine, I can't test it now. There may be cases where a row is null in the input but this treats those as non-error conditions so will not complain in logs. I'm a bit ambivalent about the need for this, can @statguy explain why a null row is not an error, might is be better to print a warning? > Read null row vectors from file in TextDelimeterReaderWriter driver > --- > > Key: MAHOUT-1799 > URL: https://issues.apache.org/jira/browse/MAHOUT-1799 > Project: Mahout > Issue Type: Improvement > Components: spark >Reporter: Jussi Jousimo >Assignee: Pat Ferrel >Priority: Minor > Fix For: 0.12.0 > > > Since some row vectors in a sparse matrix can be null, Mahout writes them out > to a file with the row label only. However, Mahout cannot read these files, > but throws an exception when it encounters a label-only row. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
