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:59 UTC
[16/16] flink git commit: [FLINK-2901] Remove Record API dependencies
from flink-tests #1
[FLINK-2901] Remove Record API dependencies from flink-tests #1
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8abae2c2
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8abae2c2
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8abae2c2
Branch: refs/heads/master
Commit: 8abae2c2f2a0a7b6a39880b90467b5209e65117d
Parents: 348d501
Author: zentol <s....@web.de>
Authored: Tue Oct 27 21:22:38 2015 +0100
Committer: Fabian Hueske <fh...@apache.org>
Committed: Tue Nov 24 18:16:51 2015 +0100
----------------------------------------------------------------------
.../runtime/taskmanager/TaskCancelTest.java | 2 +-
.../AccumulatorIterativeITCase.java | 142 ----
.../broadcastvars/BroadcastBranchingITCase.java | 142 ++--
.../test/cancelling/CancellingTestBase.java | 86 +--
.../test/cancelling/MapCancelingITCase.java | 140 ++--
.../cancelling/MatchJoinCancelingITCase.java | 240 ++----
.../clients/examples/LocalExecutorITCase.java | 34 +-
.../distributedCache/DistributedCacheTest.java | 154 ++--
.../test/failingPrograms/TaskFailureITCase.java | 179 +----
.../CoGroupConnectedComponentsITCase.java | 153 ++--
.../test/iterative/DeltaPageRankITCase.java | 108 ---
...IterationTerminationWithTerminationTail.java | 120 +--
.../IterationTerminationWithTwoTails.java | 134 ----
.../IterationWithAllReducerITCase.java | 99 +--
.../iterative/IterationWithChainingITCase.java | 114 +--
.../iterative/IterationWithUnionITCase.java | 25 +-
.../test/iterative/IterativeKMeansITCase.java | 62 --
.../flink/test/iterative/KMeansITCase.java | 62 --
.../examples/KMeansSingleStepTest.java | 179 ++++-
.../examples/RelationalQueryCompilerTest.java | 154 +++-
.../ConnectedComponentsCoGroupTest.java | 84 ++-
.../iterations/IterativeKMeansTest.java | 159 ----
.../jsonplan/DumpCompiledPlanTest.java | 84 ++-
.../optimizer/jsonplan/PreviewPlanDumpTest.java | 116 ++-
.../runtime/NetworkStackThroughputITCase.java | 32 +-
.../util/tests/IntTupleDataInFormatTest.java | 63 --
.../test/testPrograms/util/tests/TupleTest.java | 751 -------------------
.../test/util/ContractITCaseIOFormats.java | 85 ---
.../org/apache/flink/test/util/CoordVector.java | 158 ++++
.../apache/flink/test/util/FailingTestBase.java | 172 -----
.../test/util/InfiniteIntegerInputFormat.java | 48 ++
.../util/InfiniteIntegerTupleInputFormat.java | 49 ++
.../apache/flink/test/util/PointFormatter.java | 58 ++
.../apache/flink/test/util/PointInFormat.java | 96 +++
.../UniformIntTupleGeneratorInputFormat.java | 54 ++
35 files changed, 1477 insertions(+), 2861 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
index cd40c82..571152b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
@@ -173,7 +173,7 @@ public class TaskCancelTest {
}
}
- private void awaitRunning(ActorGateway jobManager, JobID jobId, FiniteDuration timeout)
+ public static void awaitRunning(ActorGateway jobManager, JobID jobId, FiniteDuration timeout)
throws Exception {
checkNotNull(jobManager);
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/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
deleted file mode 100644
index 6221c08..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java
+++ /dev/null
@@ -1,142 +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.accumulators;
-
-import java.io.Serializable;
-import java.util.Collection;
-import java.util.Iterator;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.accumulators.IntCounter;
-import org.apache.flink.api.java.record.functions.ReduceFunction;
-import org.apache.flink.api.java.record.io.CsvOutputFormat;
-import org.apache.flink.api.java.record.io.TextInputFormat;
-import org.apache.flink.api.java.record.operators.BulkIteration;
-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.ReduceOperator;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.test.util.RecordAPITestBase;
-import org.apache.flink.types.Record;
-import org.apache.flink.types.StringValue;
-import org.apache.flink.util.Collector;
-import org.junit.Assert;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
-
-/**
- * To be finished !!! Didn't test with iterations yet;-(
- */
-@SuppressWarnings("deprecation")
-@RunWith(Parameterized.class)
-public class AccumulatorIterativeITCase extends RecordAPITestBase {
-
- private static final String INPUT = "1\n" + "2\n" + "3\n";
- private static final String EXPECTED = "6\n";
-
- private static final int NUM_ITERATIONS = 3;
- private static final int NUM_SUBTASKS = 1;
-
- protected String dataPath;
- protected String resultPath;
-
- public AccumulatorIterativeITCase(Configuration config) {
- super(config);
- }
-
- @Override
- protected void preSubmit() throws Exception {
- dataPath = createTempFile("datapoints.txt", INPUT);
- resultPath = getTempFilePath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(EXPECTED, resultPath);
-
- Integer res = getJobExecutionResult().getAccumulatorResult("test");
- Assert.assertEquals(Integer.valueOf(NUM_ITERATIONS * 6), res);
- }
-
- @Override
- protected Plan getTestJob() {
- Plan plan = getTestPlanPlan(config.getInteger("IterationAllReducer#NoSubtasks", 1), dataPath, resultPath);
- return plan;
- }
-
- @Parameters
- public static Collection<Object[]> getConfigurations() {
- Configuration config1 = new Configuration();
- config1.setInteger("IterationAllReducer#NoSubtasks", NUM_SUBTASKS);
- return toParameterList(config1);
- }
-
- static Plan getTestPlanPlan(int numSubTasks, String input, String output) {
-
- FileDataSource initialInput = new FileDataSource(TextInputFormat.class, input, "input");
-
- BulkIteration iteration = new BulkIteration("Loop");
- iteration.setInput(initialInput);
- iteration.setMaximumNumberOfIterations(NUM_ITERATIONS);
-
- ReduceOperator sumReduce = ReduceOperator.builder(new SumReducer())
- .input(iteration.getPartialSolution())
- .name("Compute sum (Reduce)")
- .build();
-
- iteration.setNextPartialSolution(sumReduce);
-
- @SuppressWarnings("unchecked")
- FileDataSink finalResult = new FileDataSink(new CsvOutputFormat("\n", " ", StringValue.class), output, iteration, "Output");
-
- Plan plan = new Plan(finalResult, "Iteration with AllReducer (keyless Reducer)");
- plan.setDefaultParallelism(numSubTasks);
- return plan;
- }
-
- static final class SumReducer extends ReduceFunction implements Serializable {
-
- private static final long serialVersionUID = 1L;
-
- private IntCounter testCounter = new IntCounter();
-
- @Override
- public void reduce(Iterator<Record> records, Collector<Record> out) {
- // Compute the sum
- int sum = 0;
-
- while (records.hasNext()) {
- Record r = records.next();
- Integer value = Integer.parseInt(r.getField(0, StringValue.class).getValue());
- sum += value;
- testCounter.add(value);
- }
- out.collect(new Record(new StringValue(Integer.toString(sum))));
- }
-
- @Override
- public void close() throws Exception {
- super.close();
- getRuntimeContext().addAccumulator("test", this.testCounter);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastBranchingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastBranchingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastBranchingITCase.java
index 69be0ed..5f2950c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastBranchingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastBranchingITCase.java
@@ -19,46 +19,24 @@
package org.apache.flink.test.broadcastvars;
import java.util.Collection;
-
-import org.apache.flink.api.common.Plan;
-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.io.CsvInputFormat;
-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 java.util.List;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.JoinFunction;
+import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.JoinOperator;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.test.util.ContractITCaseIOFormats.ContractITCaseOutputFormat;
-import org.apache.flink.test.util.RecordAPITestBase;
-import org.apache.flink.types.IntValue;
-import org.apache.flink.types.Record;
-import org.apache.flink.types.StringValue;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.util.Collector;
-@SuppressWarnings("deprecation")
-public class BroadcastBranchingITCase extends RecordAPITestBase {
-
- private static final String SC1_ID_ABC = "1 61 6 29\n2 7 13 10\n3 8 13 27\n";
-
- private static final String SC2_ID_X = "1 5\n2 3\n3 6";
-
- private static final String SC3_ID_Y = "1 2\n2 3\n3 7";
-
- private static final String RESULT = "2 112\n";
-
- private String sc1Path;
- private String sc2Path;
- private String sc3Path;
- private String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- sc1Path = createTempFile("broadcastBranchingInput/map_id_abc.txt", SC1_ID_ABC);
- sc2Path = createTempFile("broadcastBranchingInput/map_id_x.txt", SC2_ID_X);
- sc3Path = createTempFile("broadcastBranchingInput/map_id_y.txt", SC3_ID_Y);
- resultPath = getTempDirPath("result");
- }
+public class BroadcastBranchingITCase extends JavaProgramTestBase {
+ private static final String RESULT = "(2,112)\n";
// Sc1(id,a,b,c) --
// \
@@ -68,58 +46,50 @@ public class BroadcastBranchingITCase extends RecordAPITestBase {
// /
// Sc3(id,y) --------
@Override
- protected Plan getTestJob() {
+ protected void testProgram() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+
// Sc1 generates M parameters a,b,c for second degree polynomials P(x) = ax^2 + bx + c identified by id
- FileDataSource sc1 = new FileDataSource(new CsvInputFormat(), sc1Path);
- CsvInputFormat.configureRecordFormat(sc1).fieldDelimiter(' ').field(StringValue.class, 0).field(IntValue.class, 1)
- .field(IntValue.class, 2).field(IntValue.class, 3);
+ DataSet<Tuple4<String, Integer, Integer, Integer>> sc1 = env
+ .fromElements(new Tuple4<>("1", 61, 6, 29), new Tuple4<>("2", 7, 13, 10), new Tuple4<>("3", 8, 13, 27));
// Sc2 generates N x values to be evaluated with the polynomial identified by id
- FileDataSource sc2 = new FileDataSource(new CsvInputFormat(), sc2Path);
- CsvInputFormat.configureRecordFormat(sc2).fieldDelimiter(' ').field(StringValue.class, 0).field(IntValue.class, 1);
+ DataSet<Tuple2<String, Integer>> sc2 = env
+ .fromElements(new Tuple2<>("1", 5), new Tuple2<>("2", 3), new Tuple2<>("3", 6));
// Sc3 generates N y values to be evaluated with the polynomial identified by id
- FileDataSource sc3 = new FileDataSource(new CsvInputFormat(), sc3Path);
- CsvInputFormat.configureRecordFormat(sc3).fieldDelimiter(' ').field(StringValue.class, 0).field(IntValue.class, 1);
+ DataSet<Tuple2<String, Integer>> sc3 = env
+ .fromElements(new Tuple2<>("1", 2), new Tuple2<>("2", 3), new Tuple2<>("3", 7));
// Jn1 matches x and y values on id and emits (id, x, y) triples
- JoinOperator jn1 = JoinOperator.builder(Jn1.class, StringValue.class, 0, 0).input1(sc2).input2(sc3).build();
+ JoinOperator<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple3<String, Integer, Integer>> jn1
+ = sc2.join(sc3).where(0).equalTo(0).with(new Jn1());
// Jn2 matches polynomial and arguments by id, computes p = min(P(x),P(y)) and emits (id, p) tuples
- JoinOperator jn2 = JoinOperator.builder(Jn2.class, StringValue.class, 0, 0).input1(jn1).input2(sc1).build();
+ JoinOperator<Tuple3<String, Integer, Integer>, Tuple4<String, Integer, Integer, Integer>, Tuple2<String, Integer>> jn2
+ = jn1.join(sc1).where(0).equalTo(0).with(new Jn2());
// Mp1 selects (id, x, y) triples where x = y and broadcasts z (=x=y) to Mp2
- MapOperator mp1 = MapOperator.builder(Mp1.class).input(jn1).build();
+ FlatMapOperator<Tuple3<String, Integer, Integer>, Tuple2<String, Integer>> mp1
+ = jn1.flatMap(new Mp1());
// Mp2 filters out all p values which can be divided by z
- MapOperator mp2 = MapOperator.builder(Mp2.class).setBroadcastVariable("z", mp1).input(jn2).build();
-
- FileDataSink output = new FileDataSink(new ContractITCaseOutputFormat(), resultPath);
- output.setParallelism(1);
- output.setInput(mp2);
+ List<Tuple2<String, Integer>> result = jn2.flatMap(new Mp2()).withBroadcastSet(mp1, "z").collect();
- return new Plan(output);
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(RESULT, resultPath);
+ JavaProgramTestBase.compareResultAsText(result, RESULT);
}
- public static class Jn1 extends JoinFunction {
+ public static class Jn1 implements JoinFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple3<String, Integer, Integer>> {
private static final long serialVersionUID = 1L;
@Override
- public void join(Record sc2, Record sc3, Collector<Record> out) throws Exception {
- Record r = new Record(3);
- r.setField(0, sc2.getField(0, StringValue.class));
- r.setField(1, sc2.getField(1, IntValue.class));
- r.setField(2, sc3.getField(1, IntValue.class));
- out.collect(r);
+ public Tuple3<String, Integer, Integer> join(Tuple2<String, Integer> first, Tuple2<String, Integer> second) throws Exception {
+ return new Tuple3<>(first.f0, first.f1, second.f1);
}
}
- public static class Jn2 extends JoinFunction {
+ public static class Jn2 implements JoinFunction<Tuple3<String, Integer, Integer>, Tuple4<String, Integer, Integer, Integer>, Tuple2<String, Integer>> {
private static final long serialVersionUID = 1L;
private static int p(int x, int a, int b, int c) {
@@ -127,35 +97,35 @@ public class BroadcastBranchingITCase extends RecordAPITestBase {
}
@Override
- public void join(Record jn1, Record sc1, Collector<Record> out) throws Exception {
- int x = jn1.getField(1, IntValue.class).getValue();
- int y = jn1.getField(2, IntValue.class).getValue();
- int a = sc1.getField(1, IntValue.class).getValue();
- int b = sc1.getField(2, IntValue.class).getValue();
- int c = sc1.getField(3, IntValue.class).getValue();
+ public Tuple2<String, Integer> join(Tuple3<String, Integer, Integer> first, Tuple4<String, Integer, Integer, Integer> second) throws Exception {
+ int x = first.f1;
+ int y = first.f2;
+ int a = second.f1;
+ int b = second.f2;
+ int c = second.f3;
int p_x = p(x, a, b, c);
int p_y = p(y, a, b, c);
int min = Math.min(p_x, p_y);
- out.collect(new Record(jn1.getField(0, StringValue.class), new IntValue(min)));
+ return new Tuple2<>(first.f0, min);
}
}
- public static class Mp1 extends MapFunction {
+ public static class Mp1 implements FlatMapFunction<Tuple3<String, Integer, Integer>, Tuple2<String, Integer>> {
private static final long serialVersionUID = 1L;
@Override
- public void map(Record jn1, Collector<Record> out) throws Exception {
- if (jn1.getField(1, IntValue.class).getValue() == jn1.getField(2, IntValue.class).getValue()) {
- out.collect(new Record(jn1.getField(0, StringValue.class), jn1.getField(1, IntValue.class)));
+ public void flatMap(Tuple3<String, Integer, Integer> value, Collector<Tuple2<String, Integer>> out) throws Exception {
+ if (value.f1.compareTo(value.f2) == 0) {
+ out.collect(new Tuple2<>(value.f0, value.f1));
}
}
}
- public static class Mp2 extends MapFunction {
+ public static class Mp2 extends RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {
private static final long serialVersionUID = 1L;
- private Collection<Record> zs;
+ private Collection<Tuple2<String, Integer>> zs;
@Override
public void open(Configuration parameters) throws Exception {
@@ -163,13 +133,13 @@ public class BroadcastBranchingITCase extends RecordAPITestBase {
}
@Override
- public void map(Record jn2, Collector<Record> out) throws Exception {
- int p = jn2.getField(1, IntValue.class).getValue();
+ public void flatMap(Tuple2<String, Integer> value, Collector<Tuple2<String, Integer>> out) throws Exception {
+ int p = value.f1;
- for (Record z : zs) {
- if (z.getField(0, StringValue.class).getValue().equals(jn2.getField(0, StringValue.class).getValue())) {
- if (p % z.getField(1, IntValue.class).getValue() != 0) {
- out.collect(jn2);
+ for (Tuple2<String, Integer> z : zs) {
+ if (z.f0.equals(value.f0)) {
+ if (p % z.f1 != 0) {
+ out.collect(value);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/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
index 1a96a1b..14c396d 100644
--- 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
@@ -19,12 +19,9 @@
package org.apache.flink.test.cancelling;
-import java.util.concurrent.atomic.AtomicReference;
-
+import java.util.concurrent.TimeUnit;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.client.JobCancellationException;
-import org.apache.flink.runtime.messages.JobManagerMessages;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.test.util.ForkableFlinkMiniCluster;
import org.apache.flink.util.TestLogger;
@@ -38,11 +35,16 @@ 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;
/**
*
@@ -80,8 +82,14 @@ public abstract class CancellingTestBase extends TestLogger {
verifyJvmOptions();
Configuration config = new Configuration();
config.setBoolean(ConfigConstants.FILESYSTEM_DEFAULT_OVERWRITE_KEY, true);
-
- this.executor = new ForkableFlinkMiniCluster(config);
+ 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
@@ -104,43 +112,27 @@ public abstract class CancellingTestBase extends TestLogger {
try {
// submit job
final JobGraph jobGraph = getJobGraph(plan);
-
- final Thread currentThread = Thread.currentThread();
- final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
-
- boolean jobSuccessfullyCancelled = false;
-
- // trigger the cancelling asynchronous
- new Thread() {
- @Override
- public void run() {
- try {
- Thread.sleep(msecsTillCanceling);
- executor.getLeaderGateway(TestingUtils.TESTING_DURATION())
- .tell(new JobManagerMessages.CancelJob(jobGraph.getJobID()));
- }
- catch (Throwable t) {
- error.set(t);
- currentThread.interrupt();
- }
- }
- }.run();
-
- try {
- executor.submitJobAndWait(jobGraph, false);
- }
- catch (JobCancellationException exception) {
- jobSuccessfullyCancelled = true;
- }
- catch (Exception e) {
- throw new IllegalStateException("Job failed.", e);
- }
-
- if (!jobSuccessfullyCancelled) {
- throw new IllegalStateException("Job was not successfully cancelled.");
- }
- }
- catch(Exception e) {
+
+ 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));
}
@@ -154,7 +146,11 @@ public abstract class CancellingTestBase extends TestLogger {
return jgg.compileJobGraph(op);
}
- public void setTaskManagerNumSlots(int taskManagerNumSlots) { this.taskManagerNumSlots = taskManagerNumSlots; }
+ public void setTaskManagerNumSlots(int taskManagerNumSlots) {
+ this.taskManagerNumSlots = taskManagerNumSlots;
+ }
- public int getTaskManagerNumSlots() { return this.taskManagerNumSlots; }
+ public int getTaskManagerNumSlots() {
+ return this.taskManagerNumSlots;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/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 36db34d..55b1911 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
@@ -18,20 +18,14 @@
package org.apache.flink.test.cancelling;
-//import org.junit.Test;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.java.record.functions.MapFunction;
-import org.apache.flink.api.java.record.operators.GenericDataSink;
-import org.apache.flink.api.java.record.operators.GenericDataSource;
-import org.apache.flink.api.java.record.operators.MapOperator;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.test.recordJobs.util.DiscardingOutputFormat;
-import org.apache.flink.test.recordJobs.util.InfiniteIntegerInputFormat;
-import org.apache.flink.types.Record;
-import org.apache.flink.util.Collector;
+import org.apache.flink.test.util.InfiniteIntegerInputFormat;
+import org.junit.Test;
-@SuppressWarnings("deprecation")
public class MapCancelingITCase extends CancellingTestBase {
private static final int parallelism = 4;
@@ -39,119 +33,85 @@ public class MapCancelingITCase extends CancellingTestBase {
setTaskManagerNumSlots(parallelism);
}
-// @Test
+ @Test
public void testMapCancelling() throws Exception {
- GenericDataSource<InfiniteIntegerInputFormat> source = new GenericDataSource<InfiniteIntegerInputFormat>(
- new InfiniteIntegerInputFormat(), "Source");
- MapOperator mapper = MapOperator.builder(IdentityMapper.class)
- .input(source)
- .name("Identity Mapper")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), mapper, "Sink");
-
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 5 * 1000, 10 * 1000);
+ executeTask(new IdentityMapper<Integer>());
}
-// @Test
+ @Test
public void testSlowMapCancelling() throws Exception {
- GenericDataSource<InfiniteIntegerInputFormat> source = new GenericDataSource<InfiniteIntegerInputFormat>(
- new InfiniteIntegerInputFormat(), "Source");
- MapOperator mapper = MapOperator.builder(DelayingIdentityMapper.class)
- .input(source)
- .name("Delay Mapper")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), mapper, "Sink");
-
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 5 * 1000, 10 * 1000);
+ executeTask(new DelayingIdentityMapper<Integer>());
}
-// @Test
+ @Test
public void testMapWithLongCancellingResponse() throws Exception {
- GenericDataSource<InfiniteIntegerInputFormat> source = new GenericDataSource<InfiniteIntegerInputFormat>(
- new InfiniteIntegerInputFormat(), "Source");
- MapOperator mapper = MapOperator.builder(LongCancelTimeIdentityMapper.class)
- .input(source)
- .name("Long Cancelling Time Mapper")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), mapper, "Sink");
-
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 10 * 1000, 10 * 1000);
+ executeTask(new LongCancelTimeIdentityMapper<Integer>());
}
-// @Test
+ @Test
public void testMapPriorToFirstRecordReading() throws Exception {
- GenericDataSource<InfiniteIntegerInputFormat> source = new GenericDataSource<InfiniteIntegerInputFormat>(
- new InfiniteIntegerInputFormat(), "Source");
- MapOperator mapper = MapOperator.builder(StuckInOpenIdentityMapper.class)
- .input(source)
- .name("Stuck-In-Open Mapper")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), mapper, "Sink");
-
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 10 * 1000, 10 * 1000);
+ executeTask(new StuckInOpenIdentityMapper<Integer>());
+ }
+
+ public void executeTask(MapFunction<Integer, Integer> mapper) throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ env
+ .createInput(new InfiniteIntegerInputFormat(false))
+ .map(mapper)
+ .output(new DiscardingOutputFormat<Integer>());
+
+ env.setParallelism(parallelism);
+
+ runAndCancelJob(env.createProgramPlan(), 5 * 1000, 10 * 1000);
}
// --------------------------------------------------------------------------------------------
- public static final class IdentityMapper extends MapFunction {
+ public static final class IdentityMapper<IN> implements MapFunction<IN, IN> {
private static final long serialVersionUID = 1L;
@Override
- public void map(Record record, Collector<Record> out) throws Exception {
- out.collect(record);
+ public IN map(IN value) throws Exception {
+ return value;
}
}
- public static final class DelayingIdentityMapper extends MapFunction {
+ public static final class DelayingIdentityMapper<IN> implements MapFunction<IN, IN> {
private static final long serialVersionUID = 1L;
-
- private static final int WAIT_TIME_PER_RECORD = 10 * 1000; // 10 sec.
+
+ private static final int WAIT_TIME_PER_VALUE = 10 * 1000; // 10 sec.
@Override
- public void map(Record record, Collector<Record> out) throws Exception {
- Thread.sleep(WAIT_TIME_PER_RECORD);
- out.collect(record);
+ public IN map(IN value) throws Exception {
+ Thread.sleep(WAIT_TIME_PER_VALUE);
+ return value;
}
}
- public static final class LongCancelTimeIdentityMapper extends MapFunction {
+ public static final class LongCancelTimeIdentityMapper<IN> implements MapFunction<IN, IN> {
private static final long serialVersionUID = 1L;
-
- private static final int WAIT_TIME_PER_RECORD = 5 * 1000; // 5 sec.
+
+ private static final int WAIT_TIME_PER_VALUE = 5 * 1000; // 5 sec.
@Override
- public void map(Record record, Collector<Record> out) throws Exception {
+ public IN map(IN value) throws Exception {
final long start = System.currentTimeMillis();
- long remaining = WAIT_TIME_PER_RECORD;
+ long remaining = WAIT_TIME_PER_VALUE;
do {
try {
Thread.sleep(remaining);
- } catch (InterruptedException iex) {}
- } while ((remaining = WAIT_TIME_PER_RECORD - System.currentTimeMillis() + start) > 0);
-
- out.collect(record);
+ } catch (InterruptedException iex) {
+ }
+ } while ((remaining = WAIT_TIME_PER_VALUE - System.currentTimeMillis() + start) > 0);
+
+ return value;
}
}
- public static final class StuckInOpenIdentityMapper extends MapFunction {
+ public static final class StuckInOpenIdentityMapper<IN> extends RichMapFunction<IN, IN> {
private static final long serialVersionUID = 1L;
-
+
@Override
public void open(Configuration parameters) throws Exception {
synchronized (this) {
@@ -160,8 +120,8 @@ public class MapCancelingITCase extends CancellingTestBase {
}
@Override
- public void map(Record record, Collector<Record> out) throws Exception {
- out.collect(record);
+ public IN map(IN value) throws Exception {
+ return value;
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/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
index 8197e27..b6f82d4 100644
--- 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
@@ -19,233 +19,135 @@
package org.apache.flink.test.cancelling;
-//import org.junit.Test;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.java.record.functions.JoinFunction;
-import org.apache.flink.api.java.record.operators.GenericDataSink;
-import org.apache.flink.api.java.record.operators.GenericDataSource;
-import org.apache.flink.api.java.record.operators.JoinOperator;
+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.test.recordJobs.util.DiscardingOutputFormat;
-import org.apache.flink.test.recordJobs.util.InfiniteIntegerInputFormat;
-import org.apache.flink.test.recordJobs.util.InfiniteIntegerInputFormatWithDelay;
-import org.apache.flink.test.recordJobs.util.UniformIntInput;
-import org.apache.flink.types.IntValue;
-import org.apache.flink.types.Record;
-import org.apache.flink.util.Collector;
+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;
-@SuppressWarnings("deprecation")
public class MatchJoinCancelingITCase extends CancellingTestBase {
private static final int parallelism = 4;
- public MatchJoinCancelingITCase(){
+ 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 {
- GenericDataSource<InfiniteIntegerInputFormatWithDelay> source1 =
- new GenericDataSource<InfiniteIntegerInputFormatWithDelay>(new InfiniteIntegerInputFormatWithDelay(), "Source 1");
-
- GenericDataSource<InfiniteIntegerInputFormatWithDelay> source2 =
- new GenericDataSource<InfiniteIntegerInputFormatWithDelay>(new InfiniteIntegerInputFormatWithDelay(), "Source 2");
-
- JoinOperator matcher = JoinOperator.builder(SimpleMatcher.class, IntValue.class, 0, 0)
- .input1(source1)
- .input2(source2)
- .name("Sort Join")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), matcher, "Sink");
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 3000, 10*1000);
+ executeTask(new SimpleMatcher<Integer>(), true);
}
// @Test
public void testCancelSortMatchWhileReadingFastInputs() throws Exception {
- GenericDataSource<InfiniteIntegerInputFormat> source1 =
- new GenericDataSource<InfiniteIntegerInputFormat>(new InfiniteIntegerInputFormat(), "Source 1");
-
- GenericDataSource<InfiniteIntegerInputFormat> source2 =
- new GenericDataSource<InfiniteIntegerInputFormat>(new InfiniteIntegerInputFormat(), "Source 2");
-
- JoinOperator matcher = JoinOperator.builder(SimpleMatcher.class, IntValue.class, 0, 0)
- .input1(source1)
- .input2(source2)
- .name("Sort Join")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), matcher, "Sink");
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 5000, 10*1000);
+ executeTask(new SimpleMatcher<Integer>(), false);
}
// @Test
public void testCancelSortMatchPriorToFirstRecordReading() throws Exception {
- GenericDataSource<InfiniteIntegerInputFormat> source1 =
- new GenericDataSource<InfiniteIntegerInputFormat>(new InfiniteIntegerInputFormat(), "Source 1");
+ executeTask(new StuckInOpenMatcher<Integer>(), false);
+ }
- GenericDataSource<InfiniteIntegerInputFormat> source2 =
- new GenericDataSource<InfiniteIntegerInputFormat>(new InfiniteIntegerInputFormat(), "Source 2");
-
- JoinOperator matcher = JoinOperator.builder(StuckInOpenMatcher.class, IntValue.class, 0, 0)
- .input1(source1)
- .input2(source2)
- .name("Stuc-In-Open Match")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), matcher, "Sink");
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 5000);
-
- runAndCancelJob(p, 10 * 1000, 10 * 1000);
+ 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 {
- GenericDataSource<UniformIntInput> source1 =
- new GenericDataSource<UniformIntInput>(new UniformIntInput(), "Source 1");
- source1.setParameter(UniformIntInput.NUM_KEYS_KEY, 50000);
- source1.setParameter(UniformIntInput.NUM_VALUES_KEY, 100);
-
- GenericDataSource<UniformIntInput> source2 =
- new GenericDataSource<UniformIntInput>(new UniformIntInput(), "Source 2");
- source2.setParameter(UniformIntInput.NUM_KEYS_KEY, 50000);
- source2.setParameter(UniformIntInput.NUM_VALUES_KEY, 100);
-
- JoinOperator matcher = JoinOperator.builder(SimpleMatcher.class, IntValue.class, 0, 0)
- .input1(source1)
- .input2(source2)
- .name("Long Cancelling Sort Join")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), matcher, "Sink");
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 30 * 1000, 30 * 1000);
+ 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 {
- GenericDataSource<UniformIntInput> source1 =
- new GenericDataSource<UniformIntInput>(new UniformIntInput(), "Source 1");
- source1.setParameter(UniformIntInput.NUM_KEYS_KEY, 500);
- source1.setParameter(UniformIntInput.NUM_VALUES_KEY, 3);
-
- GenericDataSource<UniformIntInput> source2 =
- new GenericDataSource<UniformIntInput>(new UniformIntInput(), "Source 2");
- source2.setParameter(UniformIntInput.NUM_KEYS_KEY, 500);
- source2.setParameter(UniformIntInput.NUM_VALUES_KEY, 3);
-
- JoinOperator matcher = JoinOperator.builder(DelayingMatcher.class, IntValue.class, 0, 0)
- .input1(source1)
- .input2(source2)
- .name("Long Cancelling Sort Join")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), matcher, "Sink");
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 10 * 1000, 20 * 1000);
+ executeTaskWithGenerator(new DelayingMatcher<Integer>(), 500, 3, 10 * 1000, 20 * 1000);
}
// @Test
public void testCancelSortMatchWithLongCancellingResponse() throws Exception {
-
- GenericDataSource<UniformIntInput> source1 =
- new GenericDataSource<UniformIntInput>(new UniformIntInput(), "Source 1");
- source1.setParameter(UniformIntInput.NUM_KEYS_KEY, 500);
- source1.setParameter(UniformIntInput.NUM_VALUES_KEY, 3);
-
- GenericDataSource<UniformIntInput> source2 =
- new GenericDataSource<UniformIntInput>(new UniformIntInput(), "Source 2");
- source2.setParameter(UniformIntInput.NUM_KEYS_KEY, 500);
- source2.setParameter(UniformIntInput.NUM_VALUES_KEY, 3);
-
- JoinOperator matcher = JoinOperator.builder(LongCancelTimeMatcher.class, IntValue.class, 0, 0)
- .input1(source1)
- .input2(source2)
- .name("Long Cancelling Sort Join")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), matcher, "Sink");
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(parallelism);
-
- runAndCancelJob(p, 10 * 1000, 10 * 1000);
+ executeTaskWithGenerator(new LongCancelTimeMatcher<Integer>(), 500, 3, 10 * 1000, 10 * 1000);
}
// -------------------------------------- Test System corner cases ---------------------------------
// @Test
public void testCancelSortMatchWithHighparallelism() throws Exception {
-
- GenericDataSource<InfiniteIntegerInputFormat> source1 =
- new GenericDataSource<InfiniteIntegerInputFormat>(new InfiniteIntegerInputFormat(), "Source 1");
-
- GenericDataSource<InfiniteIntegerInputFormat> source2 =
- new GenericDataSource<InfiniteIntegerInputFormat>(new InfiniteIntegerInputFormat(), "Source 2");
-
- JoinOperator matcher = JoinOperator.builder(new SimpleMatcher(), IntValue.class, 0, 0)
- .input1(source1)
- .input2(source2)
- .name("Sort Join")
- .build();
- GenericDataSink sink = new GenericDataSink(new DiscardingOutputFormat(), matcher, "Sink");
-
- Plan p = new Plan(sink);
- p.setDefaultParallelism(64);
-
- runAndCancelJob(p, 3000, 20*1000);
+ executeTask(new SimpleMatcher<Integer>(), false, 64);
}
// --------------------------------------------------------------------------------------------
- public static final class SimpleMatcher extends JoinFunction {
+ 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 void join(Record value1, Record value2, Collector<Record> out) throws Exception {
- value1.setField(1, value2.getField(0, IntValue.class));
- out.collect(value1);
+ 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 extends JoinFunction {
+ 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 void join(Record value1, Record value2, Collector<Record> out) throws Exception {
+ public Tuple2<IN, IN> join(Tuple2<IN, IN> first, Tuple2<IN, IN> second) throws Exception {
Thread.sleep(WAIT_TIME_PER_RECORD);
- value1.setField(1, value2.getField(0, IntValue.class));
- out.collect(value1);
+ return new Tuple2<>(first.f0, second.f0);
}
}
- public static final class LongCancelTimeMatcher extends JoinFunction {
+ 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 void join(Record value1, Record value2, Collector<Record> out) throws Exception {
- value1.setField(1, value2.getField(0, IntValue.class));
-
+ 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 {
@@ -253,12 +155,11 @@ public class MatchJoinCancelingITCase extends CancellingTestBase {
Thread.sleep(remaining);
} catch (InterruptedException iex) {}
} while ((remaining = WAIT_TIME_PER_RECORD - System.currentTimeMillis() + start) > 0);
-
- out.collect(value1);
+ return new Tuple2<>(first.f0, second.f0);
}
}
- public static final class StuckInOpenMatcher extends JoinFunction {
+ public static final class StuckInOpenMatcher<IN> extends RichJoinFunction<Tuple2<IN, IN>, Tuple2<IN, IN>, Tuple2<IN, IN>> {
private static final long serialVersionUID = 1L;
@Override
@@ -269,9 +170,8 @@ public class MatchJoinCancelingITCase extends CancellingTestBase {
}
@Override
- public void join(Record value1, Record value2, Collector<Record> out) throws Exception {
- value1.setField(1, value2.getField(0, IntValue.class));
- out.collect(value1);
+ 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/8abae2c2/flink-tests/src/test/java/org/apache/flink/test/clients/examples/LocalExecutorITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/clients/examples/LocalExecutorITCase.java b/flink-tests/src/test/java/org/apache/flink/test/clients/examples/LocalExecutorITCase.java
index 3e2657c..5b05b67 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/clients/examples/LocalExecutorITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/clients/examples/LocalExecutorITCase.java
@@ -23,9 +23,12 @@ import java.io.FileWriter;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.Plan;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.LocalExecutor;
-import org.apache.flink.test.recordJobs.wordcount.WordCount;
import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.util.Collector;
import org.junit.Assert;
import org.junit.Test;
@@ -46,17 +49,13 @@ public class LocalExecutorITCase {
FileWriter fw = new FileWriter(inFile);
fw.write(WordCountData.TEXT);
fw.close();
-
- // run WordCount
- WordCount wc = new WordCount();
LocalExecutor executor = new LocalExecutor();
executor.setDefaultOverwriteFiles(true);
executor.setTaskManagerNumSlots(parallelism);
executor.setPrintStatusDuringExecution(false);
executor.start();
- Plan wcPlan = wc.getPlan(Integer.valueOf(parallelism).toString(),
- inFile.toURI().toString(), outFile.toURI().toString());
+ Plan wcPlan = getWordCountPlan(inFile, outFile, parallelism);
wcPlan.setExecutionConfig(new ExecutionConfig());
executor.executePlan(wcPlan);
executor.stop();
@@ -65,4 +64,27 @@ public class LocalExecutorITCase {
Assert.fail(e.getMessage());
}
}
+
+ private Plan getWordCountPlan(File inFile, File outFile, int parallelism) {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(parallelism);
+ env.readTextFile(inFile.getAbsolutePath())
+ .flatMap(new Tokenizer())
+ .groupBy(0)
+ .sum(1)
+ .writeAsCsv(outFile.getAbsolutePath());
+ return env.createProgramPlan();
+ }
+
+ public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
+ @Override
+ public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
+ String[] tokens = value.toLowerCase().split("\\W+");
+ for (String token : tokens) {
+ if (token.length() > 0) {
+ out.collect(new Tuple2<>(token, 1));
+ }
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/flink-tests/src/test/java/org/apache/flink/test/distributedCache/DistributedCacheTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/distributedCache/DistributedCacheTest.java b/flink-tests/src/test/java/org/apache/flink/test/distributedCache/DistributedCacheTest.java
index f625c57..4fb2d95 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/distributedCache/DistributedCacheTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/distributedCache/DistributedCacheTest.java
@@ -15,7 +15,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.flink.test.distributedCache;
import java.io.BufferedReader;
@@ -24,134 +23,69 @@ import java.io.IOException;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.cache.DistributedCache.DistributedCacheEntry;
-import org.apache.flink.api.java.record.functions.MapFunction;
-import org.apache.flink.api.java.record.io.CsvOutputFormat;
-import org.apache.flink.api.java.record.io.TextInputFormat;
-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.test.testdata.WordCountData;
-import org.apache.flink.test.util.RecordAPITestBase;
-import org.apache.flink.types.IntValue;
-import org.apache.flink.types.Record;
-import org.apache.flink.types.StringValue;
+import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.util.Collector;
/**
- * Test the distributed cache via using the cache file to do a selection on the input
+ * Tests the distributed cache by comparing a text file with a distributed copy.
*/
-@SuppressWarnings("deprecation")
-public class DistributedCacheTest extends RecordAPITestBase {
-
- public static final String cacheData = "machen\n" + "zeit\n" + "heerscharen\n" + "keiner\n" + "meine\n"
- + "fuehr\n" + "triumph\n" + "kommst\n" + "frei\n" + "schaffen\n" + "gesinde\n"
- + "langbeinigen\n" + "schalk\n" + "besser\n" + "solang\n" + "meer\n" + "fragst\n"
- + "gabriel\n" + "selbst\n" + "bin\n" + "sich\n" + "du\n" + "sogar\n" + "geht\n"
- + "immer\n" + "mensch\n" + "befestigt\n" + "lebt\n" + "mag\n" + "engeln\n" + "breiten\n"
- + "blitzendes\n" + "tags\n" + "sie\n" + "plagen\n" + "allzu\n" + "meisten\n" + "o\n"
- + "pfade\n" + "kennst\n" + "nichts\n" + "gedanken\n" + "befriedigt\n" + "mich\n" + "s\n"
- + "es\n" + "verneinen\n" + "er\n" + "gleich\n" + "baeumchen\n" + "donnergang\n";
+public class DistributedCacheTest extends JavaProgramTestBase {
- public static final String selectedCounts = "machen 1\n" + "zeit 2\n" + "heerscharen 1\n" + "keiner 2\n" + "meine 3\n"
- + "fuehr 1\n" + "triumph 1\n" + "kommst 1\n" + "frei 1\n" + "schaffen 1\n" + "gesinde 1\n"
- + "langbeinigen 1\n" + "schalk 1\n" + "besser 1\n" + "solang 1\n" + "meer 4\n" + "fragst 1\n"
- + "gabriel 1\n" + "selbst 2\n" + "bin 1\n" + "sich 7\n" + "du 11\n" + "sogar 1\n" + "geht 1\n"
- + "immer 4\n" + "mensch 2\n" + "befestigt 1\n" + "lebt 2\n" + "mag 3\n" + "engeln 2\n" + "breiten 1\n"
- + "blitzendes 1\n" + "tags 1\n" + "sie 2\n" + "plagen 2\n" + "allzu 1\n" + "meisten 1\n" + "o 1\n"
- + "pfade 1\n" + "kennst 1\n" + "nichts 3\n" + "gedanken 1\n" + "befriedigt 1\n" + "mich 6\n" + "s 3\n"
- + "es 8\n" + "verneinen 1\n" + "er 13\n" + "gleich 1\n" + "baeumchen 1\n" + "donnergang 1\n";
+ public static final String data
+ = "machen\n"
+ + "zeit\n"
+ + "heerscharen\n"
+ + "keiner\n"
+ + "meine\n";
protected String textPath;
- protected String cachePath;
- protected String resultPath;
- public static class TokenizeLine extends MapFunction {
+ @Override
+ protected void preSubmit() throws Exception {
+ textPath = createTempFile("count.txt", data);
+ }
+
+ @Override
+ protected void testProgram() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.registerCachedFile(textPath, "cache_test");
+
+ List<Tuple1<String>> result = env
+ .readTextFile(textPath)
+ .flatMap(new WordChecker())
+ .collect();
+
+ compareResultAsTuples(result, data);
+ }
+
+ public static class WordChecker extends RichFlatMapFunction<String, Tuple1<String>> {
private static final long serialVersionUID = 1L;
- private Set<String> stringList = new HashSet<String>();
- @Override
- public void open(org.apache.flink.configuration.Configuration conf) {
+ private final Set<String> wordList = new HashSet<>();
+ @Override
+ public void open(Configuration conf) throws FileNotFoundException, IOException {
File file = getRuntimeContext().getDistributedCache().getFile("cache_test");
- BufferedReader reader = null;
- try {
- reader = new BufferedReader(new FileReader(file));
- String tempString = null;
- while ((tempString = reader.readLine()) != null) {
- stringList.add(tempString);
- }
- reader.close();
- } catch (FileNotFoundException e) {
- e.printStackTrace();
- } catch (IOException e) {
- e.printStackTrace();
+ BufferedReader reader = new BufferedReader(new FileReader(file));
+ String tempString;
+ while ((tempString = reader.readLine()) != null) {
+ wordList.add(tempString);
}
+ reader.close();
}
@Override
- public void map(Record record, Collector<Record> collector) {
- String line = record.getField(0, StringValue.class).getValue();
- String [] element = line.split(" ");
- String word = element[0];
- int count = Integer.parseInt(element[1]);
- if (stringList.contains(word)) {
- collector.collect(new Record(new StringValue(word), new IntValue(count)));
+ public void flatMap(String word, Collector<Tuple1<String>> out) throws Exception {
+ if (wordList.contains(word)) {
+ out.collect(new Tuple1<>(word));
}
}
}
-
- public Plan getPlan(int numSubTasks, String dataInput, String output) {
-
-
- // input is {word, count} pair
- FileDataSource source = new FileDataSource(new TextInputFormat(), dataInput, "Input Lines");
-
- //do a selection using cached file
- MapOperator mapper = MapOperator.builder(new TokenizeLine())
- .input(source)
- .name("Tokenize Lines")
- .build();
-
- FileDataSink out = new FileDataSink(new CsvOutputFormat(), output, mapper, "Selection");
- CsvOutputFormat.configureRecordFormat(out)
- .recordDelimiter('\n')
- .fieldDelimiter(' ')
- .field(StringValue.class, 0)
- .field(IntValue.class, 1);
-
- Plan plan = new Plan(out, "Distributed Cache");
- plan.setDefaultParallelism(numSubTasks);
- plan.setExecutionConfig(new ExecutionConfig());
- return plan;
- }
-
-
- @Override
- protected void preSubmit() throws Exception {
- textPath = createTempFile("count.txt", WordCountData.COUNTS);
- cachePath = createTempFile("cache.txt", cacheData);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected Plan getTestJob() {
- Plan plan = getPlan(1 , textPath, resultPath);
- try {
- plan.registerCachedFile("cache_test", new DistributedCacheEntry(cachePath, false));
- } catch (IOException ex) {
- throw new RuntimeException(ex);
- }
- return plan;
- }
-
- @Override
- protected void postSubmit() throws Exception {
- // Test results
- compareResultsByLinesInMemory(selectedCounts, resultPath);
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/TaskFailureITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/TaskFailureITCase.java b/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/TaskFailureITCase.java
index efb1c32..fef6c9f 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/TaskFailureITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/TaskFailureITCase.java
@@ -18,168 +18,71 @@
package org.apache.flink.test.failingPrograms;
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.java.record.functions.MapFunction;
-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.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.test.util.ContractITCaseIOFormats.ContractITCaseInputFormat;
-import org.apache.flink.test.util.ContractITCaseIOFormats.ContractITCaseOutputFormat;
-import org.apache.flink.test.util.FailingTestBase;
-import org.apache.flink.types.IntValue;
-import org.apache.flink.types.Record;
-import org.apache.flink.types.StringValue;
-import org.apache.flink.util.Collector;
+import java.util.List;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.junit.Assert;
/**
- * Tests whether the system recovers from a runtime exception from the user code.
+ *
+ * Tests that both jobs, the failing and the working one, are handled correctly. The first (failing) job must be
+ * canceled and the client must report the failure. The second (working) job must finish successfully and compute the
+ * correct result.
+ *
*/
-public class TaskFailureITCase extends FailingTestBase {
-
- private static final int parallelism = 4;
-
- // input for map tasks
- private static final String MAP_IN = "1 1\n2 2\n2 8\n4 4\n4 4\n6 6\n7 7\n8 8\n" +
- "1 1\n2 2\n2 2\n4 4\n4 4\n6 3\n5 9\n8 8\n" +
- "1 1\n2 2\n2 2\n3 0\n4 4\n5 9\n7 7\n8 8\n" +
- "1 1\n9 1\n5 9\n4 4\n4 4\n6 6\n7 7\n8 8\n";
-
- // expected result of working map job
- private static final String MAP_RESULT = "1 11\n2 12\n4 14\n4 14\n1 11\n2 12\n2 12\n4 14\n4 14\n" +
- "3 16\n1 11\n2 12\n2 12\n0 13\n4 14\n1 11\n4 14\n4 14\n";
-
- private String inputPath;
- private String resultPath;
-
- public TaskFailureITCase(){
- setTaskManagerNumSlots(parallelism);
- }
-
- @Override
- protected void preSubmit() throws Exception {
- inputPath = createTempFile("input", MAP_IN);
- resultPath = getTempDirPath("result");
- }
-
-
- @Override
- protected JobGraph getFailingJobGraph() throws Exception {
-
- // init data source
- FileDataSource input = new FileDataSource(new ContractITCaseInputFormat(), inputPath);
-
- // init failing map task
- MapOperator testMapper = MapOperator.builder(FailingMapper.class).build();
-
- // init data sink
- FileDataSink output = new FileDataSink(new ContractITCaseOutputFormat(), resultPath);
-
- // compose failing program
- output.setInput(testMapper);
- testMapper.setInput(input);
-
- // generate plan
- Plan plan = new Plan(output);
- plan.setExecutionConfig(new ExecutionConfig());
- plan.setDefaultParallelism(parallelism);
-
- // optimize and compile plan
- Optimizer pc = new Optimizer(new DataStatistics(), this.config);
- OptimizedPlan op = pc.compile(plan);
-
- // return job graph of failing job
- JobGraphGenerator jgg = new JobGraphGenerator();
- return jgg.compileJobGraph(op);
- }
-
+public class TaskFailureITCase extends JavaProgramTestBase {
@Override
- protected JobGraph getJobGraph() throws Exception {
-
- // init data source
- FileDataSource input = new FileDataSource(new ContractITCaseInputFormat(), inputPath);
-
- // init (working) map task
- MapOperator testMapper = MapOperator.builder(TestMapper.class).build();
-
- // init data sink
- FileDataSink output = new FileDataSink(new ContractITCaseOutputFormat(), resultPath);
-
- // compose working program
- output.setInput(testMapper);
- testMapper.setInput(input);
-
- // generate plan
- Plan plan = new Plan(output);
- plan.setExecutionConfig(new ExecutionConfig());
- plan.setDefaultParallelism(4);
-
- // optimize and compile plan
- Optimizer pc = new Optimizer(new DataStatistics(), this.config);
- OptimizedPlan op = pc.compile(plan);
-
- // return job graph of working job
- JobGraphGenerator jgg = new JobGraphGenerator();
- return jgg.compileJobGraph(op);
- }
-
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(MAP_RESULT, resultPath);
+ protected void testProgram() throws Exception {
+ //test failing version
+ try {
+ executeTask(new FailingTestMapper());
+ } catch (RuntimeException e) { //expected for collection execution
+ if (!isCollectionExecution()) {
+ Assert.fail();
+ }
+ } catch (JobExecutionException e) { //expected for cluster execution
+ if (isCollectionExecution()) {
+ Assert.fail();
+ }
+ }
+ //test correct version
+ executeTask(new TestMapper());
}
- @Override
- protected int getTimeout() {
- // time out for this job is 30 secs
- return 30;
+ private void executeTask(MapFunction<Long, Long> mapper) throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ List<Long> result = env.generateSequence(1, 9)
+ .map(mapper)
+ .collect();
+ MultipleProgramsTestBase.compareResultAsText(result, "1\n2\n3\n4\n5\n6\n7\n8\n9");
}
/**
- * working program
+ * working map function
*/
- public static class TestMapper extends MapFunction {
-
+ public static class TestMapper implements MapFunction<Long, Long> {
private static final long serialVersionUID = 1L;
-
- private final StringValue string = new StringValue();
- private final IntValue integer = new IntValue();
@Override
- public void map(Record record, Collector<Record> out) throws Exception {
-
- final StringValue keyString = record.getField(0, this.string);
- final int key = Integer.parseInt(keyString.toString());
-
- final StringValue valueString = record.getField(1, this.string);
- final int value = Integer.parseInt(valueString.toString());
-
- if (key + value < 10) {
- record.setField(0, valueString);
- this.integer.setValue(key + 10);
- record.setField(1, this.integer);
- out.collect(record);
- }
+ public Long map(Long value) throws Exception {
+ return value;
}
}
/**
- * Map Failing program
+ * failing map function
*/
- public static class FailingMapper extends MapFunction {
-
+ public static class FailingTestMapper implements MapFunction<Long, Long> {
private static final long serialVersionUID = 1L;
@Override
- public void map(Record record, Collector<Record> out) throws Exception {
+ public Long map(Long value) throws Exception {
throw new RuntimeException("This is an expected Test Exception");
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/flink-tests/src/test/java/org/apache/flink/test/iterative/CoGroupConnectedComponentsITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/CoGroupConnectedComponentsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/CoGroupConnectedComponentsITCase.java
index 9023a1f..2a0b004 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/CoGroupConnectedComponentsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/CoGroupConnectedComponentsITCase.java
@@ -19,32 +19,25 @@
package org.apache.flink.test.iterative;
import java.io.BufferedReader;
-import java.io.Serializable;
import java.util.Iterator;
-import org.apache.flink.api.common.Plan;
-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.api.common.functions.CoGroupFunction;
+import org.apache.flink.api.common.functions.JoinFunction;
+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.functions.FunctionAnnotation.ForwardedFieldsFirst;
+import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFieldsSecond;
+import org.apache.flink.api.java.operators.CoGroupOperator;
+import org.apache.flink.api.java.operators.DeltaIteration;
+import org.apache.flink.api.java.operators.JoinOperator;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.test.testdata.ConnectedComponentsData;
-import org.apache.flink.test.util.RecordAPITestBase;
-import org.apache.flink.types.LongValue;
-import org.apache.flink.types.Record;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.util.Collector;
-
-@SuppressWarnings("deprecation")
-public class CoGroupConnectedComponentsITCase extends RecordAPITestBase {
+public class CoGroupConnectedComponentsITCase extends JavaProgramTestBase {
private static final long SEED = 0xBADC0FFEEBEEFL;
@@ -53,13 +46,11 @@ public class CoGroupConnectedComponentsITCase extends RecordAPITestBase {
private static final int NUM_EDGES = 10000;
+ private static final int MAX_ITERATIONS = 100;
+
protected String verticesPath;
protected String edgesPath;
protected String resultPath;
-
- public CoGroupConnectedComponentsITCase(){
- setTaskManagerNumSlots(parallelism);
- }
@Override
@@ -68,11 +59,6 @@ public class CoGroupConnectedComponentsITCase extends RecordAPITestBase {
edgesPath = createTempFile("edges.txt", ConnectedComponentsData.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED));
resultPath = getTempFilePath("results");
}
-
- @Override
- protected Plan getTestJob() {
- return getPlan(parallelism, verticesPath, edgesPath, resultPath, 100);
- }
@Override
protected void postSubmit() throws Exception {
@@ -85,83 +71,70 @@ public class CoGroupConnectedComponentsITCase extends RecordAPITestBase {
// The test program
// --------------------------------------------------------------------------------------------
+ @Override
+ protected void testProgram() throws Exception {
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Tuple1<Long>> initialVertices = env.readCsvFile(verticesPath).fieldDelimiter(" ").types(Long.class).name("Vertices");
+
+ DataSet<Tuple2<Long, Long>> edges = env.readCsvFile(edgesPath).fieldDelimiter(" ").types(Long.class, Long.class).name("Edges");
- @ConstantFieldsFirst(0)
- @ConstantFieldsSecond(0)
- public static final class MinIdAndUpdate extends CoGroupFunction implements Serializable {
+ DataSet<Tuple2<Long, Long>> verticesWithId = initialVertices.map(new MapFunction<Tuple1<Long>, Tuple2<Long, Long>>() {
+ @Override
+ public Tuple2<Long, Long> map(Tuple1<Long> value) throws Exception {
+ return new Tuple2<>(value.f0, value.f0);
+ }
+ }).name("Assign Vertex Ids");
+
+ DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration = verticesWithId.iterateDelta(verticesWithId, MAX_ITERATIONS, 0);
+
+ JoinOperator<Tuple2<Long, Long>, Tuple2<Long, Long>, Tuple2<Long, Long>> joinWithNeighbors = iteration.getWorkset()
+ .join(edges).where(0).equalTo(0)
+ .with(new JoinFunction<Tuple2<Long, Long>, Tuple2<Long, Long>, Tuple2<Long, Long>>() {
+ @Override
+ public Tuple2<Long, Long> join(Tuple2<Long, Long> first, Tuple2<Long, Long> second) throws Exception {
+ return new Tuple2<>(second.f1, first.f1);
+ }
+ })
+ .name("Join Candidate Id With Neighbor");
+
+ CoGroupOperator<Tuple2<Long, Long>, Tuple2<Long, Long>, Tuple2<Long, Long>> minAndUpdate = joinWithNeighbors
+ .coGroup(iteration.getSolutionSet()).where(0).equalTo(0)
+ .with(new MinIdAndUpdate())
+ .name("min Id and Update");
+
+ iteration.closeWith(minAndUpdate, minAndUpdate).writeAsCsv(resultPath, "\n", " ").name("Result");
+
+ env.execute("Workset Connected Components");
+ }
+
+ @ForwardedFieldsFirst("f1->f1")
+ @ForwardedFieldsSecond("f0->f0")
+ public static final class MinIdAndUpdate implements CoGroupFunction<Tuple2<Long, Long>, Tuple2<Long, Long>, Tuple2<Long, Long>> {
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 {
+ public void coGroup(Iterable<Tuple2<Long, Long>> first, Iterable<Tuple2<Long, Long>> second, Collector<Tuple2<Long, Long>> out) throws Exception {
+ Iterator<Tuple2<Long, Long>> current = second.iterator();
if (!current.hasNext()) {
throw new Exception("Error: Id not encountered before.");
}
- Record old = current.next();
- long oldId = old.getField(1, LongValue.class).getValue();
+ Tuple2<Long, Long> old = current.next();
+ long oldId = old.f1;
long minimumComponentID = Long.MAX_VALUE;
- while (candidates.hasNext()) {
- Record candidate = candidates.next();
- long candidateComponentID = candidate.getField(1, LongValue.class).getValue();
+ for (Tuple2<Long, Long> candidate : first) {
+ long candidateComponentID = candidate.f1;
if (candidateComponentID < minimumComponentID) {
minimumComponentID = candidateComponentID;
}
}
if (minimumComponentID < oldId) {
- newComponentId.setValue(minimumComponentID);
- old.setField(1, newComponentId);
- out.collect(old);
+ out.collect(new Tuple2<>(old.f0, minimumComponentID));
}
}
}
-
- @SuppressWarnings("unchecked")
- public static Plan getPlan(int numSubTasks, String verticesInput, String edgeInput, String output, int maxIterations) {
-
- // 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/8abae2c2/flink-tests/src/test/java/org/apache/flink/test/iterative/DeltaPageRankITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/DeltaPageRankITCase.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/DeltaPageRankITCase.java
deleted file mode 100644
index bf459c6..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/DeltaPageRankITCase.java
+++ /dev/null
@@ -1,108 +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.iterative;
-
-import org.apache.flink.api.common.Plan;
-import org.apache.flink.test.recordJobs.graph.WorksetConnectedComponents;
-import org.apache.flink.test.util.RecordAPITestBase;
-
-
-public class DeltaPageRankITCase extends RecordAPITestBase {
-
- protected String verticesPath;
- protected String edgesPath;
- protected String deltasPath;
-
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- verticesPath = createTempFile("vertices.txt", INITIAL_VERTICES_WITH_RANK);
- deltasPath = createTempFile("deltas.txt", INITIAL_DELTAS);
- edgesPath = createTempFile("edges.txt", EDGES);
-
- resultPath = getTempFilePath("results");
- }
-
- @Override
- protected Plan getTestJob() {
- String[] params = { String.valueOf(parallelism) , verticesPath, edgesPath, resultPath, "3" };
-
- WorksetConnectedComponents cc = new WorksetConnectedComponents();
- return cc.getPlan(params);
- }
-
- @Override
- protected void postSubmit() throws Exception {
-// compareResultsByLinesInMemory(RESULT_RANKS, resultPath);
- }
-
-
- private static final String INITIAL_VERTICES_WITH_RANK = "1 0.025\n" +
- "2 0.125\n" +
- "3 0.0833333333333333\n" +
- "4 0.0833333333333333\n" +
- "5 0.075\n" +
- "6 0.075\n" +
- "7 0.183333333333333\n" +
- "8 0.15\n" +
- "9 0.1\n";
-
- private static final String INITIAL_DELTAS = "1 -0.075\n" +
- "2 0.025\n" +
- "3 -0.0166666666666667\n" +
- "4 -0.0166666666666667\n" +
- "5 -0.025\n" +
- "6 -0.025\n" +
- "7 0.0833333333333333\n" +
- "8 0.05\n" +
- "9 0\n";
-
- private static final String EDGES = "1 2 2\n" +
- "1 3 2\n" +
- "2 3 3\n" +
- "2 4 3\n" +
- "3 1 4\n" +
- "3 2 4\n" +
- "4 2 2\n" +
- "5 6 2\n" +
- "6 5 2\n" +
- "7 8 2\n" +
- "7 9 2\n" +
- "8 7 2\n" +
- "8 9 2\n" +
- "9 7 2\n" +
- "9 8 2\n" +
- "3 5 4\n" +
- "3 6 4\n" +
- "4 8 2\n" +
- "2 7 3\n" +
- "5 7 2\n" +
- "6 4 2\n";
-
-// private static final String RESULT_RANKS = "1 0.00698784722222222\n" +
-// "2 0.0326822916666667\n" +
-// "3 0.0186631944444444\n" +
-// "4 0.0293402777777778\n" +
-// "5 0.0220920138888889\n" +
-// "6 0.0220920138888889\n" +
-// "7 0.262152777777778\n" +
-// "8 0.260763888888889\n" +
-// "9 0.245225694444444\n";
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/8abae2c2/flink-tests/src/test/java/org/apache/flink/test/iterative/IterationTerminationWithTerminationTail.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/IterationTerminationWithTerminationTail.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/IterationTerminationWithTerminationTail.java
index 0915a42..6350533 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/IterationTerminationWithTerminationTail.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/IterationTerminationWithTerminationTail.java
@@ -18,115 +18,55 @@
package org.apache.flink.test.iterative;
-import java.io.Serializable;
-import java.util.Iterator;
-
-import org.apache.flink.api.common.Plan;
-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.io.CsvOutputFormat;
-import org.apache.flink.api.java.record.io.TextInputFormat;
-import org.apache.flink.api.java.record.operators.BulkIteration;
-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.util.RecordAPITestBase;
-import org.apache.flink.types.Record;
-import org.apache.flink.types.StringValue;
+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 org.apache.flink.util.Collector;
-import org.junit.Assert;
-
-@SuppressWarnings("deprecation")
-public class IterationTerminationWithTerminationTail extends RecordAPITestBase {
- private static final String INPUT = "1\n" + "2\n" + "3\n" + "4\n" + "5\n";
+public class IterationTerminationWithTerminationTail extends JavaProgramTestBase {
private static final String EXPECTED = "22\n";
- protected String dataPath;
- protected String resultPath;
-
- public IterationTerminationWithTerminationTail(){
- setTaskManagerNumSlots(parallelism);
- }
-
@Override
- protected void preSubmit() throws Exception {
- dataPath = createTempFile("datapoints.txt", INPUT);
- resultPath = getTempFilePath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(EXPECTED, resultPath);
- }
+ protected void testProgram() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(4);
- @Override
- protected Plan getTestJob() {
- return getTestPlanPlan(parallelism, dataPath, resultPath);
- }
-
- private static Plan getTestPlanPlan(int numSubTasks, String input, String output) {
+ DataSet<String> initialInput = env.fromElements("1", "2", "3", "4", "5").name("input");
+
+ IterativeDataSet<String> iteration = initialInput.iterate(5).name("Loop");
- FileDataSource initialInput = new FileDataSource(TextInputFormat.class, input, "input");
-
- BulkIteration iteration = new BulkIteration("Loop");
- iteration.setInput(initialInput);
- iteration.setMaximumNumberOfIterations(5);
- Assert.assertTrue(iteration.getMaximumNumberOfIterations() > 1);
+ DataSet<String> sumReduce = iteration.reduceGroup(new SumReducer()).name("Compute sum (GroupReduce");
- ReduceOperator sumReduce = ReduceOperator.builder(new SumReducer())
- .input(iteration.getPartialSolution())
- .name("Compute sum (Reduce)")
- .build();
-
- iteration.setNextPartialSolution(sumReduce);
-
- MapOperator terminationMapper = MapOperator.builder(new TerminationMapper())
- .input(sumReduce)
- .name("Compute termination criterion (Map)")
- .build();
-
- iteration.setTerminationCriterion(terminationMapper);
+ DataSet<String> terminationFilter = sumReduce.filter(new TerminationFilter()).name("Compute termination criterion (Map)");
- FileDataSink finalResult = new FileDataSink(CsvOutputFormat.class, output, iteration, "Output");
- CsvOutputFormat.configureRecordFormat(finalResult)
- .recordDelimiter('\n')
- .fieldDelimiter(' ')
- .field(StringValue.class, 0);
+ List<String> result = iteration.closeWith(sumReduce, terminationFilter).collect();
- Plan plan = new Plan(finalResult, "Iteration with AllReducer (keyless Reducer)");
- plan.setDefaultParallelism(numSubTasks);
- Assert.assertTrue(plan.getDefaultParallelism() > 1);
- return plan;
+ containsResultAsText(result, EXPECTED);
}
-
- public static final class SumReducer extends ReduceFunction implements Serializable {
-
+
+ public static final class SumReducer implements GroupReduceFunction<String, String> {
private static final long serialVersionUID = 1L;
-
+
@Override
- public void reduce(Iterator<Record> it, Collector<Record> out) {
- // Compute the sum
+ public void reduce(Iterable<String> values, Collector<String> out) throws Exception {
int sum = 0;
- while (it.hasNext()) {
- sum += Integer.parseInt(it.next().getField(0, StringValue.class).getValue()) + 1;
+ for (String value : values) {
+ sum += Integer.parseInt(value) + 1;
}
-
- out.collect(new Record(new StringValue(Integer.toString(sum))));
+ out.collect("" + sum);
}
}
-
- public static class TerminationMapper extends MapFunction implements Serializable {
+
+ public static class TerminationFilter implements FilterFunction<String> {
private static final long serialVersionUID = 1L;
-
- @Override
- public void map(Record record, Collector<Record> collector) {
-
- int currentSum = Integer.parseInt(record.getField(0, StringValue.class).getValue());
- if(currentSum < 22)
- collector.collect(record);
+ @Override
+ public boolean filter(String value) throws Exception {
+ return Integer.parseInt(value) < 22;
}
}