You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@wayang.apache.org by zk...@apache.org on 2023/05/08 06:36:28 UTC
[incubator-wayang] branch main updated: Updated Flink version and added some operator tests
This is an automated email from the ASF dual-hosted git repository.
zkaoudi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
The following commit(s) were added to refs/heads/main by this push:
new 312d4644 Updated Flink version and added some operator tests
new 2eb6ae9a Merge pull request #319 from damik3/flink-operator-tests
312d4644 is described below
commit 312d46445d4b13be584fd8d730825e4a18bcacd2
Author: damik3 <d1...@hotmail.com>
AuthorDate: Sat May 6 15:29:45 2023 +0300
Updated Flink version and added some operator tests
---
.../flink/operators/FlinkCoGroupOperatorTest.java | 121 +++++++++++++++++++++
.../operators/FlinkGlobalReduceOperatorTest.java | 102 +++++++++++++++++
.../flink/operators/FlinkJoinOperatorTest.java | 5 +-
.../operators/FlinkMapPartitionsOperatorTest.java | 72 ++++++++++++
.../flink/operators/FlinkReduceByOperatorTest.java | 12 +-
wayang-platforms/wayang-flink/pom.xml | 2 +-
wayang-tests-integration/pom.xml | 2 +-
7 files changed, 306 insertions(+), 10 deletions(-)
diff --git a/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkCoGroupOperatorTest.java b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkCoGroupOperatorTest.java
new file mode 100644
index 00000000..26cd6e22
--- /dev/null
+++ b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkCoGroupOperatorTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.wayang.flink.operators;
+
+import org.apache.wayang.basic.data.Tuple2;
+import org.apache.wayang.basic.function.ProjectionDescriptor;
+import org.apache.wayang.core.platform.ChannelInstance;
+import org.apache.wayang.core.types.DataSetType;
+import org.apache.wayang.core.types.DataUnitType;
+import org.apache.wayang.core.util.Tuple;
+import org.apache.wayang.core.util.WayangCollections;
+import org.apache.wayang.flink.channels.DataSetChannel;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.*;
+
+/**
+ * Test suite for {@link FlinkCoGroupOperator}.
+ */
+public class FlinkCoGroupOperatorTest extends FlinkOperatorTestBase {
+
+ @Test
+ public void testExecution() throws Exception {
+ // Prepare test data.
+ DataSetChannel.Instance input0 = this.createDataSetChannelInstance(Arrays.asList(
+ new Tuple2<>(1, "b"), new Tuple2<>(1, "c"), new Tuple2<>(2, "d"), new Tuple2<>(3, "e")));
+ DataSetChannel.Instance input1 = this.createDataSetChannelInstance(Arrays.asList(
+ new Tuple2<>("x", 1), new Tuple2<>("y", 1), new Tuple2<>("z", 2), new Tuple2<>("w", 4)));
+ DataSetChannel.Instance output = this.createDataSetChannelInstance();
+
+ // Build the operator.
+ FlinkCoGroupOperator<Tuple2, Tuple2, Integer> coGroup =
+ new FlinkCoGroupOperator<>(
+ new ProjectionDescriptor<>(
+ DataUnitType.createBasicUnchecked(Tuple2.class),
+ DataUnitType.createBasic(Integer.class),
+ "field0"),
+ new ProjectionDescriptor<>(
+ DataUnitType.createBasicUnchecked(Tuple2.class),
+ DataUnitType.createBasic(Integer.class),
+ "field1"),
+ DataSetType.createDefaultUnchecked(Tuple2.class),
+ DataSetType.createDefaultUnchecked(Tuple2.class)
+ );
+
+ // Set up the ChannelInstances.
+ final ChannelInstance[] inputs = new ChannelInstance[]{input0, input1};
+ final ChannelInstance[] outputs = new ChannelInstance[]{output};
+
+ // Execute.
+ this.evaluate(coGroup, inputs, outputs);
+
+ // Verify the outcome.
+ final List<Tuple2<Iterable<Tuple2<Integer, String>>, Iterable<Tuple2<String, Integer>>>> result =
+ output.<Tuple2<Iterable<Tuple2<Integer, String>>, Iterable<Tuple2<String, Integer>>>>provideDataSet().collect();
+ Collection<Tuple<Collection<Tuple2<Integer, String>>, Collection<Tuple2<String, Integer>>>> expectedGroups =
+ new ArrayList<>(Arrays.asList(
+ new Tuple<Collection<Tuple2<Integer, String>>, Collection<Tuple2<String, Integer>>>(
+ Arrays.asList(new Tuple2<>(1, "b"), new Tuple2<>(1, "c")),
+ Arrays.asList(new Tuple2<>("x", 1), new Tuple2<>("y", 1))
+ ),
+ new Tuple<Collection<Tuple2<Integer, String>>, Collection<Tuple2<String, Integer>>>(
+ Collections.singletonList(new Tuple2<>(2, "d")),
+ Collections.singletonList(new Tuple2<>("z", 2))
+ ), new Tuple<Collection<Tuple2<Integer, String>>, Collection<Tuple2<String, Integer>>>(
+ Collections.singletonList(new Tuple2<>(3, "e")),
+ Collections.emptyList()
+ ),
+ new Tuple<Collection<Tuple2<Integer, String>>, Collection<Tuple2<String, Integer>>>(
+ Collections.emptyList(),
+ Collections.singletonList(new Tuple2<>("w", 4))
+ )
+ ));
+
+ ResultLoop:
+ for (Tuple2<Iterable<Tuple2<Integer, String>>, Iterable<Tuple2<String, Integer>>> resultCoGroup : result) {
+ for (Iterator<Tuple<Collection<Tuple2<Integer, String>>, Collection<Tuple2<String, Integer>>>> i = expectedGroups.iterator();
+ i.hasNext(); ) {
+ Tuple<Collection<Tuple2<Integer, String>>, Collection<Tuple2<String, Integer>>> expectedGroup = i.next();
+ if (this.compare(expectedGroup, resultCoGroup)) {
+ i.remove();
+ continue ResultLoop;
+ }
+ }
+ Assert.fail(String.format("Unexpected group: %s", resultCoGroup));
+ }
+ Assert.assertTrue(
+ String.format("Missing groups: %s", expectedGroups),
+ expectedGroups.isEmpty()
+ );
+ }
+
+ private boolean compare(Tuple<Collection<Tuple2<Integer, String>>, Collection<Tuple2<String, Integer>>> expected,
+ Tuple2<Iterable<Tuple2<Integer, String>>, Iterable<Tuple2<String, Integer>>> actual) {
+ return this.compareGroup(expected.field0, actual.field0) && this.compareGroup(expected.field1, actual.field1);
+ }
+
+ private <T> boolean compareGroup(Collection<T> expected, Iterable<T> actual) {
+ if (expected == null) return actual == null;
+ if (actual == null) return false;
+
+ return WayangCollections.asSet(expected).equals(WayangCollections.asSet(actual));
+ }
+}
diff --git a/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkGlobalReduceOperatorTest.java b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkGlobalReduceOperatorTest.java
new file mode 100644
index 00000000..db51b102
--- /dev/null
+++ b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkGlobalReduceOperatorTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.wayang.flink.operators;
+
+import org.apache.wayang.basic.data.Tuple2;
+import org.apache.wayang.core.function.ReduceDescriptor;
+import org.apache.wayang.core.platform.ChannelInstance;
+import org.apache.wayang.core.types.DataSetType;
+import org.apache.wayang.core.types.DataUnitType;
+import org.apache.wayang.core.util.WayangCollections;
+import org.apache.wayang.flink.channels.DataSetChannel;
+import org.apache.wayang.java.channels.CollectionChannel;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Test suite for {@link FlinkGlobalReduceOperator}.
+ */
+public class FlinkGlobalReduceOperatorTest extends FlinkOperatorTestBase {
+
+ @Test
+ public void testExecution() throws Exception {
+ // Prepare test data.
+ DataSetChannel.Instance input = this.createDataSetChannelInstance(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
+ DataSetChannel.Instance output = this.createDataSetChannelInstance();
+
+ // Build the reduce operator.
+ FlinkGlobalReduceOperator<Integer> globalReduce =
+ new FlinkGlobalReduceOperator<>(
+ DataSetType.createDefaultUnchecked(Tuple2.class),
+ new ReduceDescriptor<>(
+ (a, b) -> a + b, DataUnitType.createGrouped(Integer.class),
+ DataUnitType.createBasic(Integer.class)
+ )
+ );
+
+ // Set up the ChannelInstances.
+ final ChannelInstance[] inputs = new ChannelInstance[]{input};
+ final ChannelInstance[] outputs = new ChannelInstance[]{output};
+
+ // Execute.
+ this.evaluate(globalReduce, inputs, outputs);
+
+ // Verify the outcome.
+ final List<Integer> result = output.<Integer>provideDataSet().collect();
+ Assert.assertEquals(1, result.size());
+ Assert.assertEquals(Integer.valueOf((10 + 1) * (10 / 2)), result.get(0)); // Props to Gauss!
+
+ }
+
+ @Ignore("Flink cannot reduce empty collections.")
+ @Test
+ public void testExecutionWithoutData() throws Exception {
+ // Prepare test data.
+ DataSetChannel.Instance input = this.createDataSetChannelInstance(Collections.emptyList());
+ CollectionChannel.Instance output = this.createCollectionChannelInstance();
+
+ // Build the reduce operator.
+ FlinkGlobalReduceOperator<Integer> globalReduce =
+ new FlinkGlobalReduceOperator<>(
+ DataSetType.createDefaultUnchecked(Tuple2.class),
+ new ReduceDescriptor<>(
+ (a, b) -> a + b, DataUnitType.createGrouped(Integer.class),
+ DataUnitType.createBasic(Integer.class)
+ )
+ );
+
+ // Set up the ChannelInstances.
+ final ChannelInstance[] inputs = new ChannelInstance[]{input};
+ final ChannelInstance[] outputs = new ChannelInstance[]{output};
+
+ // Execute.
+ this.evaluate(globalReduce, inputs, outputs);
+
+ // Verify the outcome.
+ final List<Integer> result = WayangCollections.asList(output.provideCollection());
+ Assert.assertEquals(1, result.size());
+ Assert.assertEquals(Integer.valueOf(0), result.get(0));
+
+ }
+}
diff --git a/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkJoinOperatorTest.java b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkJoinOperatorTest.java
index 75313b45..db703b9e 100644
--- a/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkJoinOperatorTest.java
+++ b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkJoinOperatorTest.java
@@ -25,14 +25,12 @@ import org.apache.wayang.core.types.DataSetType;
import org.apache.wayang.core.types.DataUnitType;
import org.apache.wayang.flink.channels.DataSetChannel;
import org.junit.Assert;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Disabled;
+import org.junit.Test;
import java.util.Arrays;
import java.util.List;
-//problematic
/**
* Test suite for {@link FlinkJoinOperator}.
*/
@@ -44,7 +42,6 @@ public class FlinkJoinOperatorTest extends FlinkOperatorTestBase{
// implementation of the implementation in the operator
// labels:flink,bug
@Test
- @Disabled("until validation of implementation of the FlinkJoinOperator")
public void testExecution() throws Exception {
// Prepare test data.
DataSetChannel.Instance input0 = this.createDataSetChannelInstance(Arrays.asList(
diff --git a/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkMapPartitionsOperatorTest.java b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkMapPartitionsOperatorTest.java
new file mode 100644
index 00000000..18ed5019
--- /dev/null
+++ b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkMapPartitionsOperatorTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.wayang.flink.operators;
+
+import org.apache.wayang.core.function.MapPartitionsDescriptor;
+import org.apache.wayang.core.platform.ChannelInstance;
+import org.apache.wayang.core.types.DataSetType;
+import org.apache.wayang.flink.channels.DataSetChannel;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Test suite for {@link FlinkMapPartitionsOperator}.
+ */
+public class FlinkMapPartitionsOperatorTest extends FlinkOperatorTestBase {
+
+ @Test
+ public void testExecution() throws Exception {
+ // Prepare test data.
+ DataSetChannel.Instance input = this.createDataSetChannelInstance(Arrays.asList(0, 1, 1, 2, 6));
+ DataSetChannel.Instance output = this.createDataSetChannelInstance();
+
+ // Create the mapPartitions operator.
+ FlinkMapPartitionsOperator<Integer, Integer> mapPartitionsOperator =
+ new FlinkMapPartitionsOperator<>(
+ new MapPartitionsDescriptor<>(items -> {
+ Collection<Integer> result = new LinkedList<>();
+ for (Integer item : items) {
+ result.add(item + 1);
+ }
+ return result;
+ }, Integer.class, Integer.class),
+ DataSetType.createDefaultUnchecked(Integer.class),
+ DataSetType.createDefaultUnchecked(Integer.class)
+ );
+
+ // Set up the ChannelInstances.
+ ChannelInstance[] inputs = new ChannelInstance[]{input};
+ ChannelInstance[] outputs = new ChannelInstance[]{output};
+
+ // Execute.
+ this.evaluate(mapPartitionsOperator, inputs, outputs);
+
+ // Verify the outcome.
+ final List<Integer> result = output.<Integer>provideDataSet().collect();
+ Assert.assertEquals(5, result.size());
+ Assert.assertEquals(Arrays.asList(1, 2, 2, 3, 7), result);
+
+ }
+
+}
diff --git a/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkReduceByOperatorTest.java b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkReduceByOperatorTest.java
index 63157809..0214d795 100644
--- a/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkReduceByOperatorTest.java
+++ b/wayang-platforms/wayang-flink/code/test/java/org/apache/wayang/flink/operators/FlinkReduceByOperatorTest.java
@@ -26,8 +26,7 @@ import org.apache.wayang.core.types.DataSetType;
import org.apache.wayang.core.types.DataUnitType;
import org.apache.wayang.flink.channels.DataSetChannel;
import org.junit.Assert;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Disabled;
+import org.junit.Test;
import java.util.Arrays;
import java.util.HashSet;
@@ -46,7 +45,6 @@ public class FlinkReduceByOperatorTest extends FlinkOperatorTestBase{
// implementation of the implementation in the operator
// labels:flink,bug
@Test
- @Disabled("until validation of implementation of the FlinkReduceByOperator")
public void testExecution() throws Exception {
// Prepare test data.
List<Tuple2<String, Integer>> inputList = Arrays.stream("aaabbccccdeefff".split(""))
@@ -77,7 +75,13 @@ public class FlinkReduceByOperatorTest extends FlinkOperatorTestBase{
final ChannelInstance[] outputs = new ChannelInstance[]{output};
// Execute.
- this.evaluate(reduceByOperator, inputs, outputs);
+ try {
+ this.evaluate(reduceByOperator, inputs, outputs);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
// Verify the outcome.
final Iterable<Tuple2<String, Integer>> result = output.<Tuple2<String, Integer>>provideDataSet().collect();
diff --git a/wayang-platforms/wayang-flink/pom.xml b/wayang-platforms/wayang-flink/pom.xml
index f6466edf..843cddd3 100644
--- a/wayang-platforms/wayang-flink/pom.xml
+++ b/wayang-platforms/wayang-flink/pom.xml
@@ -37,7 +37,7 @@
<properties>
<java-module-name>org.apache.wayang.platform.flink</java-module-name>
- <flink.version>1.7.1</flink.version>
+ <flink.version>1.10.0</flink.version>
</properties>
<dependencies>
diff --git a/wayang-tests-integration/pom.xml b/wayang-tests-integration/pom.xml
index a2f1f01e..c57cf0c0 100644
--- a/wayang-tests-integration/pom.xml
+++ b/wayang-tests-integration/pom.xml
@@ -36,7 +36,7 @@
<properties>
<java-module-name>org.apache.wayang.test.integration</java-module-name>
<graphchi.version>0.2.2</graphchi.version>
- <flink.version>1.7.1</flink.version>
+ <flink.version>1.10.0</flink.version>
<giraph.version>1.2.0-hadoop2</giraph.version>
</properties>