You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2014/12/18 19:46:05 UTC
[69/82] [abbrv] incubator-flink git commit: Change integration tests
to reuse cluster in order to save startup and shutdown time.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala
index 219efdc..9ea3103 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/GroupReduceITCase.scala
@@ -27,705 +27,721 @@ import org.apache.flink.api.scala.util.CollectionDataSets.{CrazyNested, POJO, Mu
CustomType}
import org.apache.flink.compiler.PactCompiler
import org.apache.flink.configuration.Configuration
-import org.apache.flink.test.util.JavaProgramTestBase
+import org.apache.flink.core.fs.FileSystem.WriteMode
+import org.apache.flink.test.util.MultipleProgramsTestBase.ExecutionMode
+import org.apache.flink.test.util.{MultipleProgramsTestBase}
import org.apache.flink.util.Collector
+import org.hamcrest.core.{IsNot, IsEqual}
+import org.junit._
+import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
import org.junit.runners.Parameterized
-import org.junit.runners.Parameterized.Parameters
import scala.collection.JavaConverters._
-import scala.collection.mutable
import org.apache.flink.api.scala._
+@RunWith(classOf[Parameterized])
+class GroupReduceITCase(mode: ExecutionMode) extends MultipleProgramsTestBase(mode) {
+ private var resultPath: String = null
+ private var expected: String = null
+ private val _tempFolder = new TemporaryFolder()
-object GroupReduceProgs {
- var NUM_PROGRAMS: Int = 26
-
- def runProgram(progId: Int, resultPath: String, onCollection: Boolean): String = {
- progId match {
- case 1 =>
- /*
- * check correctness of groupReduce on tuples with key field selector
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env)
- val reduceDs = ds.groupBy(1).reduceGroup {
- in =>
- in.map(t => (t._1, t._2)).reduce((l, r) => (l._1 + r._1, l._2))
- }
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "1,1\n" + "5,2\n" + "15,3\n" + "34,4\n" + "65,5\n" + "111,6\n"
-
- case 2 =>
- /*
- * check correctness of groupReduce on tuples with multiple key field selector
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets
- .get5TupleDataSet(env)
- val reduceDs = ds.groupBy(4, 0).reduceGroup {
- in =>
- val (i, l, l2) = in
- .map( t => (t._1, t._2, t._5))
- .reduce((l, r) => (l._1, l._2 + r._2, l._3))
- (i, l, 0, "P-)", l2)
- }
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "1,1,0,P-),1\n" + "2,3,0,P-),1\n" + "2,2,0,P-),2\n" + "3,9,0,P-),2\n" + "3,6,0," +
- "P-),3\n" + "4,17,0,P-),1\n" + "4,17,0,P-),2\n" + "5,11,0,P-),1\n" + "5,29,0,P-)," +
- "2\n" + "5,25,0,P-),3\n"
-
- case 3 =>
- /*
- * check correctness of groupReduce on tuples with key field selector and group sorting
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.get3TupleDataSet(env)
- val reduceDs = ds.groupBy(1).sortGroup(2, Order.ASCENDING).reduceGroup {
- in =>
- in.reduce((l, r) => (l._1 + r._1, l._2, l._3 + "-" + r._3))
- }
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "1,1,Hi\n" +
- "5,2,Hello-Hello world\n" +
- "15,3,Hello world, how are you?-I am fine.-Luke Skywalker\n" +
- "34,4,Comment#1-Comment#2-Comment#3-Comment#4\n" +
- "65,5,Comment#5-Comment#6-Comment#7-Comment#8-Comment#9\n" +
- "111,6,Comment#10-Comment#11-Comment#12-Comment#13-Comment#14-Comment#15\n"
-
- case 4 =>
- /*
- * check correctness of groupReduce on tuples with key extractor
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env)
- val reduceDs = ds.groupBy(_._2).reduceGroup {
- in =>
- in.map(t => (t._1, t._2)).reduce((l, r) => (l._1 + r._1, l._2))
- }
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "1,1\n" + "5,2\n" + "15,3\n" + "34,4\n" + "65,5\n" + "111,6\n"
-
- case 5 =>
- /*
- * check correctness of groupReduce on custom type with type extractor
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.getCustomTypeDataSet(env)
- val reduceDs = ds.groupBy(_.myInt).reduceGroup {
- in =>
- val iter = in.toIterator
- val o = new CustomType
- val c = iter.next()
-
- o.myString = "Hello!"
- o.myInt = c.myInt
- o.myLong = c.myLong
-
- while (iter.hasNext) {
- val next = iter.next()
- o.myLong += next.myLong
- }
- o
- }
- reduceDs.writeAsText(resultPath)
- env.execute()
- "1,0,Hello!\n" + "2,3,Hello!\n" + "3,12,Hello!\n" + "4,30,Hello!\n" + "5,60," +
- "Hello!\n" + "6,105,Hello!\n"
-
- case 6 =>
- /*
- * check correctness of all-groupreduce for tuples
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env)
- val reduceDs = ds.reduceGroup {
- in =>
- var i = 0
- var l = 0L
- for (t <- in) {
- i += t._1
- l += t._2
- }
- (i, l, "Hello World")
- }
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "231,91,Hello World\n"
-
- case 7 =>
- /*
- * check correctness of all-groupreduce for custom types
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.getCustomTypeDataSet(env)
- val reduceDs = ds.reduceGroup {
- in =>
- val o = new CustomType(0, 0, "Hello!")
- for (t <- in) {
- o.myInt += t.myInt
- o.myLong += t.myLong
- }
- o
- }
- reduceDs.writeAsText(resultPath)
- env.execute()
- "91,210,Hello!"
-
- case 8 =>
- /*
- * check correctness of groupReduce with broadcast set
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val intDs = CollectionDataSets.getIntDataSet(env)
- val ds = CollectionDataSets.get3TupleDataSet(env)
- val reduceDs = ds.groupBy(1).reduceGroup(
- new RichGroupReduceFunction[(Int, Long, String), (Int, Long, String)] {
- private var f2Replace = ""
-
- override def open(config: Configuration) {
- val ints = this.getRuntimeContext.getBroadcastVariable[Int]("ints").asScala
- f2Replace = ints.sum + ""
- }
+ @Rule
+ def tempFolder = _tempFolder
- override def reduce(
- values: Iterable[(Int, Long, String)],
- out: Collector[(Int, Long, String)]): Unit = {
- var i: Int = 0
- var l: Long = 0L
- for (t <- values.asScala) {
- i += t._1
- l = t._2
- }
- out.collect((i, l, f2Replace))
- }
- }).withBroadcastSet(intDs, "ints")
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "1,1,55\n" + "5,2,55\n" + "15,3,55\n" + "34,4,55\n" + "65,5,55\n" + "111,6,55\n"
-
- case 9 =>
- /*
- * check correctness of groupReduce if UDF returns input objects multiple times and
- * changes it in between
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env)
- .map( t => MutableTuple3(t._1, t._2, t._3) )
- val reduceDs = ds.groupBy(1).reduceGroup {
- (in, out: Collector[MutableTuple3[Int, Long, String]]) =>
- for (t <- in) {
- if (t._1 < 4) {
- t._3 = "Hi!"
- t._1 += 10
- out.collect(t)
- t._1 += 10
- t._3 = "Hi again!"
- out.collect(t)
- }
- }
+ @Before
+ def before(): Unit = {
+ resultPath = tempFolder.newFile().toURI.toString
+ }
+
+ @After
+ def after: Unit = {
+ if(expected != null) {
+ compareResultsByLinesInMemory(expected, resultPath)
+ }
+ }
+
+ @Test
+ def testCorrectnessOfGroupReduceOnTuplesWithKeyFieldSelector: Unit = {
+ /*
+ * check correctness of groupReduce on tuples with key field selector
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.groupBy(1).reduceGroup {
+ in =>
+ in.map(t => (t._1, t._2)).reduce((l, r) => (l._1 + r._1, l._2))
+ }
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1\n" + "5,2\n" + "15,3\n" + "34,4\n" + "65,5\n" + "111,6\n"
+ }
+
+ @Test
+ def testCorrectnessOfGroupReduceOnTuplesWithMultipleKeyFieldSelector: Unit = {
+ /*
+ * check correctness of groupReduce on tuples with multiple key field selector
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets
+ .get5TupleDataSet(env)
+ val reduceDs = ds.groupBy(4, 0).reduceGroup {
+ in =>
+ val (i, l, l2) = in
+ .map( t => (t._1, t._2, t._5))
+ .reduce((l, r) => (l._1, l._2 + r._2, l._3))
+ (i, l, 0, "P-)", l2)
+ }
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1,0,P-),1\n" + "2,3,0,P-),1\n" + "2,2,0,P-),2\n" + "3,9,0,P-),2\n" + "3,6,0," +
+ "P-),3\n" + "4,17,0,P-),1\n" + "4,17,0,P-),2\n" + "5,11,0,P-),1\n" + "5,29,0,P-)," +
+ "2\n" + "5,25,0,P-),3\n"
+ }
+
+ @Test
+ def testCorrectnessOfGroupReduceOnTuplesWithKeyFieldSelectorAndGroupSorting: Unit = {
+ /*
+ * check correctness of groupReduce on tuples with key field selector and group sorting
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.groupBy(1).sortGroup(2, Order.ASCENDING).reduceGroup {
+ in =>
+ in.reduce((l, r) => (l._1 + r._1, l._2, l._3 + "-" + r._3))
+ }
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1,Hi\n" +
+ "5,2,Hello-Hello world\n" +
+ "15,3,Hello world, how are you?-I am fine.-Luke Skywalker\n" +
+ "34,4,Comment#1-Comment#2-Comment#3-Comment#4\n" +
+ "65,5,Comment#5-Comment#6-Comment#7-Comment#8-Comment#9\n" +
+ "111,6,Comment#10-Comment#11-Comment#12-Comment#13-Comment#14-Comment#15\n"
+ }
+
+ @Test
+ def testCorrectnessOfGroupReduceOnTuplesWithKeyExtractor: Unit = {
+ /*
+ * check correctness of groupReduce on tuples with key extractor
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.groupBy(_._2).reduceGroup {
+ in =>
+ in.map(t => (t._1, t._2)).reduce((l, r) => (l._1 + r._1, l._2))
+ }
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1\n" + "5,2\n" + "15,3\n" + "34,4\n" + "65,5\n" + "111,6\n"
+ }
+
+ @Test
+ def testCorrectnessOfGroupReduceOnCustomTypeWithTypeExtractor: Unit = {
+ /*
+ * check correctness of groupReduce on custom type with type extractor
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getCustomTypeDataSet(env)
+ val reduceDs = ds.groupBy(_.myInt).reduceGroup {
+ in =>
+ val iter = in.toIterator
+ val o = new CustomType
+ val c = iter.next()
+
+ o.myString = "Hello!"
+ o.myInt = c.myInt
+ o.myLong = c.myLong
+
+ while (iter.hasNext) {
+ val next = iter.next()
+ o.myLong += next.myLong
}
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "11,1,Hi!\n" + "21,1,Hi again!\n" + "12,2,Hi!\n" + "22,2,Hi again!\n" + "13,2," +
- "Hi!\n" + "23,2,Hi again!\n"
-
- case 10 =>
- /*
- * check correctness of groupReduce on custom type with key extractor and combine
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.getCustomTypeDataSet(env)
-
- @RichGroupReduceFunction.Combinable
- class CustomTypeGroupReduceWithCombine
- extends RichGroupReduceFunction[CustomType, CustomType] {
- override def combine(values: Iterable[CustomType], out: Collector[CustomType]): Unit = {
- val o = new CustomType()
- for (c <- values.asScala) {
- o.myInt = c.myInt
- o.myLong += c.myLong
- o.myString = "test" + c.myInt
- }
- out.collect(o)
- }
+ o
+ }
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,0,Hello!\n" + "2,3,Hello!\n" + "3,12,Hello!\n" + "4,30,Hello!\n" + "5,60," +
+ "Hello!\n" + "6,105,Hello!\n"
+ }
- override def reduce(values: Iterable[CustomType], out: Collector[CustomType]): Unit = {
- val o = new CustomType(0, 0, "")
- for (c <- values.asScala) {
- o.myInt = c.myInt
- o.myLong += c.myLong
- o.myString = c.myString
- }
- out.collect(o)
- }
+ @Test
+ def testCorrectnessOfAllGroupReduceForTuples: Unit = {
+ /*
+ * check correctness of all-groupreduce for tuples
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.reduceGroup {
+ in =>
+ var i = 0
+ var l = 0L
+ for (t <- in) {
+ i += t._1
+ l += t._2
}
- val reduceDs = ds.groupBy(_.myInt).reduceGroup(new CustomTypeGroupReduceWithCombine)
+ (i, l, "Hello World")
+ }
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "231,91,Hello World\n"
+ }
- reduceDs.writeAsText(resultPath)
- env.execute()
- if (onCollection) {
- null
+ @Test
+ def testCorrectnessOfAllGroupReduceForCustomTypes: Unit = {
+ /*
+ * check correctness of all-groupreduce for custom types
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getCustomTypeDataSet(env)
+ val reduceDs = ds.reduceGroup {
+ in =>
+ val o = new CustomType(0, 0, "Hello!")
+ for (t <- in) {
+ o.myInt += t.myInt
+ o.myLong += t.myLong
}
- else {
- "1,0,test1\n" + "2,3,test2\n" + "3,12,test3\n" + "4,30,test4\n" + "5,60," +
- "test5\n" + "6,105,test6\n"
+ o
+ }
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "91,210,Hello!"
+ }
+
+ @Test
+ def testCorrectnessOfGroupReduceWithBroadcastSet: Unit = {
+ /*
+ * check correctness of groupReduce with broadcast set
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val intDs = CollectionDataSets.getIntDataSet(env)
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.groupBy(1).reduceGroup(
+ new RichGroupReduceFunction[(Int, Long, String), (Int, Long, String)] {
+ private var f2Replace = ""
+
+ override def open(config: Configuration) {
+ val ints = this.getRuntimeContext.getBroadcastVariable[Int]("ints").asScala
+ f2Replace = ints.sum + ""
}
- case 11 =>
- /*
- * check correctness of groupReduce on tuples with combine
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- // important because it determines how often the combiner is called
- env.setDegreeOfParallelism(2)
- val ds = CollectionDataSets.get3TupleDataSet(env)
- @RichGroupReduceFunction.Combinable
- class Tuple3GroupReduceWithCombine
- extends RichGroupReduceFunction[(Int, Long, String), (Int, String)] {
- override def combine(
- values: Iterable[(Int, Long, String)],
- out: Collector[(Int, Long, String)]): Unit = {
- var i = 0
- var l = 0L
- var s = ""
- for (t <- values.asScala) {
- i += t._1
- l = t._2
- s = "test" + t._2
- }
- out.collect((i, l, s))
+ override def reduce(
+ values: Iterable[(Int, Long, String)],
+ out: Collector[(Int, Long, String)]): Unit = {
+ var i: Int = 0
+ var l: Long = 0L
+ for (t <- values.asScala) {
+ i += t._1
+ l = t._2
}
+ out.collect((i, l, f2Replace))
+ }
+ }).withBroadcastSet(intDs, "ints")
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1,55\n" + "5,2,55\n" + "15,3,55\n" + "34,4,55\n" + "65,5,55\n" + "111,6,55\n"
+ }
- override def reduce(
- values: Iterable[(Int, Long, String)],
- out: Collector[(Int, String)]): Unit = {
- var i = 0
- var s = ""
- for (t <- values.asScala) {
- i += t._1
- s = t._3
- }
- out.collect((i, s))
+ @Test
+ def testCorrectnessOfGroupReduceIfUDFReturnsInputObjectMultipleTimesWhileChangingIt: Unit = {
+ /*
+ * check correctness of groupReduce if UDF returns input objects multiple times and
+ * changes it in between
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ .map( t => MutableTuple3(t._1, t._2, t._3) )
+ val reduceDs = ds.groupBy(1).reduceGroup {
+ (in, out: Collector[MutableTuple3[Int, Long, String]]) =>
+ for (t <- in) {
+ if (t._1 < 4) {
+ t._3 = "Hi!"
+ t._1 += 10
+ out.collect(t)
+ t._1 += 10
+ t._3 = "Hi again!"
+ out.collect(t)
}
}
- val reduceDs = ds.groupBy(1).reduceGroup(new Tuple3GroupReduceWithCombine)
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- if (onCollection) {
- null
- }
- else {
- "1,test1\n" + "5,test2\n" + "15,test3\n" + "34,test4\n" + "65,test5\n" + "111," +
- "test6\n"
- }
+ }
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "11,1,Hi!\n" + "21,1,Hi again!\n" + "12,2,Hi!\n" + "22,2,Hi again!\n" + "13,2," +
+ "Hi!\n" + "23,2,Hi again!\n"
+ }
+ @Test
+ def testCorrectnessOfGroupReduceOnCustomTypeWithKeyExtractorAndCombine: Unit = {
+ /*
+ * check correctness of groupReduce on custom type with key extractor and combine
+ */
+ org.junit.Assume.assumeThat(mode, new IsNot(new IsEqual(ExecutionMode.COLLECTION)))
- // all-groupreduce with combine
-
-
- case 12 =>
- /*
- * check correctness of all-groupreduce for tuples with combine
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.get3TupleDataSet(env).map(t => t).setParallelism(4)
-
- val cfg: Configuration = new Configuration
- cfg.setString(PactCompiler.HINT_SHIP_STRATEGY, PactCompiler.HINT_SHIP_STRATEGY_REPARTITION)
-
- @RichGroupReduceFunction.Combinable
- class Tuple3AllGroupReduceWithCombine
- extends RichGroupReduceFunction[(Int, Long, String), (Int, String)] {
- override def combine(
- values: Iterable[(Int, Long, String)],
- out: Collector[(Int, Long, String)]): Unit = {
- var i = 0
- var l = 0L
- var s = ""
- for (t <- values.asScala) {
- i += t._1
- l += t._2
- s += "test"
- }
- out.collect((i, l, s))
- }
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getCustomTypeDataSet(env)
- override def reduce(
- values: Iterable[(Int, Long, String)],
- out: Collector[(Int, String)]): Unit = {
- var i = 0
- var s = ""
- for (t <- values.asScala) {
- i += t._1 + t._2.toInt
- s += t._3
- }
- out.collect((i, s))
- }
- }
- val reduceDs = ds.reduceGroup(new Tuple3AllGroupReduceWithCombine).withParameters(cfg)
+ val reduceDs = ds.groupBy(_.myInt).reduceGroup(new CustomTypeGroupReduceWithCombine)
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- if (onCollection) {
- null
- }
- else {
- "322," +
- "testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttest\n"
- }
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected =
+ "1,0,test1\n" + "2,3,test2\n" + "3,12,test3\n" + "4,30,test4\n" + "5,60," +
+ "test5\n" + "6,105,test6\n"
+
+ }
+
+ @Test
+ def testCorrectnessOfGroupReduceOnTuplesWithCombine: Unit = {
+ /*
+ * check correctness of groupReduce on tuples with combine
+ */
+ org.junit.Assume.assumeThat(mode, new IsNot(new IsEqual(ExecutionMode.COLLECTION)))
+
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ // important because it determines how often the combiner is called
+ env.setDegreeOfParallelism(2)
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+
+ val reduceDs = ds.groupBy(1).reduceGroup(new Tuple3GroupReduceWithCombine)
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected =
+ "1,test1\n" + "5,test2\n" + "15,test3\n" + "34,test4\n" + "65,test5\n" + "111," +
+ "test6\n"
+
+ }
+
+ @Test
+ def testCorrectnessOfAllGroupReduceForTuplesWithCombine: Unit = {
+ /*
+ * check correctness of all-groupreduce for tuples with combine
+ */
+ org.junit.Assume.assumeThat(mode, new IsNot(new IsEqual(ExecutionMode.COLLECTION)))
+
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.get3TupleDataSet(env).map(t => t).setParallelism(4)
+
+ val cfg: Configuration = new Configuration
+ cfg.setString(PactCompiler.HINT_SHIP_STRATEGY, PactCompiler.HINT_SHIP_STRATEGY_REPARTITION)
+
+ val reduceDs = ds.reduceGroup(new Tuple3AllGroupReduceWithCombine).withParameters(cfg)
+
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "322," +
+ "testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttesttest\n"
+
+ }
- case 13 =>
- /*
- * check correctness of groupReduce with descending group sort
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.get3TupleDataSet(env)
- val reduceDs = ds.groupBy(1).sortGroup(2, Order.DESCENDING).reduceGroup {
- in =>
- in.reduce((l, r) => (l._1 + r._1, l._2, l._3 + "-" + r._3))
+ @Test
+ def testCorrectnessOfGroupReduceWithDescendingGroupSort: Unit = {
+ /*
+ * check correctness of groupReduce with descending group sort
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.groupBy(1).sortGroup(2, Order.DESCENDING).reduceGroup {
+ in =>
+ in.reduce((l, r) => (l._1 + r._1, l._2, l._3 + "-" + r._3))
+ }
+
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1,Hi\n" + "5,2,Hello world-Hello\n" + "15,3,Luke Skywalker-I am fine.-Hello " +
+ "world, how are you?\n" + "34,4,Comment#4-Comment#3-Comment#2-Comment#1\n" + "65,5," +
+ "Comment#9-Comment#8-Comment#7-Comment#6-Comment#5\n" + "111,6," +
+ "Comment#15-Comment#14-Comment#13-Comment#12-Comment#11-Comment#10\n"
+ }
+
+ @Test
+ def testCorrectnessOfGroupReduceOnTuplesWithTupleReturningKeySelector: Unit = {
+ /*
+ * check correctness of groupReduce on tuples with tuple-returning key selector
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets
+ .get5TupleDataSet(env)
+ val reduceDs = ds.groupBy( t => (t._1, t._5)).reduceGroup {
+ in =>
+ val (i, l, l2) = in
+ .map( t => (t._1, t._2, t._5))
+ .reduce((l, r) => (l._1, l._2 + r._2, l._3))
+ (i, l, 0, "P-)", l2)
+ }
+
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1,0,P-),1\n" + "2,3,0,P-),1\n" + "2,2,0,P-),2\n" + "3,9,0,P-),2\n" + "3,6,0," +
+ "P-),3\n" + "4,17,0,P-),1\n" + "4,17,0,P-),2\n" + "5,11,0,P-),1\n" + "5,29,0,P-)," +
+ "2\n" + "5,25,0,P-),3\n"
+ }
+
+ @Test
+ def testInputOfCombinerIsSortedForCombinableGroupReduceWithGroupSorting: Unit = {
+ /*
+ * check that input of combiner is also sorted for combinable groupReduce with group
+ * sorting
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.get3TupleDataSet(env).map { t =>
+ MutableTuple3(t._1, t._2, t._3)
+ }
+
+ val reduceDs = ds.groupBy(1)
+ .sortGroup(0, Order.ASCENDING).reduceGroup(new OrderCheckingCombinableReduce)
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1,Hi\n" + "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "7,4," +
+ "Comment#1\n" + "11,5,Comment#5\n" + "16,6,Comment#10\n"
+ }
+
+ @Test
+ def testDeepNestingAndNullValueInPojo: Unit = {
+ /*
+ * Deep nesting test
+ * + null value in pojo
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets
+ .getCrazyNestedDataSet(env)
+ val reduceDs = ds.groupBy("nest_Lvl1.nest_Lvl2.nest_Lvl3.nest_Lvl4.f1nal")
+ .reduceGroup {
+ in =>
+ var c = 0
+ var n: String = null
+ for (v <- in) {
+ c += 1
+ n = v.nest_Lvl1.nest_Lvl2.nest_Lvl3.nest_Lvl4.f1nal
}
+ (n, c)
+ }
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "aa,1\nbb,2\ncc,3\n"
+ }
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "1,1,Hi\n" + "5,2,Hello world-Hello\n" + "15,3,Luke Skywalker-I am fine.-Hello " +
- "world, how are you?\n" + "34,4,Comment#4-Comment#3-Comment#2-Comment#1\n" + "65,5," +
- "Comment#9-Comment#8-Comment#7-Comment#6-Comment#5\n" + "111,6," +
- "Comment#15-Comment#14-Comment#13-Comment#12-Comment#11-Comment#10\n"
-
- case 14 =>
- /*
- * check correctness of groupReduce on tuples with tuple-returning key selector
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets
- .get5TupleDataSet(env)
- val reduceDs = ds.groupBy( t => (t._1, t._5)).reduceGroup {
- in =>
- val (i, l, l2) = in
- .map( t => (t._1, t._2, t._5))
- .reduce((l, r) => (l._1, l._2 + r._2, l._3))
- (i, l, 0, "P-)", l2)
+ @Test
+ def testPojoContainigAWritableAndTuples: Unit = {
+ /*
+ * Test Pojo containing a Writable and Tuples
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets
+ .getPojoContainingTupleAndWritable(env)
+ val reduceDs = ds.groupBy("hadoopFan", "theTuple.*").reduceGroup(new
+ GroupReduceFunction[CollectionDataSets.PojoContainingTupleAndWritable, Integer] {
+ def reduce(
+ values: Iterable[CollectionDataSets.PojoContainingTupleAndWritable],
+ out: Collector[Integer]) {
+ var c: Int = 0
+ for (v <- values.asScala) {
+ c += 1
}
+ out.collect(c)
+ }
+ })
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1\n5\n"
+ }
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "1,1,0,P-),1\n" + "2,3,0,P-),1\n" + "2,2,0,P-),2\n" + "3,9,0,P-),2\n" + "3,6,0," +
- "P-),3\n" + "4,17,0,P-),1\n" + "4,17,0,P-),2\n" + "5,11,0,P-),1\n" + "5,29,0,P-)," +
- "2\n" + "5,25,0,P-),3\n"
-
- case 15 =>
- /*
- * check that input of combiner is also sorted for combinable groupReduce with group
- * sorting
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.get3TupleDataSet(env).map { t =>
- MutableTuple3(t._1, t._2, t._3)
+ @Test
+ def testTupleContainingPojosAndRegularFields: Unit ={
+ /*
+ * Test Tuple containing pojos and regular fields
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val ds = CollectionDataSets.getTupleContainingPojos(env)
+ val reduceDs = ds.groupBy("_1", "_2.*").reduceGroup(
+ new GroupReduceFunction[(Int, CrazyNested, POJO), Int] {
+ def reduce(values: Iterable[(Int, CrazyNested, POJO)], out: Collector[Int]) {
+ var c: Int = 0
+ for (v <- values.asScala) {
+ c += 1
+ }
+ out.collect(c)
}
+ })
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "3\n1\n"
+ }
- @RichGroupReduceFunction.Combinable
- class OrderCheckingCombinableReduce
- extends RichGroupReduceFunction[MutableTuple3[Int, Long, String],
- MutableTuple3[Int, Long, String]] {
- def reduce(
- values: Iterable[MutableTuple3[Int, Long, String]],
- out: Collector[MutableTuple3[Int, Long, String]]) {
- val it = values.iterator()
- var t = it.next()
- val i = t._1
- out.collect(t)
+ @Test
+ def testStringBasedDefinitionOnGroupSort: Unit = {
+ /*
+ * Test string-based definition on group sort, based on test:
+ * check correctness of groupReduce with descending group sort
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.get3TupleDataSet(env)
+ val reduceDs = ds.groupBy(1)
+ .sortGroup("_3", Order.DESCENDING)
+ .reduceGroup {
+ in =>
+ in.reduce((l, r) => (l._1 + r._1, l._2, l._3 + "-" + r._3))
+ }
+ reduceDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1,Hi\n" + "5,2,Hello world-Hello\n" + "15,3,Luke Skywalker-I am fine.-Hello " +
+ "world, how are you?\n" + "34,4,Comment#4-Comment#3-Comment#2-Comment#1\n" + "65,5," +
+ "Comment#9-Comment#8-Comment#7-Comment#6-Comment#5\n" + "111,6," +
+ "Comment#15-Comment#14-Comment#13-Comment#12-Comment#11-Comment#10\n"
+ }
- while (it.hasNext) {
- t = it.next()
- if (i > t._1 || (t._3 == "INVALID-ORDER!")) {
- t._3 = "INVALID-ORDER!"
- out.collect(t)
- }
- }
- }
+ @Test
+ def testIntBasedDefinitionOnGroupSortForFullNestedTuple: Unit = {
+ /*
+ * Test int-based definition on group sort, for (full) nested Tuple
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.getGroupSortedNestedTupleDataSet(env)
+ val reduceDs = ds.groupBy("_2").sortGroup(0, Order.DESCENDING)
+ .reduceGroup(new NestedTupleReducer)
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "a--(2,1)-(1,3)-(1,2)-\n" + "b--(2,2)-\n" + "c--(4,9)-(3,6)-(3,3)-\n"
+ }
- override def combine(
- values: Iterable[MutableTuple3[Int, Long, String]],
- out: Collector[MutableTuple3[Int, Long, String]]) {
- val it = values.iterator()
- var t = it.next
- val i: Int = t._1
- out.collect(t)
- while (it.hasNext) {
- t = it.next
- if (i > t._1) {
- t._3 = "INVALID-ORDER!"
- out.collect(t)
- }
- }
- }
- }
-
- val reduceDs = ds.groupBy(1)
- .sortGroup(0, Order.ASCENDING).reduceGroup(new OrderCheckingCombinableReduce)
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "1,1,Hi\n" + "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "7,4," +
- "Comment#1\n" + "11,5,Comment#5\n" + "16,6,Comment#10\n"
-
- case 16 =>
- /*
- * Deep nesting test
- * + null value in pojo
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets
- .getCrazyNestedDataSet(env)
- val reduceDs = ds.groupBy("nest_Lvl1.nest_Lvl2.nest_Lvl3.nest_Lvl4.f1nal")
- .reduceGroup {
- in =>
- var c = 0
- var n: String = null
- for (v <- in) {
- c += 1
- n = v.nest_Lvl1.nest_Lvl2.nest_Lvl3.nest_Lvl4.f1nal
- }
- (n, c)
- }
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "aa,1\nbb,2\ncc,3\n"
-
-
- case 17 =>
- // We don't have that test but keep numbering compatible to Java GroupReduceITCase
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.fromElements("Hello world").writeAsText(resultPath)
- env.execute()
- "Hello world"
-
- case 18 =>
- /*
- * Test Pojo containing a Writable and Tuples
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets
- .getPojoContainingTupleAndWritable(env)
- val reduceDs = ds.groupBy("hadoopFan", "theTuple.*").reduceGroup(new
- GroupReduceFunction[CollectionDataSets.PojoContainingTupleAndWritable, Integer] {
+ @Test
+ def testIntBasedDefinitionOnGroupSortForPartialNestedTuple: Unit = {
+ /*
+ * Test int-based definition on group sort, for (partial) nested Tuple ASC
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.getGroupSortedNestedTupleDataSet(env)
+ val reduceDs = ds.groupBy("_2")
+ .sortGroup("_1._1", Order.ASCENDING)
+ .sortGroup("_1._2", Order.ASCENDING)
+ .reduceGroup(new NestedTupleReducer)
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "a--(1,2)-(1,3)-(2,1)-\n" + "b--(2,2)-\n" + "c--(3,3)-(3,6)-(4,9)-\n"
+ }
+
+ @Test
+ def testStringBasedDefinitionOnGroupSortForPartialNestedTuple: Unit = {
+ /*
+ * Test string-based definition on group sort, for (partial) nested Tuple DESC
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.getGroupSortedNestedTupleDataSet(env)
+ val reduceDs = ds.groupBy("_2")
+ .sortGroup("_1._1", Order.DESCENDING)
+ .sortGroup("_1._2", Order.ASCENDING)
+ .reduceGroup(new NestedTupleReducer)
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "a--(2,1)-(1,2)-(1,3)-\n" + "b--(2,2)-\n" + "c--(4,9)-(3,3)-(3,6)-\n"
+ }
+
+ @Test
+ def testStringBasedDefinitionOnGroupSortForTwoGroupingKeys: Unit = {
+ /*
+ * Test string-based definition on group sort, for two grouping keys
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.getGroupSortedNestedTupleDataSet(env)
+ val reduceDs = ds.groupBy("_2")
+ .sortGroup("_1._1", Order.DESCENDING)
+ .sortGroup("_1._2", Order.DESCENDING)
+ .reduceGroup(new NestedTupleReducer)
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "a--(2,1)-(1,3)-(1,2)-\n" + "b--(2,2)-\n" + "c--(4,9)-(3,6)-(3,3)-\n"
+ }
+
+ @Test
+ def testStringBasedDefinitionOnGroupSortForTwoGroupingKeysWithPojos: Unit = {
+ /*
+ * Test string-based definition on group sort, for two grouping keys with Pojos
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.getGroupSortedPojoContainingTupleAndWritable(env)
+ val reduceDs = ds.groupBy("hadoopFan")
+ .sortGroup("theTuple._1", Order.DESCENDING)
+ .sortGroup("theTuple._2", Order.DESCENDING)
+ .reduceGroup(
+ new GroupReduceFunction[CollectionDataSets.PojoContainingTupleAndWritable, String] {
def reduce(
- values: Iterable[CollectionDataSets.PojoContainingTupleAndWritable],
- out: Collector[Integer]) {
- var c: Int = 0
- for (v <- values.asScala) {
- c += 1
+ values: Iterable[CollectionDataSets.PojoContainingTupleAndWritable],
+ out: Collector[String]) {
+ var once: Boolean = false
+ val concat: StringBuilder = new StringBuilder
+ for (value <- values.asScala) {
+ if (!once) {
+ concat.append(value.hadoopFan.get)
+ concat.append("---")
+ once = true
+ }
+ concat.append(value.theTuple)
+ concat.append("-")
}
- out.collect(c)
+ out.collect(concat.toString())
}
})
- reduceDs.writeAsText(resultPath)
- env.execute()
- "1\n5\n"
-
- case 19 =>
- /*
- * Test Tuple containing pojos and regular fields
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- val ds = CollectionDataSets.getTupleContainingPojos(env)
- val reduceDs = ds.groupBy("_1", "_2.*").reduceGroup(
- new GroupReduceFunction[(Int, CrazyNested, POJO), Int] {
- def reduce(values: Iterable[(Int, CrazyNested, POJO)], out: Collector[Int]) {
- var c: Int = 0
- for (v <- values.asScala) {
- c += 1
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1---(10,100)-\n" + "2---(30,600)-(30,400)-(30,200)-(20,201)-(20,200)-\n"
+ }
+
+ @Test
+ def testGroupingWithPojoContainingMultiplePojos: Unit = {
+ /*
+ * Test grouping with pojo containing multiple pojos (was a bug)
+ */
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ env.setDegreeOfParallelism(1)
+ val ds = CollectionDataSets.getPojoWithMultiplePojos(env)
+ val reduceDs = ds.groupBy("p2.a2")
+ .reduceGroup(
+ new GroupReduceFunction[CollectionDataSets.PojoWithMultiplePojos, String] {
+ def reduce(
+ values: Iterable[CollectionDataSets.PojoWithMultiplePojos],
+ out: Collector[String]) {
+ val concat: StringBuilder = new StringBuilder
+ for (value <- values.asScala) {
+ concat.append(value.p2.a2)
}
- out.collect(c)
+ out.collect(concat.toString())
}
})
- reduceDs.writeAsText(resultPath)
- env.execute()
- "3\n1\n"
-
- case 20 =>
- /*
- * Test string-based definition on group sort, based on test:
- * check correctness of groupReduce with descending group sort
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.get3TupleDataSet(env)
- val reduceDs = ds.groupBy(1)
- .sortGroup("_3", Order.DESCENDING)
- .reduceGroup {
- in =>
- in.reduce((l, r) => (l._1 + r._1, l._2, l._3 + "-" + r._3))
- }
- reduceDs.writeAsCsv(resultPath)
- env.execute()
- "1,1,Hi\n" + "5,2,Hello world-Hello\n" + "15,3,Luke Skywalker-I am fine.-Hello " +
- "world, how are you?\n" + "34,4,Comment#4-Comment#3-Comment#2-Comment#1\n" + "65,5," +
- "Comment#9-Comment#8-Comment#7-Comment#6-Comment#5\n" + "111,6," +
- "Comment#15-Comment#14-Comment#13-Comment#12-Comment#11-Comment#10\n"
-
- case 21 =>
- /*
- * Test int-based definition on group sort, for (full) nested Tuple
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.getGroupSortedNestedTupleDataSet(env)
- val reduceDs = ds.groupBy("_2").sortGroup(0, Order.DESCENDING)
- .reduceGroup(new NestedTupleReducer)
- reduceDs.writeAsText(resultPath)
- env.execute()
- "a--(2,1)-(1,3)-(1,2)-\n" + "b--(2,2)-\n" + "c--(4,9)-(3,6)-(3,3)-\n"
-
-
- case 22 =>
- /*
- * Test int-based definition on group sort, for (partial) nested Tuple ASC
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.getGroupSortedNestedTupleDataSet(env)
- val reduceDs = ds.groupBy("_2")
- .sortGroup("_1._1", Order.ASCENDING)
- .sortGroup("_1._2", Order.ASCENDING)
- .reduceGroup(new NestedTupleReducer)
- reduceDs.writeAsText(resultPath)
- env.execute()
- "a--(1,2)-(1,3)-(2,1)-\n" + "b--(2,2)-\n" + "c--(3,3)-(3,6)-(4,9)-\n"
-
- case 23 =>
- /*
- * Test string-based definition on group sort, for (partial) nested Tuple DESC
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.getGroupSortedNestedTupleDataSet(env)
- val reduceDs = ds.groupBy("_2")
- .sortGroup("_1._1", Order.DESCENDING)
- .sortGroup("_1._2", Order.ASCENDING)
- .reduceGroup(new NestedTupleReducer)
- reduceDs.writeAsText(resultPath)
- env.execute()
- "a--(2,1)-(1,2)-(1,3)-\n" + "b--(2,2)-\n" + "c--(4,9)-(3,3)-(3,6)-\n"
-
- case 24 =>
- /*
- * Test string-based definition on group sort, for two grouping keys
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.getGroupSortedNestedTupleDataSet(env)
- val reduceDs = ds.groupBy("_2")
- .sortGroup("_1._1", Order.DESCENDING)
- .sortGroup("_1._2", Order.DESCENDING)
- .reduceGroup(new NestedTupleReducer)
- reduceDs.writeAsText(resultPath)
- env.execute()
- "a--(2,1)-(1,3)-(1,2)-\n" + "b--(2,2)-\n" + "c--(4,9)-(3,6)-(3,3)-\n"
-
- case 25 =>
- /*
- * Test string-based definition on group sort, for two grouping keys with Pojos
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.getGroupSortedPojoContainingTupleAndWritable(env)
- val reduceDs = ds.groupBy("hadoopFan")
- .sortGroup("theTuple._1", Order.DESCENDING)
- .sortGroup("theTuple._2", Order.DESCENDING)
- .reduceGroup(
- new GroupReduceFunction[CollectionDataSets.PojoContainingTupleAndWritable, String] {
- def reduce(
- values: Iterable[CollectionDataSets.PojoContainingTupleAndWritable],
- out: Collector[String]) {
- var once: Boolean = false
- val concat: StringBuilder = new StringBuilder
- for (value <- values.asScala) {
- if (!once) {
- concat.append(value.hadoopFan.get)
- concat.append("---")
- once = true
- }
- concat.append(value.theTuple)
- concat.append("-")
- }
- out.collect(concat.toString())
- }
- })
- reduceDs.writeAsText(resultPath)
- env.execute()
- "1---(10,100)-\n" + "2---(30,600)-(30,400)-(30,200)-(20,201)-(20,200)-\n"
-
- case 26 =>
- /*
- * Test grouping with pojo containing multiple pojos (was a bug)
- */
- val env = ExecutionEnvironment.getExecutionEnvironment
- env.setDegreeOfParallelism(1)
- val ds = CollectionDataSets.getPojoWithMultiplePojos(env)
- val reduceDs = ds.groupBy("p2.a2")
- .reduceGroup(
- new GroupReduceFunction[CollectionDataSets.PojoWithMultiplePojos, String] {
- def reduce(
- values: Iterable[CollectionDataSets.PojoWithMultiplePojos],
- out: Collector[String]) {
- val concat: StringBuilder = new StringBuilder
- for (value <- values.asScala) {
- concat.append(value.p2.a2)
- }
- out.collect(concat.toString())
- }
- })
- reduceDs.writeAsText(resultPath)
- env.execute()
- "b\nccc\nee\n"
-
- case _ =>
- throw new IllegalArgumentException("Invalid program id")
- }
+ reduceDs.writeAsText(resultPath, WriteMode.OVERWRITE)
+ env.execute()
+ expected = "b\nccc\nee\n"
}
+
}
+@RichGroupReduceFunction.Combinable
+class OrderCheckingCombinableReduce
+ extends RichGroupReduceFunction[MutableTuple3[Int, Long, String],
+ MutableTuple3[Int, Long, String]] {
+ def reduce(
+ values: Iterable[MutableTuple3[Int, Long, String]],
+ out: Collector[MutableTuple3[Int, Long, String]]) {
+ val it = values.iterator()
+ var t = it.next()
+ val i = t._1
+ out.collect(t)
+
+ while (it.hasNext) {
+ t = it.next()
+ if (i > t._1 || (t._3 == "INVALID-ORDER!")) {
+ t._3 = "INVALID-ORDER!"
+ out.collect(t)
+ }
+ }
+ }
-@RunWith(classOf[Parameterized])
-class GroupReduceITCase(config: Configuration) extends JavaProgramTestBase(config) {
+ override def combine(
+ values: Iterable[MutableTuple3[Int, Long, String]],
+ out: Collector[MutableTuple3[Int, Long, String]]) {
+ val it = values.iterator()
+ var t = it.next
+ val i: Int = t._1
+ out.collect(t)
+ while (it.hasNext) {
+ t = it.next
+ if (i > t._1) {
+ t._3 = "INVALID-ORDER!"
+ out.collect(t)
+ }
+ }
+ }
+}
- private val curProgId: Int = config.getInteger("ProgramId", -1)
- private var resultPath: String = null
- private var expectedResult: String = null
+@RichGroupReduceFunction.Combinable
+class CustomTypeGroupReduceWithCombine
+ extends RichGroupReduceFunction[CustomType, CustomType] {
+ override def combine(values: Iterable[CustomType], out: Collector[CustomType]): Unit = {
+ val o = new CustomType()
+ for (c <- values.asScala) {
+ o.myInt = c.myInt
+ o.myLong += c.myLong
+ o.myString = "test" + c.myInt
+ }
+ out.collect(o)
+ }
- protected override def preSubmit(): Unit = {
- resultPath = getTempDirPath("result")
+ override def reduce(values: Iterable[CustomType], out: Collector[CustomType]): Unit = {
+ val o = new CustomType(0, 0, "")
+ for (c <- values.asScala) {
+ o.myInt = c.myInt
+ o.myLong += c.myLong
+ o.myString = c.myString
+ }
+ out.collect(o)
}
+}
- protected def testProgram(): Unit = {
- expectedResult = GroupReduceProgs.runProgram(curProgId, resultPath, isCollectionExecution)
+@RichGroupReduceFunction.Combinable
+class Tuple3GroupReduceWithCombine
+ extends RichGroupReduceFunction[(Int, Long, String), (Int, String)] {
+ override def combine(
+ values: Iterable[(Int, Long, String)],
+ out: Collector[(Int, Long, String)]): Unit = {
+ var i = 0
+ var l = 0L
+ var s = ""
+ for (t <- values.asScala) {
+ i += t._1
+ l = t._2
+ s = "test" + t._2
+ }
+ out.collect((i, l, s))
}
- protected override def postSubmit(): Unit = {
- if (expectedResult != null) compareResultsByLinesInMemory(expectedResult, resultPath)
+ override def reduce(
+ values: Iterable[(Int, Long, String)],
+ out: Collector[(Int, String)]): Unit = {
+ var i = 0
+ var s = ""
+ for (t <- values.asScala) {
+ i += t._1
+ s = t._3
+ }
+ out.collect((i, s))
}
}
-object GroupReduceITCase {
- @Parameters
- def getConfigurations: java.util.Collection[Array[AnyRef]] = {
- val configs = mutable.MutableList[Array[AnyRef]]()
- for (i <- 1 to GroupReduceProgs.NUM_PROGRAMS) {
- val config = new Configuration()
- config.setInteger("ProgramId", i)
- configs += Array(config)
+@RichGroupReduceFunction.Combinable
+class Tuple3AllGroupReduceWithCombine
+ extends RichGroupReduceFunction[(Int, Long, String), (Int, String)] {
+ override def combine(
+ values: Iterable[(Int, Long, String)],
+ out: Collector[(Int, Long, String)]): Unit = {
+ var i = 0
+ var l = 0L
+ var s = ""
+ for (t <- values.asScala) {
+ i += t._1
+ l += t._2
+ s += "test"
}
+ out.collect((i, l, s))
+ }
- configs.asJavaCollection
+ override def reduce(
+ values: Iterable[(Int, Long, String)],
+ out: Collector[(Int, String)]): Unit = {
+ var i = 0
+ var s = ""
+ for (t <- values.asScala) {
+ i += t._1 + t._2.toInt
+ s += t._3
+ }
+ out.collect((i, s))
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a0f77857/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala
index c2efd5e..8ae8674 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/operators/JoinITCase.scala
@@ -19,358 +19,369 @@ package org.apache.flink.api.scala.operators
import org.apache.flink.api.common.functions.RichJoinFunction
import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.api.scala.util.CollectionDataSets.MutableTuple3
import org.apache.flink.configuration.Configuration
-import org.apache.flink.test.util.JavaProgramTestBase
-import org.apache.flink.util.Collector
+import org.apache.flink.core.fs.FileSystem.WriteMode
+import org.apache.flink.test.util.MultipleProgramsTestBase.ExecutionMode
+import org.apache.flink.test.util.{MultipleProgramsTestBase}
+import org.junit.{Test, After, Before, Rule}
+import org.junit.rules.TemporaryFolder
import org.junit.runner.RunWith
import org.junit.runners.Parameterized
-import org.junit.runners.Parameterized.Parameters
import scala.collection.JavaConverters._
-import scala.collection.mutable
import org.apache.flink.api.scala._
+@RunWith(classOf[Parameterized])
+class JoinITCase(mode: ExecutionMode) extends MultipleProgramsTestBase(mode) {
+ private var resultPath: String = null
+ private var expected: String = null
+ private val _tempFolder = new TemporaryFolder()
-object JoinProgs {
- var NUM_PROGRAMS: Int = 20
-
- def runProgram(progId: Int, resultPath: String): String = {
- progId match {
- case 1 =>
- /*
- * UDF Join on tuples with key field positions
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
- val ds2 = CollectionDataSets.get5TupleDataSet(env)
- val joinDs = ds1.join(ds2).where(1).equalTo(1) { (l, r) => (l._3, r._4) }
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n"
-
- case 2 =>
- /*
- * UDF Join on tuples with multiple key field positions
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.get3TupleDataSet(env)
- val ds2 = CollectionDataSets.get5TupleDataSet(env)
- val joinDs = ds1.join(ds2).where(0, 1).equalTo(0, 4) { (l, r) => (l._3, r._4) }
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
- "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n"
-
- case 3 =>
- /*
- * Default Join on tuples
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
- val ds2 = CollectionDataSets.get5TupleDataSet(env)
- val joinDs = ds1.join(ds2).where(0).equalTo(2)
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "(1,1,Hi),(2,2,1,Hallo Welt,2)\n" + "(2,2,Hello),(2,3,2,Hallo Welt wie," +
- "1)\n" + "(3,2,Hello world),(3,4,3,Hallo Welt wie gehts?,2)\n"
-
- case 4 =>
- /*
- * Join with Huge
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
- val ds2 = CollectionDataSets.get5TupleDataSet(env)
- val joinDs = ds1.joinWithHuge(ds2).where(1).equalTo(1) { (l, r) => (l._3, r._4) }
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n"
+ @Rule
+ def tempFolder = _tempFolder
- case 5 =>
- /*
- * Join with Tiny
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
- val ds2 = CollectionDataSets.get5TupleDataSet(env)
- val joinDs = ds1.joinWithTiny(ds2).where(1).equalTo(1) { (l, r) => (l._3, r._4) }
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n"
+ @Before
+ def before(): Unit = {
+ resultPath = tempFolder.newFile().toURI.toString
+ }
- case 6 =>
- /*
- * Join that returns the left input object
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
- val ds2 = CollectionDataSets.get5TupleDataSet(env)
- val joinDs = ds1.join(ds2).where(1).equalTo(1) { (l, r) => l }
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n"
+ @After
+ def after: Unit = {
+ compareResultsByLinesInMemory(expected, resultPath)
+ }
- case 7 =>
- /*
- * Join that returns the right input object
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
- val ds2 = CollectionDataSets.get5TupleDataSet(env)
- val joinDs = ds1.join(ds2).where(1).equalTo(1) { (l, r) => r }
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "1,1,0,Hallo,1\n" + "2,2,1,Hallo Welt,2\n" + "2,2,1,Hallo Welt,2\n"
+ @Test
+ def testUDFJoinOnTuplesWithKeyFieldPositions: Unit = {
+ /*
+ * UDF Join on tuples with key field positions
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
+ val ds2 = CollectionDataSets.get5TupleDataSet(env)
+ val joinDs = ds1.join(ds2).where(1).equalTo(1) { (l, r) => (l._3, r._4) }
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n"
+ }
- case 8 =>
- /*
- * Join with broadcast set
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val intDs = CollectionDataSets.getIntDataSet(env)
- val ds1 = CollectionDataSets.get3TupleDataSet(env)
- val ds2 = CollectionDataSets.getSmall5TupleDataSet(env)
- val joinDs = ds1.join(ds2).where(1).equalTo(4).apply(
- new RichJoinFunction[
- (Int, Long, String),
- (Int, Long, Int, String, Long),
- (String, String, Int)] {
- private var broadcast = 41
+ @Test
+ def testUDFJoinOnTuplesWithMultipleKeyFieldPositions: Unit = {
+ /*
+ * UDF Join on tuples with multiple key field positions
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.get3TupleDataSet(env)
+ val ds2 = CollectionDataSets.get5TupleDataSet(env)
+ val joinDs = ds1.join(ds2).where(0, 1).equalTo(0, 4) { (l, r) => (l._3, r._4) }
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
+ "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n"
+ }
- override def open(config: Configuration) {
- val ints = this.getRuntimeContext.getBroadcastVariable[Int]("ints").asScala
- broadcast = ints.sum
- }
+ @Test
+ def testDefaultJoinOnTuples: Unit = {
+ /*
+ * Default Join on tuples
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
+ val ds2 = CollectionDataSets.get5TupleDataSet(env)
+ val joinDs = ds1.join(ds2).where(0).equalTo(2)
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "(1,1,Hi),(2,2,1,Hallo Welt,2)\n" + "(2,2,Hello),(2,3,2,Hallo Welt wie," +
+ "1)\n" + "(3,2,Hello world),(3,4,3,Hallo Welt wie gehts?,2)\n"
+ }
- override def join(
- first: (Int, Long, String),
- second: (Int, Long, Int, String, Long)): (String, String, Int) = {
- (first._3, second. _4, broadcast)
- }
- }
- ).withBroadcastSet(intDs, "ints")
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "Hi,Hallo,55\n" + "Hi,Hallo Welt wie,55\n" + "Hello,Hallo Welt," +
- "55\n" + "Hello world,Hallo Welt,55\n"
+ @Test
+ def testJoinWithHuge: Unit = {
+ /*
+ * Join with Huge
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
+ val ds2 = CollectionDataSets.get5TupleDataSet(env)
+ val joinDs = ds1.joinWithHuge(ds2).where(1).equalTo(1) { (l, r) => (l._3, r._4) }
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n"
+ }
- case 9 =>
- /*
- * Join on a tuple input with key field selector and a custom type input with key extractor
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmallCustomTypeDataSet(env)
- val ds2 = CollectionDataSets.get3TupleDataSet(env)
- val joinDs = ds1.join(ds2).where( _.myInt ).equalTo(0) { (l, r) => (l.myString, r._3) }
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "Hi,Hi\n" + "Hello,Hello\n" + "Hello world,Hello\n"
+ @Test
+ def testJoinWithTiny: Unit = {
+ /*
+ * Join with Tiny
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
+ val ds2 = CollectionDataSets.get5TupleDataSet(env)
+ val joinDs = ds1.joinWithTiny(ds2).where(1).equalTo(1) { (l, r) => (l._3, r._4) }
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n"
+ }
- case 10 => // 12 in Java ITCase
- /*
- * Join on a tuple input with key field selector and a custom type input with key extractor
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
- val ds2 = CollectionDataSets.getCustomTypeDataSet(env)
- val joinDs = ds1.join(ds2).where(1).equalTo(_.myLong) apply { (l, r) => (l._3, r.myString) }
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "Hi,Hello\n" + "Hello,Hello world\n" + "Hello world,Hello world\n"
+ @Test
+ def testJoinThatReturnsTheLeftInputObject: Unit = {
+ /*
+ * Join that returns the left input object
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
+ val ds2 = CollectionDataSets.get5TupleDataSet(env)
+ val joinDs = ds1.join(ds2).where(1).equalTo(1) { (l, r) => l }
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n"
+ }
- case 11 => // 13 in Java ITCase
- /*
- * (Default) Join on two custom type inputs with key extractors
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getCustomTypeDataSet(env)
- val ds2 = CollectionDataSets.getSmallCustomTypeDataSet(env)
- val joinDs = ds1.join(ds2).where(_.myInt).equalTo(_.myInt)
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "1,0,Hi,1,0,Hi\n" + "2,1,Hello,2,1,Hello\n" + "2,1,Hello,2,2,Hello world\n" + "2," +
- "2,Hello world,2,1,Hello\n" + "2,2,Hello world,2,2,Hello world\n"
+ @Test
+ def testJoinThatReturnsTheRightInputObject: Unit = {
+ /*
+ * Join that returns the right input object
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
+ val ds2 = CollectionDataSets.get5TupleDataSet(env)
+ val joinDs = ds1.join(ds2).where(1).equalTo(1) { (l, r) => r }
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,1,0,Hallo,1\n" + "2,2,1,Hallo Welt,2\n" + "2,2,1,Hallo Welt,2\n"
+ }
- case 12 => // 14 in Java ITCase
- /*
- * UDF Join on tuples with tuple-returning key selectors
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.get3TupleDataSet(env)
- val ds2 = CollectionDataSets.get5TupleDataSet(env)
- val joinDs = ds1.join(ds2).where( t => (t._1, t._2)).equalTo( t => (t._1, t._5)) apply {
- (l, r) => (l._3, r._4)
+ @Test
+ def testJoinWithBroadcastSet: Unit ={
+ /*
+ * Join with broadcast set
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val intDs = CollectionDataSets.getIntDataSet(env)
+ val ds1 = CollectionDataSets.get3TupleDataSet(env)
+ val ds2 = CollectionDataSets.getSmall5TupleDataSet(env)
+ val joinDs = ds1.join(ds2).where(1).equalTo(4).apply(
+ new RichJoinFunction[
+ (Int, Long, String),
+ (Int, Long, Int, String, Long),
+ (String, String, Int)] {
+ private var broadcast = 41
+
+ override def open(config: Configuration) {
+ val ints = this.getRuntimeContext.getBroadcastVariable[Int]("ints").asScala
+ broadcast = ints.sum
}
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
- "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n"
-
- /**
- * Joins with POJOs
- */
- case 13 => // 15 in Java ITCase
- /*
- * Join nested pojo against tuple (selected using a string)
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
- val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
- val joinDs = ds1.join(ds2).where("nestedPojo.longNumber").equalTo("_7")
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
- "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
- "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
-
- case 14 => // 16 in Java ITCase
- /*
- * Join nested pojo against tuple (selected as an integer)
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
- val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
- val joinDs = ds1.join(ds2).where("nestedPojo.longNumber").equalTo(6) // <-- difference
- joinDs.writeAsCsv(resultPath)
- env.execute()
- "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
- "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
- "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
-
- case 15 => // 17 in Java ITCase
- /*
- * selecting multiple fields using expression language
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
- val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
- val joinDs = ds1.join(ds2)
- .where("nestedPojo.longNumber", "number", "str")
- .equalTo("_7", "_1", "_2")
- joinDs.writeAsCsv(resultPath)
- env.setDegreeOfParallelism(1)
- env.execute()
- "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
- "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
- "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
- case 16 => // 18 in Java ITCase
- /*
- * nested into tuple
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
- val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
- val joinDs = ds1.join(ds2).where("nestedPojo.longNumber", "number",
- "nestedTupleWithCustom._1").equalTo("_7", "_1", "_3")
- joinDs.writeAsCsv(resultPath)
- env.setDegreeOfParallelism(1)
- env.execute()
- "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
- "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
- "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
+ override def join(
+ first: (Int, Long, String),
+ second: (Int, Long, Int, String, Long)): (String, String, Int) = {
+ (first._3, second. _4, broadcast)
+ }
+ }
+ ).withBroadcastSet(intDs, "ints")
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "Hi,Hallo,55\n" + "Hi,Hallo Welt wie,55\n" + "Hello,Hallo Welt," +
+ "55\n" + "Hello world,Hallo Welt,55\n"
+ }
- case 17 => // 19 in Java ITCase
- /*
- * nested into tuple into pojo
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
- val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
- val joinDs = ds1.join(ds2)
- .where("nestedTupleWithCustom._1",
- "nestedTupleWithCustom._2.myInt",
- "nestedTupleWithCustom._2.myLong")
- .equalTo("_3", "_4", "_5")
- joinDs.writeAsCsv(resultPath)
- env.setDegreeOfParallelism(1)
- env.execute()
- "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
- "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
- "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
+ @Test
+ def testJoinOnCustomTypeInputWithKeyExtractorAndTupleInputWithKeyFieldSelector: Unit = {
+ /*
+ * Join on a tuple input with key field selector and a custom type input with key extractor
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmallCustomTypeDataSet(env)
+ val ds2 = CollectionDataSets.get3TupleDataSet(env)
+ val joinDs = ds1.join(ds2).where( _.myInt ).equalTo(0) { (l, r) => (l.myString, r._3) }
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "Hi,Hi\n" + "Hello,Hello\n" + "Hello world,Hello\n"
+ }
- case 18 => // 20 in Java ITCase
- /*
- * Non-POJO test to verify that full-tuple keys are working.
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmallNestedTupleDataSet(env)
- val ds2 = CollectionDataSets.getSmallNestedTupleDataSet(env)
- val joinDs = ds1.join(ds2).where(0).equalTo("_1._1", "_1._2")
- joinDs.writeAsCsv(resultPath)
- env.setDegreeOfParallelism(1)
- env.execute()
- "((1,1),one),((1,1),one)\n" + "((2,2),two),((2,2),two)\n" + "((3,3),three),((3,3)," +
- "three)\n"
+ @Test
+ def testJoinOnTupleInputWithKeyFieldSelectorAndCustomTypeInputWithKeyExtractor: Unit = {
+ /*
+ * Join on a tuple input with key field selector and a custom type input with key extractor
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmall3TupleDataSet(env)
+ val ds2 = CollectionDataSets.getCustomTypeDataSet(env)
+ val joinDs = ds1.join(ds2).where(1).equalTo(_.myLong) apply { (l, r) => (l._3, r.myString) }
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "Hi,Hello\n" + "Hello,Hello world\n" + "Hello world,Hello world\n"
+ }
- case 19 => // 21 in Java ITCase
- /*
- * Non-POJO test to verify "nested" tuple-element selection.
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmallNestedTupleDataSet(env)
- val ds2 = CollectionDataSets.getSmallNestedTupleDataSet(env)
- val joinDs = ds1.join(ds2).where("_1._1").equalTo("_1._1")
- joinDs.writeAsCsv(resultPath)
- env.setDegreeOfParallelism(1)
- env.execute()
- "((1,1),one),((1,1),one)\n" + "((2,2),two),((2,2),two)\n" + "((3,3),three),((3,3),three)\n"
+ @Test
+ def testDefaultJoinOnTwoCustomTypeInputsWithKeyExtractors: Unit = {
+ /*
+ * (Default) Join on two custom type inputs with key extractors
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getCustomTypeDataSet(env)
+ val ds2 = CollectionDataSets.getSmallCustomTypeDataSet(env)
+ val joinDs = ds1.join(ds2).where(_.myInt).equalTo(_.myInt)
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1,0,Hi,1,0,Hi\n" + "2,1,Hello,2,1,Hello\n" + "2,1,Hello,2,2,Hello world\n" + "2," +
+ "2,Hello world,2,1,Hello\n" + "2,2,Hello world,2,2,Hello world\n"
+ }
- case 20 => // 22 in Java ITCase
- /*
- * full pojo with full tuple
- */
- val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
- val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
- val ds2 = CollectionDataSets.getSmallTuplebasedDataSetMatchingPojo(env)
- val joinDs = ds1.join(ds2).where("*").equalTo("*")
- joinDs.writeAsCsv(resultPath)
- env.setDegreeOfParallelism(1)
- env.execute()
- "1 First (10,100,1000,One) 10000,(10000,10,100,1000,One,1,First)\n" +
- "2 Second (20,200,2000,Two) 20000,(20000,20,200,2000,Two,2,Second)\n" +
- "3 Third (30,300,3000,Three) 30000,(30000,30,300,3000,Three,3,Third)\n"
-
- case _ =>
- throw new IllegalArgumentException("Invalid program id: " + progId)
+ @Test
+ def testUDFJoinOnTuplesWithTupleReturningKeySelectors: Unit = {
+ /*
+ * UDF Join on tuples with tuple-returning key selectors
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.get3TupleDataSet(env)
+ val ds2 = CollectionDataSets.get5TupleDataSet(env)
+ val joinDs = ds1.join(ds2).where( t => (t._1, t._2)).equalTo( t => (t._1, t._5)) apply {
+ (l, r) => (l._3, r._4)
}
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
+ "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n"
}
-}
+ @Test
+ def testNestedPojoAgainstTupleAsString: Unit = {
+ /*
+ * Join nested pojo against tuple (selected using a string)
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
+ val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
+ val joinDs = ds1.join(ds2).where("nestedPojo.longNumber").equalTo("_7")
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
+ "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
+ "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
+ }
-@RunWith(classOf[Parameterized])
-class JoinITCase(config: Configuration) extends JavaProgramTestBase(config) {
+ @Test
+ def testJoinNestedPojoAgainstTupleAsInteger: Unit = {
+ /*
+ * Join nested pojo against tuple (selected as an integer)
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
+ val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
+ val joinDs = ds1.join(ds2).where("nestedPojo.longNumber").equalTo(6) // <-- difference
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.execute()
+ expected = "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
+ "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
+ "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
+ }
- private var curProgId: Int = config.getInteger("ProgramId", -1)
- private var resultPath: String = null
- private var expectedResult: String = null
+ @Test
+ def testSelectingMultipleFieldsUsingExpressionLanguage: Unit = {
+ /*
+ * selecting multiple fields using expression language
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
+ val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
+ val joinDs = ds1.join(ds2)
+ .where("nestedPojo.longNumber", "number", "str")
+ .equalTo("_7", "_1", "_2")
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.setDegreeOfParallelism(1)
+ env.execute()
+ expected = "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
+ "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
+ "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
+ }
- protected override def preSubmit(): Unit = {
- resultPath = getTempDirPath("result")
+ @Test
+ def testNestedIntoTuple: Unit = {
+ /*
+ * nested into tuple
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
+ val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
+ val joinDs = ds1.join(ds2).where("nestedPojo.longNumber", "number",
+ "nestedTupleWithCustom._1").equalTo("_7", "_1", "_3")
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.setDegreeOfParallelism(1)
+ env.execute()
+ expected = "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
+ "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
+ "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
}
- protected def testProgram(): Unit = {
- expectedResult = JoinProgs.runProgram(curProgId, resultPath)
+ @Test
+ def testNestedIntoTupleIntoPojo: Unit = {
+ /*
+ * nested into tuple into pojo
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
+ val ds2 = CollectionDataSets.getSmallTuplebasedPojoMatchingDataSet(env)
+ val joinDs = ds1.join(ds2)
+ .where("nestedTupleWithCustom._1",
+ "nestedTupleWithCustom._2.myInt",
+ "nestedTupleWithCustom._2.myLong")
+ .equalTo("_3", "_4", "_5")
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.setDegreeOfParallelism(1)
+ env.execute()
+ expected = "1 First (10,100,1000,One) 10000,(1,First,10,100,1000,One," +
+ "10000)\n" + "2 Second (20,200,2000,Two) 20000,(2,Second,20,200,2000,Two," +
+ "20000)\n" + "3 Third (30,300,3000,Three) 30000,(3,Third,30,300,3000,Three,30000)\n"
}
- protected override def postSubmit(): Unit = {
- compareResultsByLinesInMemory(expectedResult, resultPath)
+ @Test
+ def testNonPojoFullTuple: Unit = {
+ /*
+ * Non-POJO test to verify that full-tuple keys are working.
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmallNestedTupleDataSet(env)
+ val ds2 = CollectionDataSets.getSmallNestedTupleDataSet(env)
+ val joinDs = ds1.join(ds2).where(0).equalTo("_1._1", "_1._2")
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.setDegreeOfParallelism(1)
+ env.execute()
+ expected = "((1,1),one),((1,1),one)\n" + "((2,2),two),((2,2),two)\n" + "((3,3),three),((3,3)," +
+ "three)\n"
+
}
-}
-object JoinITCase {
- @Parameters
- def getConfigurations: java.util.Collection[Array[AnyRef]] = {
- val configs = mutable.MutableList[Array[AnyRef]]()
- for (i <- 1 to JoinProgs.NUM_PROGRAMS) {
- val config = new Configuration()
- config.setInteger("ProgramId", i)
- configs += Array(config)
- }
+ @Test
+ def testNonPojoNestedTuple: Unit = {
+ /*
+ * Non-POJO test to verify "nested" tuple-element selection.
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmallNestedTupleDataSet(env)
+ val ds2 = CollectionDataSets.getSmallNestedTupleDataSet(env)
+ val joinDs = ds1.join(ds2).where("_1._1").equalTo("_1._1")
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.setDegreeOfParallelism(1)
+ env.execute()
+ expected = "((1,1),one),((1,1),one)\n" + "((2,2),two),((2,2),two)\n" + "((3,3),three),((3,3)," +
+ "three)\n"
+ }
- configs.asJavaCollection
+ @Test
+ def testFullPojoWithFullTuple: Unit = {
+ /*
+ * full pojo with full tuple
+ */
+ val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+ val ds1 = CollectionDataSets.getSmallPojoDataSet(env)
+ val ds2 = CollectionDataSets.getSmallTuplebasedDataSetMatchingPojo(env)
+ val joinDs = ds1.join(ds2).where("*").equalTo("*")
+ joinDs.writeAsCsv(resultPath, writeMode = WriteMode.OVERWRITE)
+ env.setDegreeOfParallelism(1)
+ env.execute()
+ expected = "1 First (10,100,1000,One) 10000,(10000,10,100,1000,One,1,First)\n" +
+ "2 Second (20,200,2000,Two) 20000,(20000,20,200,2000,Two,2,Second)\n" +
+ "3 Third (30,300,3000,Three) 30000,(30000,30,300,3000,Three,3,Third)\n"
}
}
-