You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2015/11/24 18:17:56 UTC
[13/16] flink git commit: [FLINK-2901] Remove Record API dependencies
from flink-tests #2
[FLINK-2901] Remove Record API dependencies from flink-tests #2
This closes #1306
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ebba20df
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ebba20df
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ebba20df
Branch: refs/heads/master
Commit: ebba20dfa87fda3719eccf5006fffa4c030caf2b
Parents: 8abae2c
Author: zentol <s....@web.de>
Authored: Tue Oct 27 21:22:45 2015 +0100
Committer: Fabian Hueske <fh...@apache.org>
Committed: Tue Nov 24 18:16:51 2015 +0100
----------------------------------------------------------------------
.../AccumulatorIterativeITCase.java | 77 +++
.../test/cancelling/CancelingTestBase.java | 156 +++++
.../test/cancelling/CancellingTestBase.java | 156 -----
.../test/cancelling/JoinCancelingITCase.java | 176 ++++++
.../test/cancelling/MapCancelingITCase.java | 2 +-
.../cancelling/MatchJoinCancelingITCase.java | 177 ------
.../IterationTerminationWithTwoTails.java | 74 +++
.../iterative/KMeansWithBroadcastSetITCase.java | 98 +++
.../recordJobs/graph/ComputeEdgeDegrees.java | 231 -------
.../graph/ConnectedComponentsWithCoGroup.java | 134 ----
.../graph/DeltaPageRankWithInitialDeltas.java | 187 ------
.../graph/EnumTrianglesOnEdgesWithDegrees.java | 186 ------
.../recordJobs/graph/EnumTrianglesRdfFoaf.java | 293 ---------
.../graph/EnumTrianglesWithDegrees.java | 104 ----
.../flink/test/recordJobs/graph/PairwiseSP.java | 435 -------------
.../graph/WorksetConnectedComponents.java | 198 ------
.../graph/triangleEnumUtil/EdgeInputFormat.java | 77 ---
.../EdgeWithDegreesInputFormat.java | 118 ----
.../EdgeWithDegreesOutputFormat.java | 61 --
.../triangleEnumUtil/TriangleOutputFormat.java | 57 --
.../test/recordJobs/kmeans/KMeansBroadcast.java | 324 ----------
.../test/recordJobs/kmeans/KMeansCross.java | 136 ----
.../recordJobs/kmeans/KMeansSingleStep.java | 300 ---------
.../recordJobs/kmeans/udfs/ComputeDistance.java | 65 --
.../udfs/ComputeDistanceParameterized.java | 77 ---
.../recordJobs/kmeans/udfs/CoordVector.java | 158 -----
.../kmeans/udfs/FindNearestCenter.java | 114 ----
.../recordJobs/kmeans/udfs/PointInFormat.java | 107 ----
.../recordJobs/kmeans/udfs/PointOutFormat.java | 78 ---
.../kmeans/udfs/RecomputeClusterCenter.java | 158 -----
.../recordJobs/relational/MergeOnlyJoin.java | 122 ----
.../test/recordJobs/relational/TPCHQuery1.java | 87 ---
.../test/recordJobs/relational/TPCHQuery10.java | 365 -----------
.../test/recordJobs/relational/TPCHQuery3.java | 277 ---------
.../relational/TPCHQuery3Unioned.java | 175 ------
.../test/recordJobs/relational/TPCHQuery4.java | 284 ---------
.../test/recordJobs/relational/TPCHQuery9.java | 251 --------
.../recordJobs/relational/TPCHQueryAsterix.java | 205 -------
.../recordJobs/relational/WebLogAnalysis.java | 341 ----------
.../query1Util/GroupByReturnFlag.java | 68 --
.../relational/query1Util/LineItemFilter.java | 89 ---
.../query1Util/LineItemFilterTest.java | 191 ------
.../relational/query9Util/AmountAggregate.java | 71 ---
.../query9Util/FilteredPartsJoin.java | 62 --
.../relational/query9Util/IntPair.java | 38 --
.../relational/query9Util/LineItemMap.java | 53 --
.../relational/query9Util/OrderMap.java | 49 --
.../relational/query9Util/OrderedPartsJoin.java | 58 --
.../relational/query9Util/PartFilter.java | 52 --
.../relational/query9Util/PartJoin.java | 58 --
.../relational/query9Util/PartListJoin.java | 55 --
.../relational/query9Util/PartsuppMap.java | 47 --
.../relational/query9Util/StringIntPair.java | 39 --
.../StringIntPairStringDataOutFormat.java | 54 --
.../relational/query9Util/SupplierMap.java | 55 --
.../relational/query9Util/SuppliersJoin.java | 57 --
.../test/recordJobs/sort/ReduceGroupSort.java | 114 ----
.../flink/test/recordJobs/sort/TeraSort.java | 74 ---
.../sort/tsUtil/TeraDistribution.java | 61 --
.../recordJobs/sort/tsUtil/TeraInputFormat.java | 49 --
.../test/recordJobs/sort/tsUtil/TeraKey.java | 157 -----
.../sort/tsUtil/TeraOutputFormat.java | 52 --
.../test/recordJobs/sort/tsUtil/TeraValue.java | 110 ----
.../recordJobs/util/DiscardingOutputFormat.java | 46 --
.../util/InfiniteIntegerInputFormat.java | 48 --
.../InfiniteIntegerInputFormatWithDelay.java | 54 --
.../recordJobs/util/IntTupleDataInFormat.java | 60 --
.../util/StringTupleDataOutFormat.java | 47 --
.../flink/test/recordJobs/util/Tuple.java | 615 -------------------
.../test/recordJobs/util/UniformIntInput.java | 93 ---
.../test/recordJobs/wordcount/WordCount.java | 166 -----
.../wordcount/WordCountAccumulators.java | 225 -------
72 files changed, 582 insertions(+), 9106 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java
new file mode 100644
index 0000000..6dc0a0b
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java
@@ -0,0 +1,77 @@
+/*
+ * 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.accumulators;
+
+import org.apache.flink.api.common.accumulators.IntCounter;
+import org.apache.flink.api.common.functions.RichGroupReduceFunction;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.api.java.operators.IterativeDataSet;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.Collector;
+import org.junit.Assert;
+
+public class AccumulatorIterativeITCase extends JavaProgramTestBase {
+ private static final int NUM_ITERATIONS = 3;
+ private static final int NUM_SUBTASKS = 1;
+ private static final String ACC_NAME = "test";
+
+ @Override
+ protected boolean skipCollectionExecution() {
+ return true;
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(NUM_SUBTASKS);
+
+ IterativeDataSet<Integer> iteration = env.fromElements(1, 2, 3).iterate(NUM_ITERATIONS);
+
+ iteration.closeWith(iteration.reduceGroup(new SumReducer())).output(new DiscardingOutputFormat());
+
+ Assert.assertEquals(Integer.valueOf(NUM_ITERATIONS * 6), (Integer)env.execute().getAccumulatorResult(ACC_NAME));
+ }
+
+ static final class SumReducer extends RichGroupReduceFunction<Integer, Integer> {
+
+ private static final long serialVersionUID = 1L;
+
+ private IntCounter testCounter = new IntCounter();
+
+ @Override
+ public void open(Configuration config) throws Exception {
+ getRuntimeContext().addAccumulator(ACC_NAME, this.testCounter);
+ }
+
+ @Override
+ public void reduce(Iterable<Integer> values, Collector<Integer> out) {
+ // Compute the sum
+ int sum = 0;
+
+ for (Integer value : values) {
+ sum += value;
+ testCounter.add(value);
+ }
+ out.collect(sum);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java
new file mode 100644
index 0000000..f6361ca
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java
@@ -0,0 +1,156 @@
+/*
+ * 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.cancelling;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.test.util.ForkableFlinkMiniCluster;
+import org.apache.flink.util.TestLogger;
+import org.junit.Assert;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.flink.api.common.Plan;
+import org.apache.flink.optimizer.DataStatistics;
+import org.apache.flink.optimizer.Optimizer;
+import org.apache.flink.optimizer.plan.OptimizedPlan;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobStatus;
+import static org.apache.flink.runtime.taskmanager.TaskCancelTest.awaitRunning;
+import static org.apache.flink.runtime.taskmanager.TaskCancelTest.cancelJob;
+import org.apache.flink.runtime.testutils.JobManagerActorTestUtils;
+import org.apache.flink.util.StringUtils;
+import org.apache.hadoop.fs.FileSystem;
+
+import org.junit.After;
+import org.junit.Before;
+import scala.concurrent.duration.FiniteDuration;
+
+/**
+ *
+ */
+public abstract class CancelingTestBase extends TestLogger {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CancelingTestBase.class);
+
+ private static final int MINIMUM_HEAP_SIZE_MB = 192;
+
+ /**
+ * Defines the number of seconds after which an issued cancel request is expected to have taken effect (i.e. the job
+ * is canceled), starting from the point in time when the cancel request is issued.
+ */
+ private static final int DEFAULT_CANCEL_FINISHED_INTERVAL = 10 * 1000;
+
+ private static final int DEFAULT_TASK_MANAGER_NUM_SLOTS = 1;
+
+ // --------------------------------------------------------------------------------------------
+
+ protected ForkableFlinkMiniCluster executor;
+
+ protected int taskManagerNumSlots = DEFAULT_TASK_MANAGER_NUM_SLOTS;
+
+ // --------------------------------------------------------------------------------------------
+
+ private void verifyJvmOptions() {
+ final long heap = Runtime.getRuntime().maxMemory() >> 20;
+ Assert.assertTrue("Insufficient java heap space " + heap + "mb - set JVM option: -Xmx" + MINIMUM_HEAP_SIZE_MB
+ + "m", heap > MINIMUM_HEAP_SIZE_MB - 50);
+ }
+
+ @Before
+ public void startCluster() throws Exception {
+ verifyJvmOptions();
+ Configuration config = new Configuration();
+ config.setBoolean(ConfigConstants.FILESYSTEM_DEFAULT_OVERWRITE_KEY, true);
+ config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 4);
+ config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
+ config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SEGMENT_SIZE_KEY, 4096);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY, 2048);
+
+ this.executor = new ForkableFlinkMiniCluster(config, false);
+ this.executor.start();
+ }
+
+ @After
+ public void stopCluster() throws Exception {
+ if (this.executor != null) {
+ this.executor.stop();
+ this.executor = null;
+ FileSystem.closeAll();
+ System.gc();
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public void runAndCancelJob(Plan plan, int msecsTillCanceling) throws Exception {
+ runAndCancelJob(plan, msecsTillCanceling, DEFAULT_CANCEL_FINISHED_INTERVAL);
+ }
+
+ public void runAndCancelJob(Plan plan, final int msecsTillCanceling, int maxTimeTillCanceled) throws Exception {
+ try {
+ // submit job
+ final JobGraph jobGraph = getJobGraph(plan);
+
+ executor.submitJobDetached(jobGraph);
+
+ // Wait for the job to make some progress and then cancel
+ awaitRunning(
+ executor.getLeaderGateway(TestingUtils.TESTING_DURATION()),
+ jobGraph.getJobID(),
+ TestingUtils.TESTING_DURATION());
+
+ Thread.sleep(msecsTillCanceling);
+
+ cancelJob(
+ executor.getLeaderGateway(TestingUtils.TESTING_DURATION()),
+ jobGraph.getJobID(),
+ new FiniteDuration(maxTimeTillCanceled, TimeUnit.MILLISECONDS));
+
+ // Wait for the job to be cancelled
+ JobManagerActorTestUtils.waitForJobStatus(jobGraph.getJobID(), JobStatus.CANCELED,
+ executor.getLeaderGateway(TestingUtils.TESTING_DURATION()),
+ TestingUtils.TESTING_DURATION());
+ } catch (Exception e) {
+ LOG.error("Exception found in runAndCancelJob.", e);
+ Assert.fail(StringUtils.stringifyException(e));
+ }
+
+ }
+
+ private JobGraph getJobGraph(final Plan plan) throws Exception {
+ final Optimizer pc = new Optimizer(new DataStatistics(), this.executor.configuration());
+ final OptimizedPlan op = pc.compile(plan);
+ final JobGraphGenerator jgg = new JobGraphGenerator();
+ return jgg.compileJobGraph(op);
+ }
+
+ public void setTaskManagerNumSlots(int taskManagerNumSlots) {
+ this.taskManagerNumSlots = taskManagerNumSlots;
+ }
+
+ public int getTaskManagerNumSlots() {
+ return this.taskManagerNumSlots;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
deleted file mode 100644
index 14c396d..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancellingTestBase.java
+++ /dev/null
@@ -1,156 +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.cancelling;
-
-import java.util.concurrent.TimeUnit;
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.apache.flink.test.util.ForkableFlinkMiniCluster;
-import org.apache.flink.util.TestLogger;
-import org.junit.Assert;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.optimizer.DataStatistics;
-import org.apache.flink.optimizer.Optimizer;
-import org.apache.flink.optimizer.plan.OptimizedPlan;
-import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
-import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.jobgraph.JobStatus;
-import static org.apache.flink.runtime.taskmanager.TaskCancelTest.awaitRunning;
-import static org.apache.flink.runtime.taskmanager.TaskCancelTest.cancelJob;
-import org.apache.flink.runtime.testutils.JobManagerActorTestUtils;
-import org.apache.flink.util.StringUtils;
-import org.apache.hadoop.fs.FileSystem;
-
-import org.junit.After;
-import org.junit.Before;
-import scala.concurrent.duration.FiniteDuration;
-
-/**
- *
- */
-public abstract class CancellingTestBase extends TestLogger {
-
- private static final Logger LOG = LoggerFactory.getLogger(CancellingTestBase.class);
-
- private static final int MINIMUM_HEAP_SIZE_MB = 192;
-
- /**
- * Defines the number of seconds after which an issued cancel request is expected to have taken effect (i.e. the job
- * is canceled), starting from the point in time when the cancel request is issued.
- */
- private static final int DEFAULT_CANCEL_FINISHED_INTERVAL = 10 * 1000;
-
- private static final int DEFAULT_TASK_MANAGER_NUM_SLOTS = 1;
-
- // --------------------------------------------------------------------------------------------
-
- protected ForkableFlinkMiniCluster executor;
-
- protected int taskManagerNumSlots = DEFAULT_TASK_MANAGER_NUM_SLOTS;
-
- // --------------------------------------------------------------------------------------------
-
- private void verifyJvmOptions() {
- final long heap = Runtime.getRuntime().maxMemory() >> 20;
- Assert.assertTrue("Insufficient java heap space " + heap + "mb - set JVM option: -Xmx" + MINIMUM_HEAP_SIZE_MB
- + "m", heap > MINIMUM_HEAP_SIZE_MB - 50);
- }
-
- @Before
- public void startCluster() throws Exception {
- verifyJvmOptions();
- Configuration config = new Configuration();
- config.setBoolean(ConfigConstants.FILESYSTEM_DEFAULT_OVERWRITE_KEY, true);
- config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 4);
- config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
- config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SEGMENT_SIZE_KEY, 4096);
- config.setInteger(ConfigConstants.TASK_MANAGER_NETWORK_NUM_BUFFERS_KEY, 2048);
-
- this.executor = new ForkableFlinkMiniCluster(config, false);
- this.executor.start();
- }
-
- @After
- public void stopCluster() throws Exception {
- if (this.executor != null) {
- this.executor.stop();
- this.executor = null;
- FileSystem.closeAll();
- System.gc();
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- public void runAndCancelJob(Plan plan, int msecsTillCanceling) throws Exception {
- runAndCancelJob(plan, msecsTillCanceling, DEFAULT_CANCEL_FINISHED_INTERVAL);
- }
-
- public void runAndCancelJob(Plan plan, final int msecsTillCanceling, int maxTimeTillCanceled) throws Exception {
- try {
- // submit job
- final JobGraph jobGraph = getJobGraph(plan);
-
- executor.submitJobDetached(jobGraph);
-
- // Wait for the job to make some progress and then cancel
- awaitRunning(
- executor.getLeaderGateway(TestingUtils.TESTING_DURATION()),
- jobGraph.getJobID(),
- TestingUtils.TESTING_DURATION());
-
- Thread.sleep(msecsTillCanceling);
-
- cancelJob(
- executor.getLeaderGateway(TestingUtils.TESTING_DURATION()),
- jobGraph.getJobID(),
- new FiniteDuration(maxTimeTillCanceled, TimeUnit.MILLISECONDS));
-
- // Wait for the job to be cancelled
- JobManagerActorTestUtils.waitForJobStatus(jobGraph.getJobID(), JobStatus.CANCELED,
- executor.getLeaderGateway(TestingUtils.TESTING_DURATION()),
- TestingUtils.TESTING_DURATION());
- } catch (Exception e) {
- LOG.error("Exception found in runAndCancelJob.", e);
- Assert.fail(StringUtils.stringifyException(e));
- }
-
- }
-
- private JobGraph getJobGraph(final Plan plan) throws Exception {
- final Optimizer pc = new Optimizer(new DataStatistics(), this.executor.configuration());
- final OptimizedPlan op = pc.compile(plan);
- final JobGraphGenerator jgg = new JobGraphGenerator();
- return jgg.compileJobGraph(op);
- }
-
- public void setTaskManagerNumSlots(int taskManagerNumSlots) {
- this.taskManagerNumSlots = taskManagerNumSlots;
- }
-
- public int getTaskManagerNumSlots() {
- return this.taskManagerNumSlots;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/cancelling/JoinCancelingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/JoinCancelingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/JoinCancelingITCase.java
new file mode 100644
index 0000000..d797e47
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/cancelling/JoinCancelingITCase.java
@@ -0,0 +1,176 @@
+/*
+ * 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.cancelling;
+
+import org.apache.flink.api.common.functions.JoinFunction;
+import org.apache.flink.api.common.functions.RichJoinFunction;
+import org.apache.flink.api.common.operators.base.JoinOperatorBase;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.operators.testutils.UniformIntTupleGenerator;
+import org.apache.flink.test.util.InfiniteIntegerTupleInputFormat;
+import org.apache.flink.test.util.UniformIntTupleGeneratorInputFormat;
+
+public class JoinCancelingITCase extends CancelingTestBase {
+ private static final int parallelism = 4;
+
+ public JoinCancelingITCase() {
+ setTaskManagerNumSlots(parallelism);
+ }
+
+ // --------------- Test Sort Matches that are canceled while still reading / sorting -----------------
+ private void executeTask(JoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> joiner, boolean slow) throws Exception {
+ executeTask(joiner, slow, parallelism);
+ }
+
+ private void executeTask(JoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> joiner, boolean slow, int parallelism) throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ DataSet<Tuple2<Integer, Integer>> input1 = env.createInput(new InfiniteIntegerTupleInputFormat(slow));
+ DataSet<Tuple2<Integer, Integer>> input2 = env.createInput(new InfiniteIntegerTupleInputFormat(slow));
+
+ input1.join(input2, JoinOperatorBase.JoinHint.REPARTITION_SORT_MERGE)
+ .where(0)
+ .equalTo(0)
+ .with(joiner)
+ .output(new DiscardingOutputFormat<Tuple2<Integer, Integer>>());
+
+ env.setParallelism(parallelism);
+
+ runAndCancelJob(env.createProgramPlan(), 5 * 1000, 10 * 1000);
+ }
+
+// @Test
+ public void testCancelSortMatchWhileReadingSlowInputs() throws Exception {
+ executeTask(new SimpleMatcher<Integer>(), true);
+ }
+
+// @Test
+ public void testCancelSortMatchWhileReadingFastInputs() throws Exception {
+ executeTask(new SimpleMatcher<Integer>(), false);
+ }
+
+// @Test
+ public void testCancelSortMatchPriorToFirstRecordReading() throws Exception {
+ executeTask(new StuckInOpenMatcher<Integer>(), false);
+ }
+
+ private void executeTaskWithGenerator(
+ JoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> joiner,
+ int keys, int vals, int msecsTillCanceling, int maxTimeTillCanceled) throws Exception {
+ UniformIntTupleGenerator g = new UniformIntTupleGenerator(keys, vals, false);
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ DataSet<Tuple2<Integer, Integer>> input1 = env.createInput(new UniformIntTupleGeneratorInputFormat(keys, vals));
+ DataSet<Tuple2<Integer, Integer>> input2 = env.createInput(new UniformIntTupleGeneratorInputFormat(keys, vals));
+
+ input1.join(input2, JoinOperatorBase.JoinHint.REPARTITION_SORT_MERGE)
+ .where(0)
+ .equalTo(0)
+ .with(joiner)
+ .output(new DiscardingOutputFormat<Tuple2<Integer, Integer>>());
+
+ env.setParallelism(parallelism);
+
+ runAndCancelJob(env.createProgramPlan(), msecsTillCanceling, maxTimeTillCanceled);
+ }
+
+// @Test
+ public void testCancelSortMatchWhileDoingHeavySorting() throws Exception {
+ executeTaskWithGenerator(new SimpleMatcher<Integer>(), 50000, 100, 30 * 1000, 30 * 1000);
+ }
+
+ // --------------- Test Sort Matches that are canceled while in the Matching Phase -----------------
+
+// @Test
+ public void testCancelSortMatchWhileJoining() throws Exception {
+ executeTaskWithGenerator(new DelayingMatcher<Integer>(), 500, 3, 10 * 1000, 20 * 1000);
+ }
+
+// @Test
+ public void testCancelSortMatchWithLongCancellingResponse() throws Exception {
+ executeTaskWithGenerator(new LongCancelTimeMatcher<Integer>(), 500, 3, 10 * 1000, 10 * 1000);
+ }
+
+ // -------------------------------------- Test System corner cases ---------------------------------
+
+// @Test
+ public void testCancelSortMatchWithHighparallelism() throws Exception {
+ executeTask(new SimpleMatcher<Integer>(), false, 64);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static final class SimpleMatcher<IN> implements JoinFunction<Tuple2<IN, IN>, Tuple2<IN, IN>, Tuple2<IN, IN>> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<IN, IN> join(Tuple2<IN, IN> first, Tuple2<IN, IN> second) throws Exception {
+ return new Tuple2<>(first.f0, second.f0);
+ }
+ }
+
+ public static final class DelayingMatcher<IN> implements JoinFunction<Tuple2<IN, IN>, Tuple2<IN, IN>, Tuple2<IN, IN>> {
+ private static final long serialVersionUID = 1L;
+
+ private static final int WAIT_TIME_PER_RECORD = 10 * 1000; // 10 sec.
+
+ @Override
+ public Tuple2<IN, IN> join(Tuple2<IN, IN> first, Tuple2<IN, IN> second) throws Exception {
+ Thread.sleep(WAIT_TIME_PER_RECORD);
+ return new Tuple2<>(first.f0, second.f0);
+ }
+ }
+
+ public static final class LongCancelTimeMatcher<IN> implements JoinFunction<Tuple2<IN, IN>, Tuple2<IN, IN>, Tuple2<IN, IN>> {
+ private static final long serialVersionUID = 1L;
+
+ private static final int WAIT_TIME_PER_RECORD = 5 * 1000; // 5 sec.
+
+ @Override
+ public Tuple2<IN, IN> join(Tuple2<IN, IN> first, Tuple2<IN, IN> second) throws Exception {
+ final long start = System.currentTimeMillis();
+ long remaining = WAIT_TIME_PER_RECORD;
+ do {
+ try {
+ Thread.sleep(remaining);
+ } catch (InterruptedException iex) {}
+ } while ((remaining = WAIT_TIME_PER_RECORD - System.currentTimeMillis() + start) > 0);
+ return new Tuple2<>(first.f0, second.f0);
+ }
+ }
+
+ public static final class StuckInOpenMatcher<IN> extends RichJoinFunction<Tuple2<IN, IN>, Tuple2<IN, IN>, Tuple2<IN, IN>> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ synchronized (this) {
+ wait();
+ }
+ }
+
+ @Override
+ public Tuple2<IN, IN> join(Tuple2<IN, IN> first, Tuple2<IN, IN> second) throws Exception {
+ return new Tuple2<>(first.f0, second.f0);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/cancelling/MapCancelingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/MapCancelingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/MapCancelingITCase.java
index 55b1911..03b1a24 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/cancelling/MapCancelingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/cancelling/MapCancelingITCase.java
@@ -26,7 +26,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.test.util.InfiniteIntegerInputFormat;
import org.junit.Test;
-public class MapCancelingITCase extends CancellingTestBase {
+public class MapCancelingITCase extends CancelingTestBase {
private static final int parallelism = 4;
public MapCancelingITCase() {
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/cancelling/MatchJoinCancelingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/MatchJoinCancelingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/MatchJoinCancelingITCase.java
deleted file mode 100644
index b6f82d4..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/cancelling/MatchJoinCancelingITCase.java
+++ /dev/null
@@ -1,177 +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.cancelling;
-
-import org.apache.flink.api.common.functions.JoinFunction;
-import org.apache.flink.api.common.functions.RichJoinFunction;
-import org.apache.flink.api.common.operators.base.JoinOperatorBase;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.io.DiscardingOutputFormat;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.operators.testutils.UniformIntTupleGenerator;
-import org.apache.flink.test.util.InfiniteIntegerTupleInputFormat;
-import org.apache.flink.test.util.UniformIntTupleGeneratorInputFormat;
-import org.junit.Test;
-
-public class MatchJoinCancelingITCase extends CancellingTestBase {
- private static final int parallelism = 4;
-
- public MatchJoinCancelingITCase() {
- setTaskManagerNumSlots(parallelism);
- }
-
- // --------------- Test Sort Matches that are canceled while still reading / sorting -----------------
- private void executeTask(JoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> joiner, boolean slow) throws Exception {
- executeTask(joiner, slow, parallelism);
- }
-
- private void executeTask(JoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> joiner, boolean slow, int parallelism) throws Exception {
- ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- DataSet<Tuple2<Integer, Integer>> input1 = env.createInput(new InfiniteIntegerTupleInputFormat(slow));
- DataSet<Tuple2<Integer, Integer>> input2 = env.createInput(new InfiniteIntegerTupleInputFormat(slow));
-
- input1.join(input2, JoinOperatorBase.JoinHint.REPARTITION_SORT_MERGE)
- .where(0)
- .equalTo(0)
- .with(joiner)
- .output(new DiscardingOutputFormat<Tuple2<Integer, Integer>>());
-
- env.setParallelism(parallelism);
-
- runAndCancelJob(env.createProgramPlan(), 5 * 1000, 10 * 1000);
- }
-
-// @Test
- public void testCancelSortMatchWhileReadingSlowInputs() throws Exception {
- executeTask(new SimpleMatcher<Integer>(), true);
- }
-
-// @Test
- public void testCancelSortMatchWhileReadingFastInputs() throws Exception {
- executeTask(new SimpleMatcher<Integer>(), false);
- }
-
-// @Test
- public void testCancelSortMatchPriorToFirstRecordReading() throws Exception {
- executeTask(new StuckInOpenMatcher<Integer>(), false);
- }
-
- private void executeTaskWithGenerator(
- JoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> joiner,
- int keys, int vals, int msecsTillCanceling, int maxTimeTillCanceled) throws Exception {
- UniformIntTupleGenerator g = new UniformIntTupleGenerator(keys, vals, false);
- ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- DataSet<Tuple2<Integer, Integer>> input1 = env.createInput(new UniformIntTupleGeneratorInputFormat(keys, vals));
- DataSet<Tuple2<Integer, Integer>> input2 = env.createInput(new UniformIntTupleGeneratorInputFormat(keys, vals));
-
- input1.join(input2, JoinOperatorBase.JoinHint.REPARTITION_SORT_MERGE)
- .where(0)
- .equalTo(0)
- .with(joiner)
- .output(new DiscardingOutputFormat<Tuple2<Integer, Integer>>());
-
- env.setParallelism(parallelism);
-
- runAndCancelJob(env.createProgramPlan(), msecsTillCanceling, maxTimeTillCanceled);
- }
-
-// @Test
- public void testCancelSortMatchWhileDoingHeavySorting() throws Exception {
- executeTaskWithGenerator(new SimpleMatcher<Integer>(), 50000, 100, 30 * 1000, 30 * 1000);
- }
-
- // --------------- Test Sort Matches that are canceled while in the Matching Phase -----------------
-
-// @Test
- public void testCancelSortMatchWhileJoining() throws Exception {
- executeTaskWithGenerator(new DelayingMatcher<Integer>(), 500, 3, 10 * 1000, 20 * 1000);
- }
-
-// @Test
- public void testCancelSortMatchWithLongCancellingResponse() throws Exception {
- executeTaskWithGenerator(new LongCancelTimeMatcher<Integer>(), 500, 3, 10 * 1000, 10 * 1000);
- }
-
- // -------------------------------------- Test System corner cases ---------------------------------
-
-// @Test
- public void testCancelSortMatchWithHighparallelism() throws Exception {
- executeTask(new SimpleMatcher<Integer>(), false, 64);
- }
-
- // --------------------------------------------------------------------------------------------
-
- public static final class SimpleMatcher<IN> implements JoinFunction<Tuple2<IN, IN>, Tuple2<IN, IN>, Tuple2<IN, IN>> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public Tuple2<IN, IN> join(Tuple2<IN, IN> first, Tuple2<IN, IN> second) throws Exception {
- return new Tuple2<>(first.f0, second.f0);
- }
- }
-
- public static final class DelayingMatcher<IN> implements JoinFunction<Tuple2<IN, IN>, Tuple2<IN, IN>, Tuple2<IN, IN>> {
- private static final long serialVersionUID = 1L;
-
- private static final int WAIT_TIME_PER_RECORD = 10 * 1000; // 10 sec.
-
- @Override
- public Tuple2<IN, IN> join(Tuple2<IN, IN> first, Tuple2<IN, IN> second) throws Exception {
- Thread.sleep(WAIT_TIME_PER_RECORD);
- return new Tuple2<>(first.f0, second.f0);
- }
- }
-
- public static final class LongCancelTimeMatcher<IN> implements JoinFunction<Tuple2<IN, IN>, Tuple2<IN, IN>, Tuple2<IN, IN>> {
- private static final long serialVersionUID = 1L;
-
- private static final int WAIT_TIME_PER_RECORD = 5 * 1000; // 5 sec.
-
- @Override
- public Tuple2<IN, IN> join(Tuple2<IN, IN> first, Tuple2<IN, IN> second) throws Exception {
- final long start = System.currentTimeMillis();
- long remaining = WAIT_TIME_PER_RECORD;
- do {
- try {
- Thread.sleep(remaining);
- } catch (InterruptedException iex) {}
- } while ((remaining = WAIT_TIME_PER_RECORD - System.currentTimeMillis() + start) > 0);
- return new Tuple2<>(first.f0, second.f0);
- }
- }
-
- public static final class StuckInOpenMatcher<IN> extends RichJoinFunction<Tuple2<IN, IN>, Tuple2<IN, IN>, Tuple2<IN, IN>> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void open(Configuration parameters) throws Exception {
- synchronized (this) {
- wait();
- }
- }
-
- @Override
- public Tuple2<IN, IN> join(Tuple2<IN, IN> first, Tuple2<IN, IN> second) throws Exception {
- return new Tuple2<>(first.f0, second.f0);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/iterative/IterationTerminationWithTwoTails.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/IterationTerminationWithTwoTails.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/IterationTerminationWithTwoTails.java
new file mode 100644
index 0000000..5a2df3f
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/IterationTerminationWithTwoTails.java
@@ -0,0 +1,74 @@
+/*
+ * 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.iterative;
+
+import java.util.List;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.IterativeDataSet;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import static org.apache.flink.test.util.TestBaseUtils.containsResultAsText;
+import org.apache.flink.util.Collector;
+
+public class IterationTerminationWithTwoTails extends JavaProgramTestBase {
+ private static final String EXPECTED = "22\n";
+
+ @Override
+ protected void testProgram() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(4);
+
+ DataSet<String> initialInput = env.fromElements("1", "2", "3", "4", "5").name("input");
+
+ IterativeDataSet<String> iteration = initialInput.iterate(5).name("Loop");
+
+ DataSet<String> sumReduce = iteration.reduceGroup(new SumReducer()).name("Compute sum (GroupReduce");
+
+ DataSet<String> terminationFilter = iteration.filter(new TerminationFilter()).name("Compute termination criterion (Map)");
+
+ List<String> result = iteration.closeWith(sumReduce, terminationFilter).collect();
+
+ containsResultAsText(result, EXPECTED);
+ }
+
+ public static final class SumReducer implements GroupReduceFunction<String, String> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void reduce(Iterable<String> values, Collector<String> out) throws Exception {
+ int sum = 0;
+ for (String value : values) {
+ sum += Integer.parseInt(value) + 1;
+ }
+ out.collect("" + sum);
+ }
+ }
+
+ public static class TerminationFilter implements FilterFunction<String> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean filter(String value) throws Exception {
+ return Integer.parseInt(value) < 21;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/iterative/KMeansWithBroadcastSetITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/KMeansWithBroadcastSetITCase.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/KMeansWithBroadcastSetITCase.java
new file mode 100644
index 0000000..8edc2aa
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/KMeansWithBroadcastSetITCase.java
@@ -0,0 +1,98 @@
+/*
+ * 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.iterative;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.IterativeDataSet;
+import org.apache.flink.examples.java.clustering.KMeans;
+import org.apache.flink.examples.java.clustering.KMeans.Point;
+import org.apache.flink.examples.java.clustering.KMeans.Centroid;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.test.testdata.KMeansData;
+
+import java.util.List;
+
+public class KMeansWithBroadcastSetITCase extends JavaProgramTestBase {
+
+ @SuppressWarnings("serial")
+ @Override
+ protected void testProgram() throws Exception {
+
+ String[] points = KMeansData.DATAPOINTS_2D.split("\n");
+ String[] centers = KMeansData.INITIAL_CENTERS_2D.split("\n");
+
+ // set up execution environment
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ // get input data
+ DataSet<Point> pointsSet = env.fromElements(points)
+ .map(new MapFunction<String, Point>() {
+ public Point map(String p) {
+ String[] fields = p.split("\\|");
+ return new Point(
+ Double.parseDouble(fields[1]),
+ Double.parseDouble(fields[2]));
+ }
+ });
+
+ DataSet <Centroid> centroidsSet = env.fromElements(centers)
+ .map(new MapFunction<String, Centroid>() {
+ public Centroid map(String c) {
+ String[] fields = c.split("\\|");
+ return new Centroid(
+ Integer.parseInt(fields[0]),
+ Double.parseDouble(fields[1]),
+ Double.parseDouble(fields[2]));
+ }
+ });
+
+ // set number of bulk iterations for KMeans algorithm
+ IterativeDataSet<Centroid> loop = centroidsSet.iterate(20);
+
+ DataSet<Centroid> newCentroids = pointsSet
+ // compute closest centroid for each point
+ .map(new KMeans.SelectNearestCenter()).withBroadcastSet(loop, "centroids")
+ // count and sum point coordinates for each centroid
+ .map(new KMeans.CountAppender())
+ .groupBy(0).reduce(new KMeans.CentroidAccumulator())
+ // compute new centroids from point counts and coordinate sums
+ .map(new KMeans.CentroidAverager());
+
+ // feed new centroids back into next iteration
+ DataSet<Centroid> finalCentroids = loop.closeWith(newCentroids);
+
+ DataSet<String> stringCentroids = finalCentroids
+ .map(new MapFunction<Centroid, String>() {
+ @Override
+ public String map(Centroid c) throws Exception {
+ return String.format("%d|%.2f|%.2f|", c.id, c.x, c.y);
+ }
+ });
+
+ List<String> result = stringCentroids.collect();
+
+ KMeansData.checkResultsWithDelta(
+ KMeansData.CENTERS_2D_AFTER_20_ITERATIONS_DOUBLE_DIGIT,
+ result,
+ 0.01);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/ComputeEdgeDegrees.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/ComputeEdgeDegrees.java b/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/ComputeEdgeDegrees.java
deleted file mode 100644
index 9ec8c73..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/ComputeEdgeDegrees.java
+++ /dev/null
@@ -1,231 +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.recordJobs.graph;
-
-import java.io.Serializable;
-import java.util.Arrays;
-import java.util.Iterator;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.Program;
-import org.apache.flink.api.common.ProgramDescription;
-import org.apache.flink.api.java.record.functions.MapFunction;
-import org.apache.flink.api.java.record.functions.ReduceFunction;
-import org.apache.flink.api.java.record.operators.FileDataSink;
-import org.apache.flink.api.java.record.operators.FileDataSource;
-import org.apache.flink.api.java.record.operators.MapOperator;
-import org.apache.flink.api.java.record.operators.ReduceOperator;
-import org.apache.flink.test.recordJobs.graph.triangleEnumUtil.EdgeInputFormat;
-import org.apache.flink.test.recordJobs.graph.triangleEnumUtil.EdgeWithDegreesOutputFormat;
-import org.apache.flink.types.IntValue;
-import org.apache.flink.types.Record;
-import org.apache.flink.util.Collector;
-
-@SuppressWarnings("deprecation")
-public class ComputeEdgeDegrees implements Program, ProgramDescription {
-
- private static final long serialVersionUID = 1L;
-
- // --------------------------------------------------------------------------------------------
- // Vertex Degree Computation
- // --------------------------------------------------------------------------------------------
-
- /**
- * Duplicates each edge such that: (u,v) becomes (u,v),(v,u)
- */
- public static final class ProjectEdge extends MapFunction implements Serializable {
- private static final long serialVersionUID = 1L;
-
- private final Record copy = new Record();
-
- @Override
- public void map(Record record, Collector<Record> out) throws Exception {
- this.copy.setField(0, record.getField(1, IntValue.class));
- this.copy.setField(1, record.getField(0, IntValue.class));
-
- out.collect(this.copy);
- out.collect(record);
- }
- }
-
- /**
- * Creates for all records in the group a record of the form (v1, v2, c1, c2), where
- * v1 is the lexicographically smaller vertex id and the count for the vertex that
- * was the key contains the number of edges associated with it. The other count is zero.
- * This reducer also eliminates duplicate edges.
- */
- public static final class CountEdges extends ReduceFunction implements Serializable {
- private static final long serialVersionUID = 1L;
-
- private final Record result = new Record();
-
- private final IntValue firstVertex = new IntValue();
- private final IntValue secondVertex = new IntValue();
- private final IntValue firstCount = new IntValue();
- private final IntValue secondCount = new IntValue();
-
- private int[] vals = new int[1024];
-
- @Override
- public void reduce(Iterator<Record> records, Collector<Record> out) throws Exception {
- int[] vals = this.vals;
- int len = 0;
- int key = -1;
-
- // collect all values
- while (records.hasNext()) {
- final Record rec = records.next();
- final int id = rec.getField(1, IntValue.class).getValue();
- if (key == -1) {
- key = rec.getField(0, IntValue.class).getValue();
- }
-
- if (len >= vals.length) {
- vals = new int[vals.length * 2];
- System.arraycopy(this.vals, 0, vals, 0, this.vals.length);
- this.vals = vals;
- }
- vals[len++] = id;
- }
-
- // sort the values to and uniquify them
- Arrays.sort(vals, 0, len);
- int k = 0;
- for (int curr = -1, i = 0; i < len; i++) {
- int val = vals[i];
- if (val != curr) {
- curr = val;
- vals[k] = vals[i];
- k++;
- }
- else {
- vals[k] = vals[i];
- }
- }
- len = k;
-
- // create such that the vertex with the lower id is always the first
- // both vertices contain a count, which is zero for the non-key vertices
- for (int i = 0; i < len; i++) {
- final int e2 = vals[i];
- if (key <= e2) {
- firstVertex.setValue(key);
- secondVertex.setValue(e2);
- firstCount.setValue(len);
- secondCount.setValue(0);
- } else {
- firstVertex.setValue(e2);
- secondVertex.setValue(key);
- firstCount.setValue(0);
- secondCount.setValue(len);
- }
- this.result.setField(0, firstVertex);
- this.result.setField(1, secondVertex);
- this.result.setField(2, firstCount);
- this.result.setField(3, secondCount);
- out.collect(result);
- }
- }
- }
-
- /**
- * Takes the two separate edge entries (v1, v2, c1, 0) and (v1, v2, 0, c2)
- * and creates an entry (v1, v2, c1, c2).
- */
- public static final class JoinCountsAndUniquify extends ReduceFunction implements Serializable {
- private static final long serialVersionUID = 1L;
-
- private final IntValue count1 = new IntValue();
- private final IntValue count2 = new IntValue();
-
- @Override
- public void reduce(Iterator<Record> records, Collector<Record> out) throws Exception {
- Record rec = null;
- int c1 = 0, c2 = 0;
- int numValues = 0;
-
- while (records.hasNext()) {
- rec = records.next();
- final int f1 = rec.getField(2, IntValue.class).getValue();
- final int f2 = rec.getField(3, IntValue.class).getValue();
- c1 += f1;
- c2 += f2;
- numValues++;
- }
-
- if (numValues != 2 || c1 == 0 || c2 == 0) {
- throw new RuntimeException("JoinCountsAndUniquify Problem: key1=" +
- rec.getField(0, IntValue.class).getValue() + ", key2=" +
- rec.getField(1, IntValue.class).getValue() +
- "values=" + numValues + ", c1=" + c1 + ", c2=" + c2);
- }
-
- count1.setValue(c1);
- count2.setValue(c2);
- rec.setField(2, count1);
- rec.setField(3, count2);
- out.collect(rec);
- }
- }
-
- // --------------------------------------------------------------------------------------------
- // Triangle Enumeration
- // --------------------------------------------------------------------------------------------
-
- /**
- * Assembles the Plan of the triangle enumeration example Pact program.
- */
- @Override
- public Plan getPlan(String... args) {
- // parse job parameters
- final int numSubTasks = (args.length > 0 ? Integer.parseInt(args[0]) : 1);
- final String edgeInput = args.length > 1 ? args[1] : "";
- final String output = args.length > 2 ? args[2] : "";
- final char delimiter = args.length > 3 ? (char) Integer.parseInt(args[3]) : ',';
-
-
- FileDataSource edges = new FileDataSource(new EdgeInputFormat(), edgeInput, "Input Edges");
- edges.setParameter(EdgeInputFormat.ID_DELIMITER_CHAR, delimiter);
-
- MapOperator projectEdge = MapOperator.builder(new ProjectEdge())
- .input(edges).name("Project Edge").build();
-
- ReduceOperator edgeCounter = ReduceOperator.builder(new CountEdges(), IntValue.class, 0)
- .input(projectEdge).name("Count Edges for Vertex").build();
-
- ReduceOperator countJoiner = ReduceOperator.builder(new JoinCountsAndUniquify())
- .keyField(IntValue.class, 0)
- .keyField(IntValue.class, 1)
- .input(edgeCounter)
- .name("Join Counts")
- .build();
-
- FileDataSink triangles = new FileDataSink(new EdgeWithDegreesOutputFormat(), output, countJoiner, "Unique Edges With Degrees");
-
- Plan p = new Plan(triangles, "Normalize Edges and compute Vertex Degrees");
- p.setDefaultParallelism(numSubTasks);
- return p;
- }
-
- @Override
- public String getDescription() {
- return "Parameters: [noSubStasks] [input file] [output file] [vertex delimiter]";
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/ConnectedComponentsWithCoGroup.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/ConnectedComponentsWithCoGroup.java b/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/ConnectedComponentsWithCoGroup.java
deleted file mode 100644
index cc6fd01..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/ConnectedComponentsWithCoGroup.java
+++ /dev/null
@@ -1,134 +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.recordJobs.graph;
-
-import java.io.Serializable;
-import java.util.Iterator;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.Program;
-import org.apache.flink.api.java.record.functions.CoGroupFunction;
-import org.apache.flink.api.java.record.functions.FunctionAnnotation.ConstantFieldsFirst;
-import org.apache.flink.api.java.record.functions.FunctionAnnotation.ConstantFieldsSecond;
-import org.apache.flink.api.java.record.io.CsvInputFormat;
-import org.apache.flink.api.java.record.io.CsvOutputFormat;
-import org.apache.flink.api.java.record.operators.CoGroupOperator;
-import org.apache.flink.api.java.record.operators.DeltaIteration;
-import org.apache.flink.api.java.record.operators.FileDataSink;
-import org.apache.flink.api.java.record.operators.FileDataSource;
-import org.apache.flink.api.java.record.operators.JoinOperator;
-import org.apache.flink.api.java.record.operators.MapOperator;
-import org.apache.flink.test.recordJobs.graph.WorksetConnectedComponents.DuplicateLongMap;
-import org.apache.flink.test.recordJobs.graph.WorksetConnectedComponents.NeighborWithComponentIDJoin;
-import org.apache.flink.types.LongValue;
-import org.apache.flink.types.Record;
-import org.apache.flink.util.Collector;
-
-@SuppressWarnings("deprecation")
-public class ConnectedComponentsWithCoGroup implements Program {
-
- private static final long serialVersionUID = 1L;
-
- @ConstantFieldsFirst(0)
- @ConstantFieldsSecond(0)
- public static final class MinIdAndUpdate extends CoGroupFunction implements Serializable {
- private static final long serialVersionUID = 1L;
-
- private final LongValue newComponentId = new LongValue();
-
- @Override
- public void coGroup(Iterator<Record> candidates, Iterator<Record> current, Collector<Record> out) throws Exception {
- if (!current.hasNext()) {
- throw new Exception("Error: Id not encountered before.");
- }
- Record old = current.next();
- long oldId = old.getField(1, LongValue.class).getValue();
-
- long minimumComponentID = Long.MAX_VALUE;
-
- while (candidates.hasNext()) {
- long candidateComponentID = candidates.next().getField(1, LongValue.class).getValue();
- if (candidateComponentID < minimumComponentID) {
- minimumComponentID = candidateComponentID;
- }
- }
-
- if (minimumComponentID < oldId) {
- newComponentId.setValue(minimumComponentID);
- old.setField(1, newComponentId);
- out.collect(old);
- }
- }
-
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public Plan getPlan(String... args) {
- // parse job parameters
- final int numSubTasks = (args.length > 0 ? Integer.parseInt(args[0]) : 1);
- final String verticesInput = (args.length > 1 ? args[1] : "");
- final String edgeInput = (args.length > 2 ? args[2] : "");
- final String output = (args.length > 3 ? args[3] : "");
- final int maxIterations = (args.length > 4 ? Integer.parseInt(args[4]) : 1);
-
- // data source for initial vertices
- FileDataSource initialVertices = new FileDataSource(new CsvInputFormat(' ', LongValue.class), verticesInput, "Vertices");
-
- MapOperator verticesWithId = MapOperator.builder(DuplicateLongMap.class).input(initialVertices).name("Assign Vertex Ids").build();
-
- DeltaIteration iteration = new DeltaIteration(0, "Connected Components Iteration");
- iteration.setInitialSolutionSet(verticesWithId);
- iteration.setInitialWorkset(verticesWithId);
- iteration.setMaximumNumberOfIterations(maxIterations);
-
- // create DataSourceContract for the edges
- FileDataSource edges = new FileDataSource(new CsvInputFormat(' ', LongValue.class, LongValue.class), edgeInput, "Edges");
-
- // create CrossOperator for distance computation
- JoinOperator joinWithNeighbors = JoinOperator.builder(new NeighborWithComponentIDJoin(), LongValue.class, 0, 0)
- .input1(iteration.getWorkset())
- .input2(edges)
- .name("Join Candidate Id With Neighbor")
- .build();
-
- CoGroupOperator minAndUpdate = CoGroupOperator.builder(new MinIdAndUpdate(), LongValue.class, 0, 0)
- .input1(joinWithNeighbors)
- .input2(iteration.getSolutionSet())
- .name("Min Id and Update")
- .build();
-
- iteration.setNextWorkset(minAndUpdate);
- iteration.setSolutionSetDelta(minAndUpdate);
-
- // create DataSinkContract for writing the new cluster positions
- FileDataSink result = new FileDataSink(new CsvOutputFormat(), output, iteration, "Result");
- CsvOutputFormat.configureRecordFormat(result)
- .recordDelimiter('\n')
- .fieldDelimiter(' ')
- .field(LongValue.class, 0)
- .field(LongValue.class, 1);
-
- // return the PACT plan
- Plan plan = new Plan(result, "Workset Connected Components");
- plan.setDefaultParallelism(numSubTasks);
- return plan;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/DeltaPageRankWithInitialDeltas.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/DeltaPageRankWithInitialDeltas.java b/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/DeltaPageRankWithInitialDeltas.java
deleted file mode 100644
index 81adbf5..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/DeltaPageRankWithInitialDeltas.java
+++ /dev/null
@@ -1,187 +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.recordJobs.graph;
-
-import java.util.Iterator;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.Program;
-import org.apache.flink.api.common.ProgramDescription;
-import org.apache.flink.api.java.record.functions.JoinFunction;
-import org.apache.flink.api.java.record.functions.ReduceFunction;
-import org.apache.flink.api.java.record.functions.FunctionAnnotation.ConstantFields;
-import org.apache.flink.api.java.record.functions.FunctionAnnotation.ConstantFieldsSecond;
-import org.apache.flink.api.java.record.io.CsvInputFormat;
-import org.apache.flink.api.java.record.io.CsvOutputFormat;
-import org.apache.flink.api.java.record.operators.DeltaIteration;
-import org.apache.flink.api.java.record.operators.FileDataSink;
-import org.apache.flink.api.java.record.operators.FileDataSource;
-import org.apache.flink.api.java.record.operators.JoinOperator;
-import org.apache.flink.api.java.record.operators.ReduceOperator;
-import org.apache.flink.api.java.record.operators.ReduceOperator.Combinable;
-import org.apache.flink.types.DoubleValue;
-import org.apache.flink.types.LongValue;
-import org.apache.flink.types.Record;
-import org.apache.flink.util.Collector;
-
-@SuppressWarnings("deprecation")
-public class DeltaPageRankWithInitialDeltas implements Program, ProgramDescription {
-
- private static final long serialVersionUID = 1L;
-
-
- @ConstantFieldsSecond(0)
- public static final class RankComparisonMatch extends JoinFunction {
-
- private static final long serialVersionUID = 1L;
-
- private final DoubleValue newRank = new DoubleValue();
-
- @Override
- public void join(Record vertexWithDelta, Record vertexWithOldRank, Collector<Record> out) throws Exception {
- DoubleValue deltaVal = vertexWithDelta.getField(1, DoubleValue.class);
- DoubleValue currentVal = vertexWithOldRank.getField(1, DoubleValue.class);
-
- newRank.setValue(deltaVal.getValue() + currentVal.getValue());
- vertexWithOldRank.setField(1, newRank);
-
- out.collect(vertexWithOldRank);
- }
- }
-
- @Combinable
- @ConstantFields(0)
- public static final class UpdateRankReduceDelta extends ReduceFunction {
-
- private static final long serialVersionUID = 1L;
-
- private final DoubleValue newRank = new DoubleValue();
-
- @Override
- public void reduce(Iterator<Record> records, Collector<Record> out) {
-
- double rankSum = 0.0;
- double rank;
- Record rec = null;
-
- while (records.hasNext()) {
- rec = records.next();
- rank = rec.getField(1, DoubleValue.class).getValue();
- rankSum += rank;
- }
-
- // ignore small deltas
- if (Math.abs(rankSum) > 0.00001) {
- newRank.setValue(rankSum);
- rec.setField(1, newRank);
- out.collect(rec);
- }
- }
- }
-
- public class PRDependenciesComputationMatchDelta extends JoinFunction {
-
- private static final long serialVersionUID = 1L;
-
- private final Record result = new Record();
- private final DoubleValue partRank = new DoubleValue();
-
- /*
- * (srcId, trgId, weight) x (vId, rank) => (trgId, rank / weight)
- */
- @Override
- public void join(Record vertexWithRank, Record edgeWithWeight, Collector<Record> out) throws Exception {
- result.setField(0, edgeWithWeight.getField(1, LongValue.class));
- final long outLinks = edgeWithWeight.getField(2, LongValue.class).getValue();
- final double rank = vertexWithRank.getField(1, DoubleValue.class).getValue();
- partRank.setValue(rank / (double) outLinks);
- result.setField(1, partRank);
-
- out.collect(result);
- }
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public Plan getPlan(String... args) {
- // parse job parameters
- final int numSubTasks = (args.length > 0 ? Integer.parseInt(args[0]) : 1);
- final String solutionSetInput = (args.length > 1 ? args[1] : "");
- final String deltasInput = (args.length > 2 ? args[2] : "");
- final String dependencySetInput = (args.length > 3 ? args[3] : "");
- final String output = (args.length > 4 ? args[4] : "");
- final int maxIterations = (args.length > 5 ? Integer.parseInt(args[5]) : 1);
-
- // create DataSourceContract for the initalSolutionSet
- FileDataSource initialSolutionSet = new FileDataSource(new CsvInputFormat(' ', LongValue.class, DoubleValue.class), solutionSetInput, "Initial Solution Set");
-
- // create DataSourceContract for the initalDeltaSet
- FileDataSource initialDeltaSet = new FileDataSource(new CsvInputFormat(' ', LongValue.class, DoubleValue.class), deltasInput, "Initial DeltaSet");
-
- // create DataSourceContract for the edges
- FileDataSource dependencySet = new FileDataSource(new CsvInputFormat(' ', LongValue.class, LongValue.class, LongValue.class), dependencySetInput, "Dependency Set");
-
- DeltaIteration iteration = new DeltaIteration(0, "Delta PageRank");
- iteration.setInitialSolutionSet(initialSolutionSet);
- iteration.setInitialWorkset(initialDeltaSet);
- iteration.setMaximumNumberOfIterations(maxIterations);
-
- JoinOperator dependenciesMatch = JoinOperator.builder(PRDependenciesComputationMatchDelta.class,
- LongValue.class, 0, 0)
- .input1(iteration.getWorkset())
- .input2(dependencySet)
- .name("calculate dependencies")
- .build();
-
- ReduceOperator updateRanks = ReduceOperator.builder(UpdateRankReduceDelta.class, LongValue.class, 0)
- .input(dependenciesMatch)
- .name("update ranks")
- .build();
-
- JoinOperator oldRankComparison = JoinOperator.builder(RankComparisonMatch.class, LongValue.class, 0, 0)
- .input1(updateRanks)
- .input2(iteration.getSolutionSet())
- .name("comparison with old ranks")
- .build();
-
- iteration.setNextWorkset(updateRanks);
- iteration.setSolutionSetDelta(oldRankComparison);
-
- // create DataSinkContract for writing the final ranks
- FileDataSink result = new FileDataSink(CsvOutputFormat.class, output, iteration, "Final Ranks");
- CsvOutputFormat.configureRecordFormat(result)
- .recordDelimiter('\n')
- .fieldDelimiter(' ')
- .field(LongValue.class, 0)
- .field(DoubleValue.class, 1);
-
- // return the PACT plan
- Plan plan = new Plan(result, "Delta PageRank");
- plan.setDefaultParallelism(numSubTasks);
- return plan;
-
- }
-
-
- @Override
- public String getDescription() {
- return "Parameters: <numberOfSubTasks> <initialSolutionSet(pageId, rank)> <deltas(pageId, delta)> <dependencySet(srcId, trgId, out_links)> <out> <maxIterations>";
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/EnumTrianglesOnEdgesWithDegrees.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/EnumTrianglesOnEdgesWithDegrees.java b/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/EnumTrianglesOnEdgesWithDegrees.java
deleted file mode 100644
index b2328ba..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/EnumTrianglesOnEdgesWithDegrees.java
+++ /dev/null
@@ -1,186 +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.recordJobs.graph;
-
-import java.io.Serializable;
-import java.util.Iterator;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.Program;
-import org.apache.flink.api.common.ProgramDescription;
-import org.apache.flink.api.java.record.functions.JoinFunction;
-import org.apache.flink.api.java.record.functions.MapFunction;
-import org.apache.flink.api.java.record.functions.ReduceFunction;
-import org.apache.flink.api.java.record.operators.FileDataSink;
-import org.apache.flink.api.java.record.operators.FileDataSource;
-import org.apache.flink.api.java.record.operators.JoinOperator;
-import org.apache.flink.api.java.record.operators.MapOperator;
-import org.apache.flink.api.java.record.operators.ReduceOperator;
-import org.apache.flink.test.recordJobs.graph.triangleEnumUtil.EdgeWithDegreesInputFormat;
-import org.apache.flink.test.recordJobs.graph.triangleEnumUtil.TriangleOutputFormat;
-import org.apache.flink.types.IntValue;
-import org.apache.flink.types.Record;
-import org.apache.flink.util.Collector;
-
-/**
- * An implementation of the triangle enumeration, which expects its input to
- * encode the degrees of the vertices. The algorithm selects the lower-degree vertex for the
- * enumeration of open triads.
- */
-@SuppressWarnings("deprecation")
-public class EnumTrianglesOnEdgesWithDegrees implements Program, ProgramDescription {
-
- private static final long serialVersionUID = 1L;
-
- // --------------------------------------------------------------------------------------------
- // Triangle Enumeration
- // --------------------------------------------------------------------------------------------
-
- public static final class ProjectOutCounts extends MapFunction implements Serializable {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void map(Record record, Collector<Record> out) throws Exception {
- record.setNumFields(2);
- out.collect(record);
- }
- }
-
- public static final class ProjectToLowerDegreeVertex extends MapFunction implements Serializable {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void map(Record record, Collector<Record> out) throws Exception {
- final int d1 = record.getField(2, IntValue.class).getValue();
- final int d2 = record.getField(3, IntValue.class).getValue();
- if (d1 > d2) {
- IntValue first = record.getField(1, IntValue.class);
- IntValue second = record.getField(0, IntValue.class);
- record.setField(0, first);
- record.setField(1, second);
- }
- record.setNumFields(2);
- out.collect(record);
- }
- }
-
- public static final class BuildTriads extends ReduceFunction implements Serializable {
- private static final long serialVersionUID = 1L;
-
- private final IntValue firstVertex = new IntValue();
- private final IntValue secondVertex = new IntValue();
-
- private int[] edgeCache = new int[1024];
-
- @Override
- public void reduce(Iterator<Record> records, Collector<Record> out) throws Exception {
- int len = 0;
-
- Record rec = null;
- while (records.hasNext()) {
- rec = records.next();
- final int e1 = rec.getField(1, IntValue.class).getValue();
-
- for (int i = 0; i < len; i++) {
- final int e2 = this.edgeCache[i];
-
- if (e1 <= e2) {
- firstVertex.setValue(e1);
- secondVertex.setValue(e2);
- } else {
- firstVertex.setValue(e2);
- secondVertex.setValue(e1);
- }
-
- rec.setField(1, firstVertex);
- rec.setField(2, secondVertex);
- out.collect(rec);
- }
-
- if (len >= this.edgeCache.length) {
- int[] na = new int[this.edgeCache.length * 2];
- System.arraycopy(this.edgeCache, 0, na, 0, this.edgeCache.length);
- this.edgeCache = na;
- }
- this.edgeCache[len++] = e1;
- }
- }
- }
-
- public static class CloseTriads extends JoinFunction implements Serializable {
- private static final long serialVersionUID = 1L;
- @Override
- public void join(Record triangle, Record missingEdge, Collector<Record> out) throws Exception {
- out.collect(triangle);
- }
- }
-
- /**
- * Assembles the Plan of the triangle enumeration example Pact program.
- */
- @Override
- public Plan getPlan(String... args) {
-
- // parse job parameters
- int numSubTasks = args.length > 0 ? Integer.parseInt(args[0]) : 1;
- String edgeInput = args.length > 1 ? args[1] : "";
- String output = args.length > 2 ? args[2] : "";
-
- FileDataSource edges = new FileDataSource(new EdgeWithDegreesInputFormat(), edgeInput, "Input Edges with Degrees");
- edges.setParameter(EdgeWithDegreesInputFormat.VERTEX_DELIMITER_CHAR, '|');
- edges.setParameter(EdgeWithDegreesInputFormat.DEGREE_DELIMITER_CHAR, ',');
-
- // =========================== Triangle Enumeration ============================
-
- MapOperator toLowerDegreeEdge = MapOperator.builder(new ProjectToLowerDegreeVertex())
- .input(edges)
- .name("Select lower-degree Edge")
- .build();
-
- MapOperator projectOutCounts = MapOperator.builder(new ProjectOutCounts())
- .input(edges)
- .name("Project to vertex Ids only")
- .build();
-
- ReduceOperator buildTriads = ReduceOperator.builder(new BuildTriads(), IntValue.class, 0)
- .input(toLowerDegreeEdge)
- .name("Build Triads")
- .build();
-
- JoinOperator closeTriads = JoinOperator.builder(new CloseTriads(), IntValue.class, 1, 0)
- .keyField(IntValue.class, 2, 1)
- .input1(buildTriads)
- .input2(projectOutCounts)
- .name("Close Triads")
- .build();
- closeTriads.setParameter("INPUT_SHIP_STRATEGY", "SHIP_REPARTITION_HASH");
- closeTriads.setParameter("LOCAL_STRATEGY", "LOCAL_STRATEGY_HASH_BUILD_SECOND");
-
- FileDataSink triangles = new FileDataSink(new TriangleOutputFormat(), output, closeTriads, "Triangles");
-
- Plan p = new Plan(triangles, "Enumerate Triangles");
- p.setDefaultParallelism(numSubTasks);
- return p;
- }
-
- @Override
- public String getDescription() {
- return "Parameters: [noSubStasks] [input file] [output file]";
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/ebba20df/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/EnumTrianglesRdfFoaf.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/EnumTrianglesRdfFoaf.java b/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/EnumTrianglesRdfFoaf.java
deleted file mode 100644
index ba8b54a..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/recordJobs/graph/EnumTrianglesRdfFoaf.java
+++ /dev/null
@@ -1,293 +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.recordJobs.graph;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Iterator;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.Program;
-import org.apache.flink.api.common.ProgramDescription;
-import org.apache.flink.api.java.record.functions.JoinFunction;
-import org.apache.flink.api.java.record.functions.ReduceFunction;
-import org.apache.flink.api.java.record.functions.FunctionAnnotation.ConstantFields;
-import org.apache.flink.api.java.record.functions.FunctionAnnotation.ConstantFieldsFirstExcept;
-import org.apache.flink.api.java.record.io.CsvOutputFormat;
-import org.apache.flink.api.java.record.io.DelimitedInputFormat;
-import org.apache.flink.api.java.record.operators.FileDataSink;
-import org.apache.flink.api.java.record.operators.FileDataSource;
-import org.apache.flink.api.java.record.operators.JoinOperator;
-import org.apache.flink.api.java.record.operators.ReduceOperator;
-import org.apache.flink.types.Record;
-import org.apache.flink.types.StringValue;
-import org.apache.flink.util.Collector;
-
-/**
- * Implementation of the triangle enumeration example Pact program.
- * The program expects a file with RDF triples (in XML serialization) as input. Triples must be separated by linebrakes.
- *
- * The program filters for foaf:knows predicates to identify relationships between two entities (typically persons).
- * Relationships are interpreted as edges in a social graph. Then the program enumerates all triangles which are build
- * by edges in that graph.
- *
- * Usually, triangle enumeration is used as a pre-processing step to identify highly connected subgraphs.
- * The algorithm was published as MapReduce job by J. Cohen in "Graph Twiddling in a MapReduce World".
- * The Pact version was described in "MapReduce and PACT - Comparing Data Parallel Programming Models" (BTW 2011).
- */
-@SuppressWarnings("deprecation")
-public class EnumTrianglesRdfFoaf implements Program, ProgramDescription {
-
- private static final long serialVersionUID = 1L;
-
- /**
- * Reads RDF triples and filters on the foaf:knows RDF predicate.
- * The foaf:knows RDF predicate indicates that the RDF subject and object (typically of type foaf:person) know each
- * other.
- * Therefore, knowing connections between people are extracted and handles as graph edges.
- * The EdgeInFormat filters all rdf triples with foaf:knows predicates. The subjects and objects URLs are
- * compared.
- * The lexicographically smaller URL is set as the first field of the output record, the greater one as the second field.
- */
- public static class EdgeInFormat extends DelimitedInputFormat {
- private static final long serialVersionUID = 1L;
-
- private final StringValue rdfSubj = new StringValue();
- private final StringValue rdfPred = new StringValue();
- private final StringValue rdfObj = new StringValue();
-
- @Override
- public Record readRecord(Record target, byte[] bytes, int offset, int numBytes) {
- final int limit = offset + numBytes;
- int startPos = offset;
-
- // read RDF subject
- startPos = parseVarLengthEncapsulatedStringField(bytes, startPos, limit, ' ', rdfSubj, '"');
- if (startPos < 0) {
- // invalid record, exit
- return null;
- }
- // read RDF predicate
- startPos = parseVarLengthEncapsulatedStringField(bytes, startPos, limit, ' ', rdfPred, '"');
- if (startPos < 0 || !rdfPred.getValue().equals("<http://xmlns.com/foaf/0.1/knows>")) {
- // invalid record or predicate is not a foaf-knows predicate, exit
- return null;
- }
- // read RDF object
- startPos = parseVarLengthEncapsulatedStringField(bytes, startPos, limit, ' ', rdfObj, '"');
- if (startPos < 0) {
- // invalid record, exit
- return null;
- }
-
- // compare RDF subject and object
- if (rdfSubj.compareTo(rdfObj) <= 0) {
- // subject is smaller, subject becomes first attribute, object second
- target.setField(0, rdfSubj);
- target.setField(1, rdfObj);
- } else {
- // object is smaller, object becomes first attribute, subject second
- target.setField(0, rdfObj);
- target.setField(1, rdfSubj);
- }
-
- return target;
- }
-
- /*
- * Utility method to efficiently parse encapsulated, variable length strings
- */
- private int parseVarLengthEncapsulatedStringField(byte[] bytes, int startPos, int limit, char delim, StringValue field, char encaps) {
-
- boolean isEncaps = false;
-
- // check whether string is encapsulated
- if (bytes[startPos] == encaps) {
- isEncaps = true;
- }
-
- if (isEncaps) {
- // string is encapsulated
- for (int i = startPos; i < limit; i++) {
- if (bytes[i] == encaps) {
- if (bytes[i+1] == delim) {
- field.setValueAscii(bytes, startPos, i-startPos+1);
- return i+2;
- }
- }
- }
- return -1;
- } else {
- // string is not encapsulated
- int i;
- for (i = startPos; i < limit; i++) {
- if (bytes[i] == delim) {
- field.setValueAscii(bytes, startPos, i-startPos);
- return i+1;
- }
- }
- if (i == limit) {
- field.setValueAscii(bytes, startPos, i-startPos);
- return i+1;
- } else {
- return -1;
- }
- }
- }
- }
-
- /**
- * Builds triads (open triangle) from all two edges that share a vertex.
- * The common vertex is
- */
- @ConstantFields(0)
- public static class BuildTriads extends ReduceFunction implements Serializable {
- private static final long serialVersionUID = 1L;
-
- // list of non-matching vertices
- private final ArrayList<StringValue> otherVertices = new ArrayList<StringValue>(32);
-
- // matching vertex
- private final StringValue matchVertex = new StringValue();
-
- // mutable output record
- private final Record result = new Record();
-
- // initialize list of non-matching vertices for one vertex
- public BuildTriads() {
- this.otherVertices.add(new StringValue());
- }
-
- @Override
- public void reduce(Iterator<Record> records, Collector<Record> out) throws Exception {
- // read the first edge
- final Record rec = records.next();
- // read the matching vertex
- rec.getFieldInto(0, this.matchVertex);
- // read the non-matching vertex and add it to the list
- rec.getFieldInto(1, this.otherVertices.get(0));
-
- // set the matching vertex in the output record
- this.result.setField(0, this.matchVertex);
-
- int numEdges = 1;
- // while there are more edges
- while (records.hasNext()) {
-
- // read the next edge
- final Record next = records.next();
-
- final StringValue myVertex;
- // obtain an object to store the non-matching vertex
- if (numEdges >= this.otherVertices.size()) {
- // we need an additional vertex object
- // create the object
- myVertex = new StringValue();
- // and put it in the list
- this.otherVertices.add(myVertex);
- } else {
- // we reuse a previously created object from the list
- myVertex = this.otherVertices.get(numEdges);
- }
- // read the non-matching vertex into the obtained object
- next.getFieldInto(1, myVertex);
-
- // combine the current edge with all vertices in the non-matching vertex list
- for (int i = 0; i < numEdges; i++) {
- // get the other non-matching vertex
- final StringValue otherVertex = this.otherVertices.get(i);
- // add my and other vertex to the output record depending on their ordering
- if (otherVertex.compareTo(myVertex) < 0) {
- this.result.setField(1, otherVertex);
- this.result.setField(2, myVertex);
- out.collect(this.result);
- } else {
- next.setField(2, otherVertex);
- out.collect(next);
- }
- }
-
- numEdges++;
- }
- }
- }
-
- /**
- * Matches all missing edges with existing edges from input.
- * If the missing edge for a triad is found, the triad is transformed to a triangle by adding the missing edge.
- */
- @ConstantFieldsFirstExcept({})
- public static class CloseTriads extends JoinFunction implements Serializable {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void join(Record triad, Record missingEdge, Collector<Record> out) throws Exception {
- // emit triangle (already contains missing edge at field 0
- out.collect(triad);
- }
- }
-
- /**
- * Assembles the Plan of the triangle enumeration example Pact program.
- */
- @Override
- public Plan getPlan(String... args) {
-
- // parse job parameters
- int numSubTasks = (args.length > 0 ? Integer.parseInt(args[0]) : 1);
- String edgeInput = (args.length > 1 ? args[1] : "");
- String output = (args.length > 2 ? args[2] : "");
-
- FileDataSource edges = new FileDataSource(new EdgeInFormat(), edgeInput, "BTC Edges");
-
- ReduceOperator buildTriads = ReduceOperator.builder(new BuildTriads(), StringValue.class, 0)
- .name("Build Triads")
- .build();
-
- JoinOperator closeTriads = JoinOperator.builder(new CloseTriads(), StringValue.class, 1, 0)
- .keyField(StringValue.class, 2, 1)
- .name("Close Triads")
- .build();
- closeTriads.setParameter("INPUT_LEFT_SHIP_STRATEGY", "SHIP_REPARTITION_HASH");
- closeTriads.setParameter("INPUT_RIGHT_SHIP_STRATEGY", "SHIP_REPARTITION_HASH");
- closeTriads.setParameter("LOCAL_STRATEGY", "LOCAL_STRATEGY_HASH_BUILD_SECOND");
-
- FileDataSink triangles = new FileDataSink(new CsvOutputFormat(), output, "Output");
- CsvOutputFormat.configureRecordFormat(triangles)
- .recordDelimiter('\n')
- .fieldDelimiter(' ')
- .field(StringValue.class, 0)
- .field(StringValue.class, 1)
- .field(StringValue.class, 2);
-
- triangles.setInput(closeTriads);
- closeTriads.setSecondInput(edges);
- closeTriads.setFirstInput(buildTriads);
- buildTriads.setInput(edges);
-
- Plan plan = new Plan(triangles, "Enumerate Triangles");
- plan.setDefaultParallelism(numSubTasks);
- return plan;
- }
-
- @Override
- public String getDescription() {
- return "Parameters: [numSubStasks] [inputRDFTriples] [outputTriangles]";
- }
-}