MAHOUT-1764: Flink: standard backend tests
Project: http://git-wip-us.apache.org/repos/asf/mahout/repo Commit: http://git-wip-us.apache.org/repos/asf/mahout/commit/ceb1f050 Tree: http://git-wip-us.apache.org/repos/asf/mahout/tree/ceb1f050 Diff: http://git-wip-us.apache.org/repos/asf/mahout/diff/ceb1f050 Branch: refs/heads/flink-binding Commit: ceb1f050cfaf50548bcad83f661d0ad01fe96122 Parents: 4bafc76 Author: Alexey Grigorev <[email protected]> Authored: Thu Aug 13 17:03:40 2015 +0200 Committer: Alexey Grigorev <[email protected]> Committed: Fri Sep 25 17:41:58 2015 +0200 ---------------------------------------------------------------------- .../apache/mahout/flinkbindings/package.scala | 2 +- .../flinkbindings/DistributedFlinkSuit.scala | 47 ------ .../flinkbindings/DistributedFlinkSuite.scala | 47 ++++++ .../mahout/flinkbindings/RLikeOpsSuite.scala | 2 +- .../mahout/flinkbindings/UseCasesSuite.scala | 2 +- .../mahout/flinkbindings/blas/LATestSuit.scala | 158 ------------------- .../mahout/flinkbindings/blas/LATestSuite.scala | 158 +++++++++++++++++++ .../DistributedDecompositionsSuite.scala | 19 +++ .../standard/DrmLikeOpsSuite.scala | 19 +++ .../flinkbindings/standard/DrmLikeSuite.scala | 19 +++ .../standard/RLikeDrmOpsSuite.scala | 19 +++ 11 files changed, 284 insertions(+), 208 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/main/scala/org/apache/mahout/flinkbindings/package.scala ---------------------------------------------------------------------- diff --git a/flink/src/main/scala/org/apache/mahout/flinkbindings/package.scala b/flink/src/main/scala/org/apache/mahout/flinkbindings/package.scala index 955d8b1..aa253ab 100644 --- a/flink/src/main/scala/org/apache/mahout/flinkbindings/package.scala +++ b/flink/src/main/scala/org/apache/mahout/flinkbindings/package.scala @@ -18,7 +18,7 @@ */ package org.apache.mahout -import scala.Array.canBuildFrom +import scala.Array._ import scala.reflect.ClassTag import org.apache.flink.api.common.functions.FilterFunction http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/DistributedFlinkSuit.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/DistributedFlinkSuit.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/DistributedFlinkSuit.scala deleted file mode 100644 index 126a8f4..0000000 --- a/flink/src/test/scala/org/apache/mahout/flinkbindings/DistributedFlinkSuit.scala +++ /dev/null @@ -1,47 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.mahout.flinkbindings - -import org.apache.flink.api.java.ExecutionEnvironment -import org.apache.mahout.flinkbindings._ -import org.apache.mahout.math.drm.DistributedContext -import org.apache.mahout.test.DistributedMahoutSuite -import org.scalatest.Suite - - -trait DistributedFlinkSuit extends DistributedMahoutSuite { this: Suite => - - protected implicit var mahoutCtx: DistributedContext = _ - protected var env: ExecutionEnvironment = null - - def initContext() { - env = ExecutionEnvironment.getExecutionEnvironment - mahoutCtx = wrapContext(env) - } - - override def beforeEach() { - initContext() - } - - override def afterEach() { - super.afterEach() -// env.execute("Mahout Flink Binding Test Suite") - } - -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/DistributedFlinkSuite.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/DistributedFlinkSuite.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/DistributedFlinkSuite.scala new file mode 100644 index 0000000..dd76ff4 --- /dev/null +++ b/flink/src/test/scala/org/apache/mahout/flinkbindings/DistributedFlinkSuite.scala @@ -0,0 +1,47 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.mahout.flinkbindings + +import org.apache.flink.api.java.ExecutionEnvironment +import org.apache.mahout.flinkbindings._ +import org.apache.mahout.math.drm.DistributedContext +import org.apache.mahout.test.DistributedMahoutSuite +import org.scalatest.Suite + + +trait DistributedFlinkSuite extends DistributedMahoutSuite { this: Suite => + + protected implicit var mahoutCtx: DistributedContext = _ + protected var env: ExecutionEnvironment = null + + def initContext() { + env = ExecutionEnvironment.getExecutionEnvironment + mahoutCtx = wrapContext(env) + } + + override def beforeEach() { + initContext() + } + + override def afterEach() { + super.afterEach() +// env.execute("Mahout Flink Binding Test Suite") + } + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/RLikeOpsSuite.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/RLikeOpsSuite.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/RLikeOpsSuite.scala index fa924a9..99f6718 100644 --- a/flink/src/test/scala/org/apache/mahout/flinkbindings/RLikeOpsSuite.scala +++ b/flink/src/test/scala/org/apache/mahout/flinkbindings/RLikeOpsSuite.scala @@ -32,7 +32,7 @@ import org.slf4j.LoggerFactory @RunWith(classOf[JUnitRunner]) -class RLikeOpsSuite extends FunSuite with DistributedFlinkSuit { +class RLikeOpsSuite extends FunSuite with DistributedFlinkSuite { val LOGGER = LoggerFactory.getLogger(getClass()) http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/UseCasesSuite.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/UseCasesSuite.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/UseCasesSuite.scala index a144f6d..07d62dc 100644 --- a/flink/src/test/scala/org/apache/mahout/flinkbindings/UseCasesSuite.scala +++ b/flink/src/test/scala/org/apache/mahout/flinkbindings/UseCasesSuite.scala @@ -32,7 +32,7 @@ import org.scalatest.FunSuite import org.scalatest.junit.JUnitRunner @RunWith(classOf[JUnitRunner]) -class UseCasesSuite extends FunSuite with DistributedFlinkSuit { +class UseCasesSuite extends FunSuite with DistributedFlinkSuite { val LOGGER = LoggerFactory.getLogger(getClass()) http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/blas/LATestSuit.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/blas/LATestSuit.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/blas/LATestSuit.scala deleted file mode 100644 index 9f8a888..0000000 --- a/flink/src/test/scala/org/apache/mahout/flinkbindings/blas/LATestSuit.scala +++ /dev/null @@ -1,158 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.mahout.flinkbindings.blas - -import org.scalatest.FunSuite -import org.apache.mahout.math._ -import scalabindings._ -import RLikeOps._ -import drm._ -import org.apache.mahout.flinkbindings._ -import org.junit.runner.RunWith -import org.scalatest.junit.JUnitRunner -import org.apache.mahout.math.drm.logical.OpAx -import org.apache.mahout.flinkbindings.drm.CheckpointedFlinkDrm -import org.apache.mahout.flinkbindings.drm.RowsFlinkDrm -import org.apache.mahout.math.drm.logical._ - -@RunWith(classOf[JUnitRunner]) -class LATestSuit extends FunSuite with DistributedFlinkSuit { - - test("Ax blockified") { - val inCoreA = dense((1, 2, 3), (2, 3, 4), (3, 4, 5)) - val A = drmParallelize(m = inCoreA, numPartitions = 2) - val x: Vector = (0, 1, 2) - - val opAx = new OpAx(A, x) - val res = FlinkOpAx.blockifiedBroadcastAx(opAx, A) - val drm = new CheckpointedFlinkDrm(res.deblockify.ds) - val output = drm.collect - - val b = output(::, 0) - assert(b == dvec(8, 11, 14)) - } - - test("At sparseTrick") { - val inCoreA = dense((1, 2, 3), (2, 3, 4)) - val A = drmParallelize(m = inCoreA, numPartitions = 2) - - val opAt = new OpAt(A) - val res = FlinkOpAt.sparseTrick(opAt, A) - val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreA.ncol, _ncol=inCoreA.nrow) - val output = drm.collect - - assert((output - inCoreA.t).norm < 1e-6) - } - - test("AtB notZippable") { - val inCoreAt = dense((1, 2), (2, 3), (3, 4)) - - val At = drmParallelize(m = inCoreAt, numPartitions = 2) - - val inCoreB = dense((1, 2), (3, 4), (11, 4)) - val B = drmParallelize(m = inCoreB, numPartitions = 2) - - val opAtB = new OpAtB(At, B) - val res = FlinkOpAtB.notZippable(opAtB, At, B) - - val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreAt.ncol, _ncol=inCoreB.ncol) - val output = drm.collect - - val expected = inCoreAt.t %*% inCoreB - assert((output - expected).norm < 1e-6) - } - - test("AewScalar opScalarNoSideEffect") { - val inCoreA = dense((1, 2), (2, 3), (3, 4)) - val A = drmParallelize(m = inCoreA, numPartitions = 2) - val scalar = 5.0 - - val op = new OpAewScalar(A, scalar, "*") - val res = FlinkOpAewScalar.opScalarNoSideEffect(op, A, scalar) - - val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreA.nrow, _ncol=inCoreA.ncol) - val output = drm.collect - - val expected = inCoreA * scalar - assert((output - expected).norm < 1e-6) - } - - test("AewB rowWiseJoinNoSideEffect") { - val inCoreA = dense((1, 2), (2, 3), (3, 4)) - val A = drmParallelize(m = inCoreA, numPartitions = 2) - - val op = new OpAewB(A, A, "*") - val res = FlinkOpAewB.rowWiseJoinNoSideEffect(op, A, A) - - val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreA.nrow, _ncol=inCoreA.ncol) - val output = drm.collect - - assert((output - (inCoreA * inCoreA)).norm < 1e-6) - } - - test("Cbind") { - val inCoreA = dense((1, 2), (2, 3), (3, 4)) - val inCoreB = dense((4, 4), (5, 5), (6, 7)) - val A = drmParallelize(m = inCoreA, numPartitions = 2) - val B = drmParallelize(m = inCoreB, numPartitions = 2) - - val op = new OpCbind(A, B) - val res = FlinkOpCBind.cbind(op, A, B) - - val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreA.nrow, - _ncol=(inCoreA.ncol + inCoreB.ncol)) - val output = drm.collect - - val expected = dense((1, 2, 4, 4), (2, 3, 5, 5), (3, 4, 6, 7)) - assert((output - expected).norm < 1e-6) - } - - test("slice") { - val inCoreA = dense((1, 2), (2, 3), (3, 4), (4, 4), (5, 5), (6, 7)) - val A = drmParallelize(m = inCoreA, numPartitions = 2) - - val range = 2 until 5 - val op = new OpRowRange(A, range) - val res = FlinkOpRowRange.slice(op, A) - - val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=op.nrow, - _ncol=inCoreA.ncol) - val output = drm.collect - - val expected = inCoreA(2 until 5, ::) - assert((output - expected).norm < 1e-6) - } - - test("A times inCoreB") { - val inCoreA = dense((1, 2, 3), (2, 3, 1), (3, 4, 4), (4, 4, 5), (5, 5, 7), (6, 7, 11)) - val inCoreB = dense((2, 1), (3, 4), (5, 11)) - val A = drmParallelize(m = inCoreA, numPartitions = 2) - - val op = new OpTimesRightMatrix(A, inCoreB) - val res = FlinkOpTimesRightMatrix.drmTimesInCore(op, A, inCoreB) - - val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=op.nrow, - _ncol=op.ncol) - val output = drm.collect - - val expected = inCoreA %*% inCoreB - assert((output - expected).norm < 1e-6) - } - -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/blas/LATestSuite.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/blas/LATestSuite.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/blas/LATestSuite.scala new file mode 100644 index 0000000..2db7f91 --- /dev/null +++ b/flink/src/test/scala/org/apache/mahout/flinkbindings/blas/LATestSuite.scala @@ -0,0 +1,158 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.mahout.flinkbindings.blas + +import org.scalatest.FunSuite +import org.apache.mahout.math._ +import scalabindings._ +import RLikeOps._ +import drm._ +import org.apache.mahout.flinkbindings._ +import org.junit.runner.RunWith +import org.scalatest.junit.JUnitRunner +import org.apache.mahout.math.drm.logical.OpAx +import org.apache.mahout.flinkbindings.drm.CheckpointedFlinkDrm +import org.apache.mahout.flinkbindings.drm.RowsFlinkDrm +import org.apache.mahout.math.drm.logical._ + +@RunWith(classOf[JUnitRunner]) +class LATestSuite extends FunSuite with DistributedFlinkSuite { + + test("Ax blockified") { + val inCoreA = dense((1, 2, 3), (2, 3, 4), (3, 4, 5)) + val A = drmParallelize(m = inCoreA, numPartitions = 2) + val x: Vector = (0, 1, 2) + + val opAx = new OpAx(A, x) + val res = FlinkOpAx.blockifiedBroadcastAx(opAx, A) + val drm = new CheckpointedFlinkDrm(res.deblockify.ds) + val output = drm.collect + + val b = output(::, 0) + assert(b == dvec(8, 11, 14)) + } + + test("At sparseTrick") { + val inCoreA = dense((1, 2, 3), (2, 3, 4)) + val A = drmParallelize(m = inCoreA, numPartitions = 2) + + val opAt = new OpAt(A) + val res = FlinkOpAt.sparseTrick(opAt, A) + val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreA.ncol, _ncol=inCoreA.nrow) + val output = drm.collect + + assert((output - inCoreA.t).norm < 1e-6) + } + + test("AtB notZippable") { + val inCoreAt = dense((1, 2), (2, 3), (3, 4)) + + val At = drmParallelize(m = inCoreAt, numPartitions = 2) + + val inCoreB = dense((1, 2), (3, 4), (11, 4)) + val B = drmParallelize(m = inCoreB, numPartitions = 2) + + val opAtB = new OpAtB(At, B) + val res = FlinkOpAtB.notZippable(opAtB, At, B) + + val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreAt.ncol, _ncol=inCoreB.ncol) + val output = drm.collect + + val expected = inCoreAt.t %*% inCoreB + assert((output - expected).norm < 1e-6) + } + + test("AewScalar opScalarNoSideEffect") { + val inCoreA = dense((1, 2), (2, 3), (3, 4)) + val A = drmParallelize(m = inCoreA, numPartitions = 2) + val scalar = 5.0 + + val op = new OpAewScalar(A, scalar, "*") + val res = FlinkOpAewScalar.opScalarNoSideEffect(op, A, scalar) + + val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreA.nrow, _ncol=inCoreA.ncol) + val output = drm.collect + + val expected = inCoreA * scalar + assert((output - expected).norm < 1e-6) + } + + test("AewB rowWiseJoinNoSideEffect") { + val inCoreA = dense((1, 2), (2, 3), (3, 4)) + val A = drmParallelize(m = inCoreA, numPartitions = 2) + + val op = new OpAewB(A, A, "*") + val res = FlinkOpAewB.rowWiseJoinNoSideEffect(op, A, A) + + val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreA.nrow, _ncol=inCoreA.ncol) + val output = drm.collect + + assert((output - (inCoreA * inCoreA)).norm < 1e-6) + } + + test("Cbind") { + val inCoreA = dense((1, 2), (2, 3), (3, 4)) + val inCoreB = dense((4, 4), (5, 5), (6, 7)) + val A = drmParallelize(m = inCoreA, numPartitions = 2) + val B = drmParallelize(m = inCoreB, numPartitions = 2) + + val op = new OpCbind(A, B) + val res = FlinkOpCBind.cbind(op, A, B) + + val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=inCoreA.nrow, + _ncol=(inCoreA.ncol + inCoreB.ncol)) + val output = drm.collect + + val expected = dense((1, 2, 4, 4), (2, 3, 5, 5), (3, 4, 6, 7)) + assert((output - expected).norm < 1e-6) + } + + test("slice") { + val inCoreA = dense((1, 2), (2, 3), (3, 4), (4, 4), (5, 5), (6, 7)) + val A = drmParallelize(m = inCoreA, numPartitions = 2) + + val range = 2 until 5 + val op = new OpRowRange(A, range) + val res = FlinkOpRowRange.slice(op, A) + + val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=op.nrow, + _ncol=inCoreA.ncol) + val output = drm.collect + + val expected = inCoreA(2 until 5, ::) + assert((output - expected).norm < 1e-6) + } + + test("A times inCoreB") { + val inCoreA = dense((1, 2, 3), (2, 3, 1), (3, 4, 4), (4, 4, 5), (5, 5, 7), (6, 7, 11)) + val inCoreB = dense((2, 1), (3, 4), (5, 11)) + val A = drmParallelize(m = inCoreA, numPartitions = 2) + + val op = new OpTimesRightMatrix(A, inCoreB) + val res = FlinkOpTimesRightMatrix.drmTimesInCore(op, A, inCoreB) + + val drm = new CheckpointedFlinkDrm(res.deblockify.ds, _nrow=op.nrow, + _ncol=op.ncol) + val output = drm.collect + + val expected = inCoreA %*% inCoreB + assert((output - expected).norm < 1e-6) + } + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DistributedDecompositionsSuite.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DistributedDecompositionsSuite.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DistributedDecompositionsSuite.scala new file mode 100644 index 0000000..f13597a --- /dev/null +++ b/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DistributedDecompositionsSuite.scala @@ -0,0 +1,19 @@ +package org.apache.mahout.flinkbindings.standard + +import org.apache.mahout.flinkbindings._ +import org.apache.mahout.math._ +import org.apache.mahout.math.drm._ +import org.apache.mahout.math.drm.RLikeDrmOps._ +import org.apache.mahout.math.scalabindings._ +import org.apache.mahout.math.scalabindings.RLikeOps._ +import org.junit.runner.RunWith +import org.scalatest.FunSuite +import org.scalatest.junit.JUnitRunner +import org.apache.mahout.math.decompositions.DistributedDecompositionsSuiteBase + + +@RunWith(classOf[JUnitRunner]) +class DistributedDecompositionsSuite extends FunSuite with DistributedFlinkSuite + with DistributedDecompositionsSuiteBase { + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DrmLikeOpsSuite.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DrmLikeOpsSuite.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DrmLikeOpsSuite.scala new file mode 100644 index 0000000..7f6b2c8 --- /dev/null +++ b/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DrmLikeOpsSuite.scala @@ -0,0 +1,19 @@ +package org.apache.mahout.flinkbindings.standard + +import org.apache.mahout.flinkbindings._ +import org.apache.mahout.math._ +import org.apache.mahout.math.drm._ +import org.apache.mahout.math.drm.RLikeDrmOps._ +import org.apache.mahout.math.scalabindings._ +import org.apache.mahout.math.scalabindings.RLikeOps._ +import org.junit.runner.RunWith +import org.scalatest.FunSuite +import org.scalatest.junit.JUnitRunner +import org.apache.mahout.math.decompositions.DistributedDecompositionsSuiteBase + + +@RunWith(classOf[JUnitRunner]) +class DrmLikeOpsSuite extends FunSuite with DistributedFlinkSuite + with DrmLikeOpsSuiteBase { + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DrmLikeSuite.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DrmLikeSuite.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DrmLikeSuite.scala new file mode 100644 index 0000000..7cfc48b --- /dev/null +++ b/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/DrmLikeSuite.scala @@ -0,0 +1,19 @@ +package org.apache.mahout.flinkbindings.standard + +import org.apache.mahout.flinkbindings._ +import org.apache.mahout.math._ +import org.apache.mahout.math.drm._ +import org.apache.mahout.math.drm.RLikeDrmOps._ +import org.apache.mahout.math.scalabindings._ +import org.apache.mahout.math.scalabindings.RLikeOps._ +import org.junit.runner.RunWith +import org.scalatest.FunSuite +import org.scalatest.junit.JUnitRunner +import org.apache.mahout.math.decompositions.DistributedDecompositionsSuiteBase + + +@RunWith(classOf[JUnitRunner]) +class DrmLikeSuite extends FunSuite with DistributedFlinkSuite + with DrmLikeSuiteBase { + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/mahout/blob/ceb1f050/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/RLikeDrmOpsSuite.scala ---------------------------------------------------------------------- diff --git a/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/RLikeDrmOpsSuite.scala b/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/RLikeDrmOpsSuite.scala new file mode 100644 index 0000000..1ba03b1 --- /dev/null +++ b/flink/src/test/scala/org/apache/mahout/flinkbindings/standard/RLikeDrmOpsSuite.scala @@ -0,0 +1,19 @@ +package org.apache.mahout.flinkbindings.standard + +import org.apache.mahout.flinkbindings._ +import org.apache.mahout.math._ +import org.apache.mahout.math.drm._ +import org.apache.mahout.math.drm.RLikeDrmOps._ +import org.apache.mahout.math.scalabindings._ +import org.apache.mahout.math.scalabindings.RLikeOps._ +import org.junit.runner.RunWith +import org.scalatest.FunSuite +import org.scalatest.junit.JUnitRunner +import org.apache.mahout.math.decompositions.DistributedDecompositionsSuiteBase + + +@RunWith(classOf[JUnitRunner]) +class RLikeDrmOpsSuite extends FunSuite with DistributedFlinkSuite + with RLikeDrmOpsSuiteBase { + +} \ No newline at end of file
