From 4bdb03173e398c8dbcf9fe68bd05f0d325cb0b69 Mon Sep 17 00:00:00 2001 From: Stephan Ewen Date: Mon, 18 Jan 2016 18:27:53 +0100 Subject: [PATCH] [FLINK-3255] [streaming] Disable parallelism-dependent chaining optimization --- .../streaming/api/graph/StreamGraph.java | 8 +- .../api/graph/StreamingJobGraphGenerator.java | 15 ++-- .../api/graph/StreamGraphGeneratorTest.java | 8 +- .../graph/StreamingJobGraphGeneratorTest.java | 77 +++++++++++++++---- 4 files changed, 76 insertions(+), 32 deletions(-) diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java index a1e68e1be0c55b..fa8c9d48459aed 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java @@ -465,7 +465,7 @@ public Collection getStreamNodes() { } public Set>> getOperators() { - Set>> operatorSet = new HashSet>>(); + Set>> operatorSet = new HashSet<>(); for (StreamNode vertex : streamNodes.values()) { operatorSet.add(new Tuple2>(vertex.getId(), vertex .getOperator())); @@ -496,7 +496,7 @@ public Tuple2 createIterationSourceAndSink(int loopId, i sinks.add(sink.getId()); setParallelism(sink.getId(), parallelism); - iterationSourceSinkPairs.add(new Tuple2(source, sink)); + iterationSourceSinkPairs.add(new Tuple2<>(source, sink)); source.setOperatorName("IterationSource-" + loopId); sink.setOperatorName("IterationSink-" + loopId); @@ -505,7 +505,7 @@ public Tuple2 createIterationSourceAndSink(int loopId, i this.vertexIDtoLoopTimeout.put(source.getId(), timeout); this.vertexIDtoLoopTimeout.put(sink.getId(), timeout); - return new Tuple2(source, sink); + return new Tuple2<>(source, sink); } public Set> getIterationSourceSinkPairs() { @@ -518,7 +518,7 @@ private void removeEdge(StreamEdge edge) { } private void removeVertex(StreamNode toRemove) { - Set edgesToRemove = new HashSet(); + Set edgesToRemove = new HashSet<>(); edgesToRemove.addAll(toRemove.getInEdges()); edgesToRemove.addAll(toRemove.getOutEdges()); diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java index ad96cbf1d84408..50c6a156ef359f 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java @@ -229,14 +229,14 @@ private List createChain( return transitiveOutEdges; } else { - return new ArrayList(); + return new ArrayList<>(); } } private String createChainedName(Integer vertexID, List chainedOutputs) { String operatorName = streamGraph.getStreamNode(vertexID).getOperatorName(); if (chainedOutputs.size() > 1) { - List outputChainedNames = new ArrayList(); + List outputChainedNames = new ArrayList<>(); for (StreamEdge chainable : chainedOutputs) { outputChainedNames.add(chainedNames.get(chainable.getTargetId())); } @@ -395,8 +395,7 @@ private boolean isChainable(StreamEdge edge) { && (headOperator.getChainingStrategy() == ChainingStrategy.HEAD || headOperator.getChainingStrategy() == ChainingStrategy.ALWAYS || headOperator.getChainingStrategy() == ChainingStrategy.FORCE_ALWAYS) - && (edge.getPartitioner() instanceof ForwardPartitioner || downStreamVertex - .getParallelism() == 1) + && (edge.getPartitioner() instanceof ForwardPartitioner) && upStreamVertex.getParallelism() == downStreamVertex.getParallelism() && (streamGraph.isChainingEnabled() || outOperator.getChainingStrategy() == ChainingStrategy.FORCE_ALWAYS); @@ -404,7 +403,7 @@ private boolean isChainable(StreamEdge edge) { private void setSlotSharing() { - Map slotSharingGroups = new HashMap(); + Map slotSharingGroups = new HashMap<>(); for (Entry entry : jobVertices.entrySet()) { @@ -446,15 +445,15 @@ private void configureCheckpointing() { // collect the vertices that receive "trigger checkpoint" messages. // currently, these are all the sources - List triggerVertices = new ArrayList(); + List triggerVertices = new ArrayList<>(); // collect the vertices that need to acknowledge the checkpoint // currently, these are all vertices - List ackVertices = new ArrayList(jobVertices.size()); + List ackVertices = new ArrayList<>(jobVertices.size()); // collect the vertices that receive "commit checkpoint" messages // currently, these are all vertices - List commitVertices = new ArrayList(); + List commitVertices = new ArrayList<>(); for (JobVertex vertex : jobVertices.values()) { if (vertex.isInputVertex()) { diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java index bf8fbfe416f586..734199bbf72285 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java @@ -51,7 +51,7 @@ * specific tests, for example in {@link org.apache.flink.streaming.api.IterateTest} for * iterations. */ -public class StreamGraphGeneratorTest extends StreamingMultipleProgramsTestBase { +public class StreamGraphGeneratorTest { /** * This tests whether virtual Transformations behave correctly. @@ -282,14 +282,12 @@ public TypeInformation getTypeInformation() { } @Override - public void processElement(StreamRecord element) throws Exception { + public void processElement(StreamRecord element) { output.collect(element); } @Override - public void processWatermark(Watermark mark) throws Exception { - - } + public void processWatermark(Watermark mark) {} @Override public void setOutputType(TypeInformation outTypeInfo, ExecutionConfig executionConfig) { diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java index e8064289188d7c..b5f1e20eae3cdb 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java @@ -14,29 +14,30 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.flink.streaming.api.graph; import java.io.IOException; import java.util.Random; import org.apache.flink.api.common.ExecutionConfig; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.runtime.jobgraph.JobGraph; +import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; -import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase; -import org.apache.flink.streaming.util.TestStreamEnvironment; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; import org.apache.flink.util.InstantiationUtil; -import org.junit.Assert; + import org.junit.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -public class StreamingJobGraphGeneratorTest extends StreamingMultipleProgramsTestBase { - private static final Logger LOG = LoggerFactory.getLogger(StreamingJobGraphGeneratorTest.class); +import static org.junit.Assert.*; + +public class StreamingJobGraphGeneratorTest { @Test public void testExecutionConfigSerialization() throws IOException, ClassNotFoundException { final long seed = System.currentTimeMillis(); - LOG.info("Test seed: {}", new Long(seed)); final Random r = new Random(seed); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -76,16 +77,62 @@ public void testExecutionConfigSerialization() throws IOException, ClassNotFound config.setParallelism(dop); JobGraph jobGraph = compiler.createJobGraph("test"); - ExecutionConfig executionConfig = (ExecutionConfig) InstantiationUtil.readObjectFromConfig( + + ExecutionConfig executionConfig = InstantiationUtil.readObjectFromConfig( jobGraph.getJobConfiguration(), ExecutionConfig.CONFIG_KEY, Thread.currentThread().getContextClassLoader()); - Assert.assertEquals(closureCleanerEnabled, executionConfig.isClosureCleanerEnabled()); - Assert.assertEquals(forceAvroEnabled, executionConfig.isForceAvroEnabled()); - Assert.assertEquals(forceKryoEnabled, executionConfig.isForceKryoEnabled()); - Assert.assertEquals(objectReuseEnabled, executionConfig.isObjectReuseEnabled()); - Assert.assertEquals(sysoutLoggingEnabled, executionConfig.isSysoutLoggingEnabled()); - Assert.assertEquals(dop, executionConfig.getParallelism()); + assertNotNull(executionConfig); + + assertEquals(closureCleanerEnabled, executionConfig.isClosureCleanerEnabled()); + assertEquals(forceAvroEnabled, executionConfig.isForceAvroEnabled()); + assertEquals(forceKryoEnabled, executionConfig.isForceKryoEnabled()); + assertEquals(objectReuseEnabled, executionConfig.isObjectReuseEnabled()); + assertEquals(sysoutLoggingEnabled, executionConfig.isSysoutLoggingEnabled()); + assertEquals(dop, executionConfig.getParallelism()); + } + + @Test + public void testParallelismOneNotChained() { + + // --------- the program --------- + + final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + + DataStream> input = env + .fromElements("a", "b", "c", "d", "e", "f") + .map(new MapFunction>() { + @Override + public Tuple2 map(String value) { + return new Tuple2<>(value, value); + } + }); + + DataStream> result = input + .keyBy(0) + .map(new MapFunction, Tuple2>() { + + @Override + public Tuple2 map(Tuple2 value) { + return value; + } + }); + + result.addSink(new SinkFunction>() { + @Override + public void invoke(Tuple2 value) {} + }); + + // --------- the job graph --------- + + StreamGraph streamGraph = env.getStreamGraph(); + streamGraph.setJobName("test job"); + JobGraph jobGraph = streamGraph.getJobGraph(); + + assertEquals(2, jobGraph.getNumberOfVertices()); + assertEquals(1, jobGraph.getVerticesAsArray()[0].getParallelism()); + assertEquals(1, jobGraph.getVerticesAsArray()[1].getParallelism()); } }