You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gr...@apache.org on 2017/07/12 23:44:19 UTC
[19/22] flink git commit: [FLINK-6731] [tests] Activate strict
checkstyle for flink-tests
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountITCase.java
new file mode 100644
index 0000000..3dfb672
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountITCase.java
@@ -0,0 +1,51 @@
+/*
+ * 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.flink.test.example.java;
+
+import org.apache.flink.examples.java.wordcount.WordCount;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
+/**
+ * Test {@link WordCount}.
+ */
+public class WordCountITCase extends JavaProgramTestBase {
+
+ protected String textPath;
+ protected String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("text.txt", WordCountData.TEXT);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ WordCount.main(new String[] {
+ "--input", textPath,
+ "--output", resultPath });
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountNestedPOJOITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountNestedPOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountNestedPOJOITCase.java
new file mode 100644
index 0000000..3ee9e5e
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountNestedPOJOITCase.java
@@ -0,0 +1,129 @@
+/*
+ * 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.flink.test.example.java;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * WordCount with nested POJO example.
+ */
+@SuppressWarnings("serial")
+public class WordCountNestedPOJOITCase extends JavaProgramTestBase implements Serializable {
+ private static final long serialVersionUID = 1L;
+ protected String textPath;
+ protected String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("text.txt", WordCountData.TEXT);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ DataSet<String> text = env.readTextFile(textPath);
+
+ DataSet<WC> counts = text
+ .flatMap(new Tokenizer())
+ .groupBy("complex.someTest")
+ .reduce(new ReduceFunction<WC>() {
+ private static final long serialVersionUID = 1L;
+ public WC reduce(WC value1, WC value2) {
+ return new WC(value1.complex.someTest, value1.count + value2.count);
+ }
+ });
+
+ counts.writeAsText(resultPath);
+
+ env.execute("WordCount with custom data types example");
+ }
+
+ private static final class Tokenizer implements FlatMapFunction<String, WC> {
+
+ @Override
+ public void flatMap(String value, Collector<WC> out) {
+ // normalize and split the line
+ String[] tokens = value.toLowerCase().split("\\W+");
+
+ // emit the pairs
+ for (String token : tokens) {
+ if (token.length() > 0) {
+ out.collect(new WC(token, 1));
+ }
+ }
+ }
+ }
+
+ /**
+ * POJO with nested POJO.
+ */
+ public static class WC { // is a pojo
+ public ComplexNestedClass complex; // is a pojo
+ public int count; // is a BasicType
+
+ public WC() {
+ }
+
+ public WC(String t, int c) {
+ this.count = c;
+ this.complex = new ComplexNestedClass();
+ this.complex.word = new Tuple3<Long, Long, String>(0L, 0L, "egal");
+ this.complex.date = new Date();
+ this.complex.someFloat = 0.0f;
+ this.complex.someNumber = 666;
+ this.complex.someTest = t;
+ }
+
+ @Override
+ public String toString() {
+ return this.complex.someTest + " " + count;
+ }
+ }
+
+ /**
+ * Nested POJO.
+ */
+ public static class ComplexNestedClass { // pojo
+ public static int ignoreStaticField;
+ public transient int ignoreTransientField;
+ public Date date; // generic type
+ public Integer someNumber; // BasicType
+ public float someFloat; // BasicType
+ public Tuple3<Long, Long, String> word; //Tuple Type with three basic types
+ public String someTest;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSimplePOJOITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSimplePOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSimplePOJOITCase.java
new file mode 100644
index 0000000..ab76c12
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSimplePOJOITCase.java
@@ -0,0 +1,110 @@
+/*
+ * 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.flink.test.example.java;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+
+/**
+ * WordCount with simple POJO example.
+ */
+public class WordCountSimplePOJOITCase extends JavaProgramTestBase implements Serializable {
+ private static final long serialVersionUID = 1L;
+ protected String textPath;
+ protected String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("text.txt", WordCountData.TEXT);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<String> text = env.readTextFile(textPath);
+
+ DataSet<WC> counts = text
+ .flatMap(new Tokenizer())
+ .groupBy("word")
+ .reduce(new ReduceFunction<WC>() {
+ private static final long serialVersionUID = 1L;
+
+ public WC reduce(WC value1, WC value2) {
+ return new WC(value1.word, value1.count + value2.count);
+ }
+ });
+
+ counts.writeAsText(resultPath);
+
+ env.execute("WordCount with custom data types example");
+ }
+
+ private static final class Tokenizer implements FlatMapFunction<String, WC> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void flatMap(String value, Collector<WC> out) {
+ // normalize and split the line
+ String[] tokens = value.toLowerCase().split("\\W+");
+
+ // emit the pairs
+ for (String token : tokens) {
+ if (token.length() > 0) {
+ out.collect(new WC(token, 1));
+ }
+ }
+ }
+ }
+
+ /**
+ * POJO with word and count.
+ */
+ public static class WC {
+ public WC() {}
+
+ public WC(String w, int c) {
+ word = w;
+ count = c;
+ }
+
+ public String word;
+ public int count;
+
+ @Override
+ public String toString() {
+ return word + " " + count;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassInterfacePOJOITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassInterfacePOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassInterfacePOJOITCase.java
new file mode 100644
index 0000000..688916c
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassInterfacePOJOITCase.java
@@ -0,0 +1,169 @@
+/*
+ * 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.flink.test.example.java;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+
+/**
+ * WordCount with subclass and interface example.
+ */
+@SuppressWarnings("serial")
+public class WordCountSubclassInterfacePOJOITCase extends JavaProgramTestBase implements Serializable {
+ private static final long serialVersionUID = 1L;
+ protected String textPath;
+ protected String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("text.txt", WordCountData.TEXT);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ DataSet<String> text = env.readTextFile(textPath);
+
+ DataSet<WCBase> counts = text
+ .flatMap(new Tokenizer())
+ .groupBy("word")
+ .reduce(new ReduceFunction<WCBase>() {
+ private static final long serialVersionUID = 1L;
+ public WCBase reduce(WCBase value1, WCBase value2) {
+ WC wc1 = (WC) value1;
+ WC wc2 = (WC) value2;
+ int c = wc1.secretCount.getCount() + wc2.secretCount.getCount();
+ wc1.secretCount.setCount(c);
+ return wc1;
+ }
+ })
+ .map(new MapFunction<WCBase, WCBase>() {
+ @Override
+ public WCBase map(WCBase value) throws Exception {
+ WC wc = (WC) value;
+ wc.count = wc.secretCount.getCount();
+ return wc;
+ }
+ });
+
+ counts.writeAsText(resultPath);
+
+ env.execute("WordCount with custom data types example");
+ }
+
+ private static final class Tokenizer implements FlatMapFunction<String, WCBase> {
+
+ @Override
+ public void flatMap(String value, Collector<WCBase> out) {
+ // normalize and split the line
+ String[] tokens = value.toLowerCase().split("\\W+");
+ // emit the pairs
+ for (String token : tokens) {
+ if (token.length() > 0) {
+ out.collect(new WC(token, 1));
+ }
+ }
+ }
+ }
+
+ /**
+ * Abstract POJO.
+ */
+ public abstract static class WCBase {
+ public String word;
+ public int count;
+
+ public WCBase(String w, int c) {
+ this.word = w;
+ this.count = c;
+ }
+
+ @Override
+ public String toString() {
+ return word + " " + count;
+ }
+ }
+
+ /**
+ * POJO interface.
+ */
+ public interface CrazyCounter {
+ int getCount();
+
+ void setCount(int c);
+ }
+
+ /**
+ * Implementation of POJO interface.
+ */
+ public static class CrazyCounterImpl implements CrazyCounter {
+ public int countz;
+
+ public CrazyCounterImpl() {
+ }
+
+ public CrazyCounterImpl(int c) {
+ this.countz = c;
+ }
+
+ @Override
+ public int getCount() {
+ return countz;
+ }
+
+ @Override
+ public void setCount(int c) {
+ this.countz = c;
+ }
+
+ }
+
+ /**
+ * Subclass of abstract POJO.
+ */
+ public static class WC extends WCBase {
+ public CrazyCounter secretCount;
+
+ public WC() {
+ super(null, 0);
+ }
+
+ public WC(String w, int c) {
+ super(w, 0);
+ this.secretCount = new CrazyCounterImpl(c);
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassPOJOITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassPOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassPOJOITCase.java
new file mode 100644
index 0000000..450af37
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountSubclassPOJOITCase.java
@@ -0,0 +1,133 @@
+/*
+ * 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.flink.test.example.java;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+
+/**
+ * WordCount with custom data types example.
+ */
+@SuppressWarnings("serial")
+public class WordCountSubclassPOJOITCase extends JavaProgramTestBase implements Serializable {
+ private static final long serialVersionUID = 1L;
+ protected String textPath;
+ protected String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("text.txt", WordCountData.TEXT);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ DataSet<String> text = env.readTextFile(textPath);
+
+ DataSet<WCBase> counts = text
+ .flatMap(new Tokenizer())
+ .groupBy("word")
+ .reduce(new ReduceFunction<WCBase>() {
+ private static final long serialVersionUID = 1L;
+ public WCBase reduce(WCBase value1, WCBase value2) {
+ WC wc1 = (WC) value1;
+ WC wc2 = (WC) value2;
+ return new WC(value1.word, wc1.secretCount + wc2.secretCount);
+ }
+ })
+ .map(new MapFunction<WCBase, WCBase>() {
+ @Override
+ public WCBase map(WCBase value) throws Exception {
+ WC wc = (WC) value;
+ wc.count = wc.secretCount;
+ return wc;
+ }
+ });
+
+ counts.writeAsText(resultPath);
+
+ env.execute("WordCount with custom data types example");
+ }
+
+ private static final class Tokenizer implements FlatMapFunction<String, WCBase> {
+
+ @Override
+ public void flatMap(String value, Collector<WCBase> out) {
+ // normalize and split the line
+ String[] tokens = value.toLowerCase().split("\\W+");
+ // emit the pairs
+ for (String token : tokens) {
+ if (token.length() > 0) {
+ out.collect(new WC(token, 1));
+ }
+ }
+ }
+ }
+
+ /**
+ * Abstract POJO.
+ */
+ public abstract static class WCBase {
+ public String word;
+ public int count;
+
+ public WCBase(String w, int c) {
+ this.word = w;
+ this.count = c;
+ }
+
+ @Override
+ public String toString() {
+ return word + " " + count;
+ }
+ }
+
+ /**
+ * Subclass of abstract POJO.
+ */
+ public static class WC extends WCBase {
+
+ public int secretCount;
+
+ public WC() {
+ super(null, 0);
+ }
+
+ public WC(String w, int c) {
+ super(w, 0);
+ this.secretCount = c;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountWithCollectionITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountWithCollectionITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountWithCollectionITCase.java
new file mode 100644
index 0000000..405026c
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/java/WordCountWithCollectionITCase.java
@@ -0,0 +1,70 @@
+/*
+ * 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.flink.test.example.java;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.aggregation.Aggregations;
+import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.examples.java.wordcount.WordCount;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
+import org.junit.Assert;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * WordCount with collection example.
+ */
+public class WordCountWithCollectionITCase extends JavaProgramTestBase {
+
+ private final List<Tuple2<String, Integer>> resultsCollected = new ArrayList<Tuple2<String, Integer>>();
+
+ @Override
+ protected void postSubmit() throws Exception {
+ String[] result = new String[resultsCollected.size()];
+ for (int i = 0; i < result.length; i++) {
+ result[i] = resultsCollected.get(i).toString();
+ }
+ Arrays.sort(result);
+
+ String[] expected = WordCountData.COUNTS_AS_TUPLES.split("\n");
+ Arrays.sort(expected);
+
+ Assert.assertEquals("Different number of lines in expected and obtained result.", expected.length, result.length);
+ Assert.assertArrayEquals(expected, result);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<String> text = env.fromElements(WordCountData.TEXT);
+ DataSet<Tuple2<String, Integer>> words = text.flatMap(new WordCount.Tokenizer());
+ DataSet<Tuple2<String, Integer>> result = words.groupBy(0).aggregate(Aggregations.SUM, 1);
+
+ result.output(new LocalCollectionOutputFormat<Tuple2<String, Integer>>(resultsCollected));
+ env.execute("Word Count Collection");
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/scala/ConnectedComponentsITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/scala/ConnectedComponentsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/scala/ConnectedComponentsITCase.java
new file mode 100644
index 0000000..0391556
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/scala/ConnectedComponentsITCase.java
@@ -0,0 +1,65 @@
+/*
+ * 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.flink.test.example.scala;
+
+import org.apache.flink.examples.scala.graph.ConnectedComponents;
+import org.apache.flink.test.testdata.ConnectedComponentsData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
+import java.io.BufferedReader;
+
+/**
+ * Test for {@link ConnectedComponents}.
+ */
+public class ConnectedComponentsITCase extends JavaProgramTestBase {
+
+ private static final long SEED = 0xBADC0FFEEBEEFL;
+
+ private static final int NUM_VERTICES = 1000;
+
+ private static final int NUM_EDGES = 10000;
+
+ private String verticesPath;
+ private String edgesPath;
+ private String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ verticesPath = createTempFile("vertices.txt", ConnectedComponentsData.getEnumeratingVertices(NUM_VERTICES));
+ edgesPath = createTempFile("edges.txt", ConnectedComponentsData.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED));
+ resultPath = getTempFilePath("results");
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ ConnectedComponents.main(new String[] {
+ "--vertices", verticesPath,
+ "--edges", edgesPath,
+ "--output", resultPath,
+ "--iterations", "100"});
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ for (BufferedReader reader : getResultReader(resultPath)) {
+ ConnectedComponentsData.checkOddEvenResult(reader);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/scala/EnumTriangleITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/scala/EnumTriangleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/scala/EnumTriangleITCase.java
new file mode 100644
index 0000000..7f9ac0e
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/scala/EnumTriangleITCase.java
@@ -0,0 +1,52 @@
+/*
+ * 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.flink.test.example.scala;
+
+import org.apache.flink.examples.scala.graph.EnumTriangles;
+import org.apache.flink.test.testdata.EnumTriangleData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
+/**
+ * Test {@link EnumTriangles}.
+ */
+public class EnumTriangleITCase extends JavaProgramTestBase {
+
+ protected String edgePath;
+ protected String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ edgePath = createTempFile("edges", EnumTriangleData.EDGES);
+ resultPath = getTempDirPath("triangles");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(EnumTriangleData.TRIANGLES_BY_ID, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ EnumTriangles.main(new String[] {
+ "--edges", edgePath,
+ "--output", resultPath });
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/scala/PageRankITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/scala/PageRankITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/scala/PageRankITCase.java
new file mode 100644
index 0000000..93c4f56
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/scala/PageRankITCase.java
@@ -0,0 +1,100 @@
+/*
+ * 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.flink.test.example.scala;
+
+import org.apache.flink.examples.scala.graph.PageRankBasic;
+import org.apache.flink.test.testdata.PageRankData;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.File;
+
+/**
+ * Test for {@link PageRankBasic}.
+ */
+@RunWith(Parameterized.class)
+public class PageRankITCase extends MultipleProgramsTestBase {
+
+ public PageRankITCase(TestExecutionMode mode){
+ super(mode);
+ }
+
+ private String verticesPath;
+ private String edgesPath;
+ private String resultPath;
+ private String expected;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ public void before() throws Exception{
+ File resultFile = tempFolder.newFile();
+ //Delete file because the Scala API does not respect WriteMode set by the configuration
+ resultFile.delete();
+ resultPath = resultFile.toURI().toString();
+
+ File verticesFile = tempFolder.newFile();
+ Files.write(PageRankData.VERTICES, verticesFile, Charsets.UTF_8);
+
+ File edgesFile = tempFolder.newFile();
+ Files.write(PageRankData.EDGES, edgesFile, Charsets.UTF_8);
+
+ verticesPath = verticesFile.toURI().toString();
+ edgesPath = edgesFile.toURI().toString();
+ }
+
+ @After
+ public void after() throws Exception{
+ compareKeyValuePairsWithDelta(expected, resultPath, " ", 0.01);
+ }
+
+ @Test
+ public void testPageRankWithSmallNumberOfIterations() throws Exception {
+ PageRankBasic.main(new String[] {
+ "--pages", verticesPath,
+ "--links", edgesPath,
+ "--output", resultPath,
+ "--numPages", PageRankData.NUM_VERTICES + "",
+ "--iterations", "3"});
+ expected = PageRankData.RANKS_AFTER_3_ITERATIONS;
+ }
+
+ @Test
+ public void testPageRankWithConvergence() throws Exception {
+ // start with a very high number of iteration such that the dynamic convergence criterion must handle termination
+ PageRankBasic.main(new String[] {
+ "--pages", verticesPath,
+ "--links", edgesPath,
+ "--output", resultPath,
+ "--numPages", PageRankData.NUM_VERTICES + "",
+ "--iterations", "1000"});
+ expected = PageRankData.RANKS_AFTER_EPSILON_0_0001_CONVERGENCE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/scala/TransitiveClosureITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/scala/TransitiveClosureITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/scala/TransitiveClosureITCase.java
new file mode 100644
index 0000000..3963801
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/scala/TransitiveClosureITCase.java
@@ -0,0 +1,64 @@
+/*
+ * 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.flink.test.example.scala;
+
+import org.apache.flink.examples.scala.graph.TransitiveClosureNaive;
+import org.apache.flink.test.testdata.ConnectedComponentsData;
+import org.apache.flink.test.testdata.TransitiveClosureData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
+import java.io.BufferedReader;
+
+/**
+ * Test for {@link TransitiveClosureNaive}.
+ */
+public class TransitiveClosureITCase extends JavaProgramTestBase {
+
+ private static final long SEED = 0xBADC0FFEEBEEFL;
+
+ private static final int NUM_VERTICES = 100;
+
+ private static final int NUM_EDGES = 500;
+
+ private String edgesPath;
+ private String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ edgesPath = createTempFile("edges.txt", ConnectedComponentsData.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED));
+ resultPath = getTempFilePath("results");
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ TransitiveClosureNaive.main(new String [] {
+ "--edges", edgesPath,
+ "--output", resultPath,
+ "--iterations", "5"});
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ for (BufferedReader reader : getResultReader(resultPath)) {
+ TransitiveClosureData.checkOddEvenResult(reader);
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/scala/WebLogAnalysisITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/scala/WebLogAnalysisITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/scala/WebLogAnalysisITCase.java
new file mode 100644
index 0000000..962a8a0
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/scala/WebLogAnalysisITCase.java
@@ -0,0 +1,57 @@
+/*
+ * 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.flink.test.example.scala;
+
+import org.apache.flink.examples.scala.relational.WebLogAnalysis;
+import org.apache.flink.test.testdata.WebLogAnalysisData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
+/**
+ * Test for {@link WebLogAnalysis}.
+ */
+public class WebLogAnalysisITCase extends JavaProgramTestBase {
+
+ private String docsPath;
+ private String ranksPath;
+ private String visitsPath;
+ private String resultPath;
+
+ @Override
+ protected void preSubmit() throws Exception {
+ docsPath = createTempFile("docs", WebLogAnalysisData.DOCS);
+ ranksPath = createTempFile("ranks", WebLogAnalysisData.RANKS);
+ visitsPath = createTempFile("visits", WebLogAnalysisData.VISITS);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(WebLogAnalysisData.EXCEPTED_RESULT, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ WebLogAnalysis.main(new String[]{
+ "--documents", docsPath,
+ "--ranks", ranksPath,
+ "--visits", visitsPath,
+ "--output", resultPath});
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/example/scala/WordCountITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/scala/WordCountITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/scala/WordCountITCase.java
new file mode 100644
index 0000000..d7a00b2
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/scala/WordCountITCase.java
@@ -0,0 +1,57 @@
+/*
+ * 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.flink.test.example.scala;
+
+import org.apache.flink.examples.scala.wordcount.WordCount;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
+/**
+ * Test {@link WordCount}.
+ */
+public class WordCountITCase extends JavaProgramTestBase {
+
+ protected String textPath;
+ protected String resultPath;
+
+ public WordCountITCase(){
+ setParallelism(4);
+ setNumTaskManagers(2);
+ setTaskManagerNumSlots(2);
+ }
+
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("text.txt", WordCountData.TEXT);
+ resultPath = getTempDirPath("result");
+ }
+
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ WordCount.main(new String[] {
+ "--input", textPath,
+ "--output", resultPath });
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/ConnectedComponentsITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/ConnectedComponentsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/ConnectedComponentsITCase.java
deleted file mode 100644
index d5b6b38..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/ConnectedComponentsITCase.java
+++ /dev/null
@@ -1,63 +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.flink.test.exampleJavaPrograms;
-
-import java.io.BufferedReader;
-
-import org.apache.flink.examples.java.graph.ConnectedComponents;
-import org.apache.flink.test.testdata.ConnectedComponentsData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-
-public class ConnectedComponentsITCase extends JavaProgramTestBase {
-
- private static final long SEED = 0xBADC0FFEEBEEFL;
-
- private static final int NUM_VERTICES = 1000;
-
- private static final int NUM_EDGES = 10000;
-
-
- private String verticesPath;
- private String edgesPath;
- private String resultPath;
-
-
- @Override
- protected void preSubmit() throws Exception {
- verticesPath = createTempFile("vertices.txt", ConnectedComponentsData.getEnumeratingVertices(NUM_VERTICES));
- edgesPath = createTempFile("edges.txt", ConnectedComponentsData.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED));
- resultPath = getTempFilePath("results");
- }
-
- @Override
- protected void testProgram() throws Exception {
- ConnectedComponents.main(
- "--vertices", verticesPath,
- "--edges", edgesPath,
- "--output", resultPath,
- "--iterations", "100");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- for (BufferedReader reader : getResultReader(resultPath)) {
- ConnectedComponentsData.checkOddEvenResult(reader);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/EnumTriangleBasicITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/EnumTriangleBasicITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/EnumTriangleBasicITCase.java
deleted file mode 100644
index 4d61b37..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/EnumTriangleBasicITCase.java
+++ /dev/null
@@ -1,48 +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.flink.test.exampleJavaPrograms;
-
-import org.apache.flink.examples.java.graph.EnumTriangles;
-import org.apache.flink.test.testdata.EnumTriangleData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-
-public class EnumTriangleBasicITCase extends JavaProgramTestBase {
-
- protected String edgePath;
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- edgePath = createTempFile("edges", EnumTriangleData.EDGES);
- resultPath = getTempDirPath("triangles");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(EnumTriangleData.TRIANGLES_BY_ID, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- EnumTriangles.main(new String[] {
- "--edges", edgePath,
- "--output", resultPath });
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/PageRankITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/PageRankITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/PageRankITCase.java
deleted file mode 100644
index efd534c..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/PageRankITCase.java
+++ /dev/null
@@ -1,92 +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.flink.test.exampleJavaPrograms;
-
-import java.io.File;
-
-import com.google.common.base.Charsets;
-import com.google.common.io.Files;
-
-import org.apache.flink.examples.java.graph.PageRank;
-import org.apache.flink.test.testdata.PageRankData;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
-
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-@RunWith(Parameterized.class)
-public class PageRankITCase extends MultipleProgramsTestBase {
-
- public PageRankITCase(TestExecutionMode mode){
- super(mode);
- }
-
- private String verticesPath;
- private String edgesPath;
- private String resultPath;
- private String expected;
-
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- resultPath = tempFolder.newFile().toURI().toString();
- File verticesFile = tempFolder.newFile();
- Files.write(PageRankData.VERTICES, verticesFile, Charsets.UTF_8);
-
- File edgesFile = tempFolder.newFile();
- Files.write(PageRankData.EDGES, edgesFile, Charsets.UTF_8);
-
- verticesPath = verticesFile.toURI().toString();
- edgesPath = edgesFile.toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareKeyValuePairsWithDelta(expected, resultPath, " ", 0.01);
- }
-
- @Test
- public void testPageRankSmallNumberOfIterations() throws Exception {
- PageRank.main(new String[]{
- "--pages", verticesPath,
- "--links", edgesPath,
- "--output", resultPath,
- "--numPages", PageRankData.NUM_VERTICES + "",
- "--iterations", "3"});
- expected = PageRankData.RANKS_AFTER_3_ITERATIONS;
- }
-
- @Test
- public void testPageRankWithConvergenceCriterion() throws Exception {
- PageRank.main(new String[]{
- "--pages", verticesPath,
- "--links", edgesPath,
- "--output", resultPath,
- "--numPages", PageRankData.NUM_VERTICES + "",
- "--vertices", "1000"});
- expected = PageRankData.RANKS_AFTER_EPSILON_0_0001_CONVERGENCE;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/TransitiveClosureITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/TransitiveClosureITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/TransitiveClosureITCase.java
deleted file mode 100644
index 123f616..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/TransitiveClosureITCase.java
+++ /dev/null
@@ -1,63 +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.flink.test.exampleJavaPrograms;
-
-
-import java.io.BufferedReader;
-
-import org.apache.flink.examples.java.graph.TransitiveClosureNaive;
-import org.apache.flink.test.testdata.ConnectedComponentsData;
-import org.apache.flink.test.testdata.TransitiveClosureData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-
-public class TransitiveClosureITCase extends JavaProgramTestBase {
-
- private static final long SEED = 0xBADC0FFEEBEEFL;
-
- private static final int NUM_VERTICES = 100;
-
- private static final int NUM_EDGES = 500;
-
- private String edgesPath;
- private String resultPath;
-
-
- @Override
- protected void preSubmit() throws Exception {
- edgesPath = createTempFile("edges.txt", ConnectedComponentsData.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED));
- resultPath = getTempFilePath("results");
- }
-
- @Override
- protected void testProgram() throws Exception {
- TransitiveClosureNaive.main(
- "--edges", edgesPath,
- "--output", resultPath,
- "--iterations", "5");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- for (BufferedReader reader : getResultReader(resultPath)) {
- TransitiveClosureData.checkOddEvenResult(reader);
- }
- }
-}
-
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WebLogAnalysisITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WebLogAnalysisITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WebLogAnalysisITCase.java
deleted file mode 100644
index b5ce8e4..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WebLogAnalysisITCase.java
+++ /dev/null
@@ -1,53 +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.flink.test.exampleJavaPrograms;
-
-
-import org.apache.flink.examples.java.relational.WebLogAnalysis;
-import org.apache.flink.test.testdata.WebLogAnalysisData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-
-public class WebLogAnalysisITCase extends JavaProgramTestBase {
-
- private String docsPath;
- private String ranksPath;
- private String visitsPath;
- private String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- docsPath = createTempFile("docs", WebLogAnalysisData.DOCS);
- ranksPath = createTempFile("ranks", WebLogAnalysisData.RANKS);
- visitsPath = createTempFile("visits", WebLogAnalysisData.VISITS);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(WebLogAnalysisData.EXCEPTED_RESULT, resultPath);
- }
- @Override
- protected void testProgram() throws Exception {
- WebLogAnalysis.main(new String[] {
- "--documents", docsPath,
- "--ranks", ranksPath,
- "--visits", visitsPath,
- "--output", resultPath});
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountITCase.java
deleted file mode 100644
index 3c47c5c..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountITCase.java
+++ /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.flink.test.exampleJavaPrograms;
-
-import org.apache.flink.examples.java.wordcount.WordCount;
-import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-
-public class WordCountITCase extends JavaProgramTestBase {
-
- protected String textPath;
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- textPath = createTempFile("text.txt", WordCountData.TEXT);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- WordCount.main(new String[] {
- "--input", textPath,
- "--output", resultPath });
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountNestedPOJOITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountNestedPOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountNestedPOJOITCase.java
deleted file mode 100644
index f1e75fe..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountNestedPOJOITCase.java
+++ /dev/null
@@ -1,118 +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.flink.test.exampleJavaPrograms;
-
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.api.common.functions.ReduceFunction;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-import org.apache.flink.util.Collector;
-
-import java.io.Serializable;
-import java.util.Date;
-
-@SuppressWarnings("serial")
-public class WordCountNestedPOJOITCase extends JavaProgramTestBase implements Serializable {
- private static final long serialVersionUID = 1L;
- protected String textPath;
- protected String resultPath;
-
-
- @Override
- protected void preSubmit() throws Exception {
- textPath = createTempFile("text.txt", WordCountData.TEXT);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- DataSet<String> text = env.readTextFile(textPath);
-
- DataSet<WC> counts = text
- .flatMap(new Tokenizer())
- .groupBy("complex.someTest")
- .reduce(new ReduceFunction<WC>() {
- private static final long serialVersionUID = 1L;
- public WC reduce(WC value1, WC value2) {
- return new WC(value1.complex.someTest, value1.count + value2.count);
- }
- });
-
- counts.writeAsText(resultPath);
-
- env.execute("WordCount with custom data types example");
- }
-
- public static final class Tokenizer implements FlatMapFunction<String, WC> {
-
- @Override
- public void flatMap(String value, Collector<WC> out) {
- // normalize and split the line
- String[] tokens = value.toLowerCase().split("\\W+");
-
- // emit the pairs
- for (String token : tokens) {
- if (token.length() > 0) {
- out.collect(new WC(token, 1));
- }
- }
- }
- }
-
- public static class WC { // is a pojo
- public ComplexNestedClass complex; // is a pojo
- public int count; // is a BasicType
-
- public WC() {
- }
- public WC(String t, int c) {
- this.count = c;
- this.complex = new ComplexNestedClass();
- this.complex.word = new Tuple3<Long, Long, String>(0L, 0L, "egal");
- this.complex.date = new Date();
- this.complex.someFloat = 0.0f;
- this.complex.someNumber = 666;
- this.complex.someTest = t;
- }
- @Override
- public String toString() {
- return this.complex.someTest+" "+count;
- }
- }
-
- public static class ComplexNestedClass { // pojo
- public static int ignoreStaticField;
- public transient int ignoreTransientField;
- public Date date; // generic type
- public Integer someNumber; // BasicType
- public float someFloat; // BasicType
- public Tuple3<Long, Long, String> word; //Tuple Type with three basic types
- public String someTest;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSimplePOJOITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSimplePOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSimplePOJOITCase.java
deleted file mode 100644
index 7d20597..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSimplePOJOITCase.java
+++ /dev/null
@@ -1,102 +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.flink.test.exampleJavaPrograms;
-
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.api.common.functions.ReduceFunction;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-import org.apache.flink.util.Collector;
-
-import java.io.Serializable;
-
-
-public class WordCountSimplePOJOITCase extends JavaProgramTestBase implements Serializable {
- private static final long serialVersionUID = 1L;
- protected String textPath;
- protected String resultPath;
-
-
- @Override
- protected void preSubmit() throws Exception {
- textPath = createTempFile("text.txt", WordCountData.TEXT);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- DataSet<String> text = env.readTextFile(textPath);
-
- DataSet<WC> counts = text
- .flatMap(new Tokenizer())
- .groupBy("word")
- .reduce(new ReduceFunction<WC>() {
- private static final long serialVersionUID = 1L;
-
- public WC reduce(WC value1, WC value2) {
- return new WC(value1.word, value1.count + value2.count);
- }
- });
-
- counts.writeAsText(resultPath);
-
- env.execute("WordCount with custom data types example");
- }
-
- public static final class Tokenizer implements FlatMapFunction<String, WC> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void flatMap(String value, Collector<WC> out) {
- // normalize and split the line
- String[] tokens = value.toLowerCase().split("\\W+");
-
- // emit the pairs
- for (String token : tokens) {
- if (token.length() > 0) {
- out.collect(new WC(token, 1));
- }
- }
- }
- }
-
- public static class WC {
- public WC() {}
- public WC(String w, int c) {
- word = w;
- count = c;
- }
- public String word;
- public int count;
- @Override
- public String toString() {
- return word + " " + count;
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSubclassInterfacePOJOITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSubclassInterfacePOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSubclassInterfacePOJOITCase.java
deleted file mode 100644
index 05ffc88..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSubclassInterfacePOJOITCase.java
+++ /dev/null
@@ -1,152 +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.flink.test.exampleJavaPrograms;
-
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.common.functions.ReduceFunction;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-import org.apache.flink.util.Collector;
-
-import java.io.Serializable;
-
-@SuppressWarnings("serial")
-public class WordCountSubclassInterfacePOJOITCase extends JavaProgramTestBase implements Serializable {
- private static final long serialVersionUID = 1L;
- protected String textPath;
- protected String resultPath;
-
-
- @Override
- protected void preSubmit() throws Exception {
- textPath = createTempFile("text.txt", WordCountData.TEXT);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- DataSet<String> text = env.readTextFile(textPath);
-
- DataSet<WCBase> counts = text
- .flatMap(new Tokenizer())
- .groupBy("word")
- .reduce(new ReduceFunction<WCBase>() {
- private static final long serialVersionUID = 1L;
- public WCBase reduce(WCBase value1, WCBase value2) {
- WC wc1 = (WC) value1;
- WC wc2 = (WC) value2;
- int c = wc1.secretCount.getCount() + wc2.secretCount.getCount();
- wc1.secretCount.setCount(c);
- return wc1;
- }
- })
- .map(new MapFunction<WCBase, WCBase>() {
- @Override
- public WCBase map(WCBase value) throws Exception {
- WC wc = (WC) value;
- wc.count = wc.secretCount.getCount();
- return wc;
- }
- });
-
- counts.writeAsText(resultPath);
-
- env.execute("WordCount with custom data types example");
- }
-
- public static final class Tokenizer implements FlatMapFunction<String, WCBase> {
-
- @Override
- public void flatMap(String value, Collector<WCBase> out) {
- // normalize and split the line
- String[] tokens = value.toLowerCase().split("\\W+");
- // emit the pairs
- for (String token : tokens) {
- if (token.length() > 0) {
- out.collect(new WC(token, 1));
- }
- }
- }
- }
-
- public static abstract class WCBase {
- public String word;
- public int count;
-
- public WCBase(String w, int c) {
- this.word = w;
- this.count = c;
- }
- @Override
- public String toString() {
- return word+" "+count;
- }
- }
-
- public static interface CrazyCounter {
- public int getCount();
- public void setCount(int c);
- }
-
- public static class CrazyCounterImpl implements CrazyCounter {
- public int countz;
-
- public CrazyCounterImpl() {
- }
-
- public CrazyCounterImpl(int c) {
- this.countz = c;
- }
-
- @Override
- public int getCount() {
- return countz;
- }
-
- @Override
- public void setCount(int c) {
- this.countz = c;
- }
-
- }
-
- public static class WC extends WCBase {
- public CrazyCounter secretCount;
-
- public WC() {
- super(null, 0);
- }
-
- public WC(String w, int c) {
- super(w, 0);
- this.secretCount = new CrazyCounterImpl(c);
- }
-
- }
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSubclassPOJOITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSubclassPOJOITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSubclassPOJOITCase.java
deleted file mode 100644
index f74ee16..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountSubclassPOJOITCase.java
+++ /dev/null
@@ -1,123 +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.flink.test.exampleJavaPrograms;
-
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.common.functions.ReduceFunction;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-import org.apache.flink.util.Collector;
-
-import java.io.Serializable;
-
-@SuppressWarnings("serial")
-public class WordCountSubclassPOJOITCase extends JavaProgramTestBase implements Serializable {
- private static final long serialVersionUID = 1L;
- protected String textPath;
- protected String resultPath;
-
-
- @Override
- protected void preSubmit() throws Exception {
- textPath = createTempFile("text.txt", WordCountData.TEXT);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- DataSet<String> text = env.readTextFile(textPath);
-
- DataSet<WCBase> counts = text
- .flatMap(new Tokenizer())
- .groupBy("word")
- .reduce(new ReduceFunction<WCBase>() {
- private static final long serialVersionUID = 1L;
- public WCBase reduce(WCBase value1, WCBase value2) {
- WC wc1 = (WC) value1;
- WC wc2 = (WC) value2;
- return new WC(value1.word, wc1.secretCount + wc2.secretCount);
- }
- })
- .map(new MapFunction<WCBase, WCBase>() {
- @Override
- public WCBase map(WCBase value) throws Exception {
- WC wc = (WC) value;
- wc.count = wc.secretCount;
- return wc;
- }
- });
-
- counts.writeAsText(resultPath);
-
- env.execute("WordCount with custom data types example");
- }
-
- public static final class Tokenizer implements FlatMapFunction<String, WCBase> {
-
- @Override
- public void flatMap(String value, Collector<WCBase> out) {
- // normalize and split the line
- String[] tokens = value.toLowerCase().split("\\W+");
- // emit the pairs
- for (String token : tokens) {
- if (token.length() > 0) {
- out.collect(new WC(token, 1));
- }
- }
- }
- }
-
- public static abstract class WCBase {
- public String word;
- public int count;
-
- public WCBase(String w, int c) {
- this.word = w;
- this.count = c;
- }
- @Override
- public String toString() {
- return word+" "+count;
- }
- }
-
- public static class WC extends WCBase {
-
- public int secretCount;
-
- public WC() {
- super(null, 0);
- }
-
- public WC(String w, int c) {
- super(w, 0);
- this.secretCount = c;
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountWithCollectionITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountWithCollectionITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountWithCollectionITCase.java
deleted file mode 100644
index d4404da..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleJavaPrograms/WordCountWithCollectionITCase.java
+++ /dev/null
@@ -1,66 +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.flink.test.exampleJavaPrograms;
-
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
-import org.apache.flink.api.java.aggregation.Aggregations;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.examples.java.wordcount.WordCount;
-import org.apache.flink.test.testdata.WordCountData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-import org.junit.Assert;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-public class WordCountWithCollectionITCase extends JavaProgramTestBase {
-
- private final List<Tuple2<String, Integer>> resultsCollected = new ArrayList<Tuple2<String, Integer>>();
-
- @Override
- protected void postSubmit() throws Exception {
- String[] result = new String[resultsCollected.size()];
- for (int i = 0; i < result.length; i++) {
- result[i] = resultsCollected.get(i).toString();
- }
- Arrays.sort(result);
-
- String[] expected = WordCountData.COUNTS_AS_TUPLES.split("\n");
- Arrays.sort(expected);
-
- Assert.assertEquals("Different number of lines in expected and obtained result.", expected.length, result.length);
- Assert.assertArrayEquals(expected, result);
- }
-
-
- @Override
- protected void testProgram() throws Exception {
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- DataSet<String> text = env.fromElements(WordCountData.TEXT);
- DataSet<Tuple2<String, Integer>> words = text.flatMap(new WordCount.Tokenizer());
- DataSet<Tuple2<String, Integer>> result = words.groupBy(0).aggregate(Aggregations.SUM, 1);
-
- result.output(new LocalCollectionOutputFormat<Tuple2<String, Integer>>(resultsCollected));
- env.execute("Word Count Collection");
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/ConnectedComponentsITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/ConnectedComponentsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/ConnectedComponentsITCase.java
deleted file mode 100644
index f3a1d03..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/ConnectedComponentsITCase.java
+++ /dev/null
@@ -1,64 +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.flink.test.exampleScalaPrograms;
-
-import org.apache.flink.examples.scala.graph.ConnectedComponents;
-import org.apache.flink.test.testdata.ConnectedComponentsData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-
-import java.io.BufferedReader;
-
-public class ConnectedComponentsITCase extends JavaProgramTestBase {
-
- private static final long SEED = 0xBADC0FFEEBEEFL;
-
- private static final int NUM_VERTICES = 1000;
-
- private static final int NUM_EDGES = 10000;
-
-
- private String verticesPath;
- private String edgesPath;
- private String resultPath;
-
-
- @Override
- protected void preSubmit() throws Exception {
- verticesPath = createTempFile("vertices.txt", ConnectedComponentsData.getEnumeratingVertices(NUM_VERTICES));
- edgesPath = createTempFile("edges.txt", ConnectedComponentsData.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED));
- resultPath = getTempFilePath("results");
- }
-
- @Override
- protected void testProgram() throws Exception {
- ConnectedComponents.main(new String[] {
- "--vertices", verticesPath,
- "--edges", edgesPath,
- "--output", resultPath,
- "--iterations", "100"});
- }
-
- @Override
- protected void postSubmit() throws Exception {
- for (BufferedReader reader : getResultReader(resultPath)) {
- ConnectedComponentsData.checkOddEvenResult(reader);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/EnumTriangleITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/EnumTriangleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/EnumTriangleITCase.java
deleted file mode 100644
index 9d5ffdf..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/EnumTriangleITCase.java
+++ /dev/null
@@ -1,48 +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.flink.test.exampleScalaPrograms;
-
-import org.apache.flink.examples.scala.graph.EnumTriangles;
-import org.apache.flink.test.testdata.EnumTriangleData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-
-public class EnumTriangleITCase extends JavaProgramTestBase {
-
- protected String edgePath;
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- edgePath = createTempFile("edges", EnumTriangleData.EDGES);
- resultPath = getTempDirPath("triangles");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(EnumTriangleData.TRIANGLES_BY_ID, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- EnumTriangles.main(new String[] {
- "--edges", edgePath,
- "--output", resultPath });
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/PageRankITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/PageRankITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/PageRankITCase.java
deleted file mode 100644
index e3e8936..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/PageRankITCase.java
+++ /dev/null
@@ -1,95 +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.flink.test.exampleScalaPrograms;
-
-import java.io.File;
-
-import com.google.common.base.Charsets;
-import com.google.common.io.Files;
-import org.apache.flink.examples.scala.graph.PageRankBasic;
-import org.apache.flink.test.testdata.PageRankData;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-@RunWith(Parameterized.class)
-public class PageRankITCase extends MultipleProgramsTestBase {
-
- public PageRankITCase(TestExecutionMode mode){
- super(mode);
- }
-
- private String verticesPath;
- private String edgesPath;
- private String resultPath;
- private String expected;
-
- @Rule
- public TemporaryFolder tempFolder = new TemporaryFolder();
-
- @Before
- public void before() throws Exception{
- File resultFile = tempFolder.newFile();
- //Delete file because the Scala API does not respect WriteMode set by the configuration
- resultFile.delete();
- resultPath = resultFile.toURI().toString();
-
- File verticesFile = tempFolder.newFile();
- Files.write(PageRankData.VERTICES, verticesFile, Charsets.UTF_8);
-
- File edgesFile = tempFolder.newFile();
- Files.write(PageRankData.EDGES, edgesFile, Charsets.UTF_8);
-
- verticesPath = verticesFile.toURI().toString();
- edgesPath = edgesFile.toURI().toString();
- }
-
- @After
- public void after() throws Exception{
- compareKeyValuePairsWithDelta(expected, resultPath, " ", 0.01);
- }
-
- @Test
- public void testPageRankWithSmallNumberOfIterations() throws Exception {
- PageRankBasic.main(new String[] {
- "--pages", verticesPath,
- "--links", edgesPath,
- "--output", resultPath,
- "--numPages", PageRankData.NUM_VERTICES+"",
- "--iterations", "3"});
- expected = PageRankData.RANKS_AFTER_3_ITERATIONS;
- }
-
- @Test
- public void testPageRankWithConvergence() throws Exception {
- // start with a very high number of iteration such that the dynamic convergence criterion must handle termination
- PageRankBasic.main(new String[] {
- "--pages", verticesPath,
- "--links", edgesPath,
- "--output", resultPath,
- "--numPages", PageRankData.NUM_VERTICES+"",
- "--iterations", "1000"});
- expected = PageRankData.RANKS_AFTER_EPSILON_0_0001_CONVERGENCE;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9bd491e0/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/TransitiveClosureITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/TransitiveClosureITCase.java b/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/TransitiveClosureITCase.java
deleted file mode 100644
index 9804f4c..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/exampleScalaPrograms/TransitiveClosureITCase.java
+++ /dev/null
@@ -1,63 +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.flink.test.exampleScalaPrograms;
-
-
-import java.io.BufferedReader;
-
-import org.apache.flink.examples.scala.graph.TransitiveClosureNaive;
-import org.apache.flink.test.testdata.ConnectedComponentsData;
-import org.apache.flink.test.testdata.TransitiveClosureData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-
-public class TransitiveClosureITCase extends JavaProgramTestBase {
-
- private static final long SEED = 0xBADC0FFEEBEEFL;
-
- private static final int NUM_VERTICES = 100;
-
- private static final int NUM_EDGES = 500;
-
- private String edgesPath;
- private String resultPath;
-
-
- @Override
- protected void preSubmit() throws Exception {
- edgesPath = createTempFile("edges.txt", ConnectedComponentsData.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED));
- resultPath = getTempFilePath("results");
- }
-
- @Override
- protected void testProgram() throws Exception {
- TransitiveClosureNaive.main(new String [] {
- "--edges", edgesPath,
- "--output", resultPath,
- "--iterations", "5"});
- }
-
- @Override
- protected void postSubmit() throws Exception {
- for (BufferedReader reader : getResultReader(resultPath)) {
- TransitiveClosureData.checkOddEvenResult(reader);
- }
- }
-}
-