You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/06/17 11:41:38 UTC
[flink] branch master updated: [FLINK-12823][datastream] Add
ShuffleMode property to PartitionTransformation
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f2e7320 [FLINK-12823][datastream] Add ShuffleMode property to PartitionTransformation
f2e7320 is described below
commit f2e7320fc3b2898aefb22730395c46e9e83e585a
Author: ifndef-SleePy <mm...@gmail.com>
AuthorDate: Thu Jun 13 15:16:41 2019 +0800
[FLINK-12823][datastream] Add ShuffleMode property to PartitionTransformation
This also wires the property all the way through to the JobGraph.
---
.../flink/streaming/api/graph/StreamEdge.java | 22 ++++++++++++
.../flink/streaming/api/graph/StreamGraph.java | 30 ++++++++++++-----
.../streaming/api/graph/StreamGraphGenerator.java | 3 +-
.../api/graph/StreamingJobGraphGenerator.java | 20 +++++++++--
.../transformations/PartitionTransformation.java | 30 ++++++++++++++++-
.../streaming/api/transformations/ShuffleMode.java | 39 ++++++++++++++++++++++
.../api/graph/StreamingJobGraphGeneratorTest.java | 31 +++++++++++++++++
7 files changed, 162 insertions(+), 13 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
index 33a16b0..48c70d0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
@@ -18,12 +18,15 @@
package org.apache.flink.streaming.api.graph;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.api.transformations.ShuffleMode;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
import org.apache.flink.util.OutputTag;
import java.io.Serializable;
import java.util.List;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
* An edge in the streaming topology. One edge like this does not necessarily
* gets converted to a connection between two job vertices (due to
@@ -70,8 +73,22 @@ public class StreamEdge implements Serializable {
*/
private final String targetOperatorName;
+ private final ShuffleMode shuffleMode;
+
public StreamEdge(StreamNode sourceVertex, StreamNode targetVertex, int typeNumber,
List<String> selectedNames, StreamPartitioner<?> outputPartitioner, OutputTag outputTag) {
+ this(sourceVertex,
+ targetVertex,
+ typeNumber,
+ selectedNames,
+ outputPartitioner,
+ outputTag,
+ ShuffleMode.PIPELINED);
+ }
+
+ public StreamEdge(StreamNode sourceVertex, StreamNode targetVertex, int typeNumber,
+ List<String> selectedNames, StreamPartitioner<?> outputPartitioner, OutputTag outputTag,
+ ShuffleMode shuffleMode) {
this.sourceId = sourceVertex.getId();
this.targetId = targetVertex.getId();
this.typeNumber = typeNumber;
@@ -80,6 +97,7 @@ public class StreamEdge implements Serializable {
this.outputTag = outputTag;
this.sourceOperatorName = sourceVertex.getOperatorName();
this.targetOperatorName = targetVertex.getOperatorName();
+ this.shuffleMode = checkNotNull(shuffleMode);
this.edgeId = sourceVertex + "_" + targetVertex + "_" + typeNumber + "_" + selectedNames
+ "_" + outputPartitioner;
@@ -109,6 +127,10 @@ public class StreamEdge implements Serializable {
return outputPartitioner;
}
+ public ShuffleMode getShuffleMode() {
+ return shuffleMode;
+ }
+
public void setPartitioner(StreamPartitioner<?> partitioner) {
this.outputPartitioner = partitioner;
}
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 63b4724..4cc18eb 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
@@ -27,6 +27,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.MissingTypeInfo;
import org.apache.flink.optimizer.plan.StreamingPlan;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -36,6 +37,7 @@ import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.transformations.ShuffleMode;
import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
import org.apache.flink.streaming.runtime.partitioner.RebalancePartitioner;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
@@ -87,7 +89,7 @@ public class StreamGraph extends StreamingPlan {
private Set<Integer> sinks;
private Map<Integer, Tuple2<Integer, List<String>>> virtualSelectNodes;
private Map<Integer, Tuple2<Integer, OutputTag>> virtualSideOutputNodes;
- private Map<Integer, Tuple2<Integer, StreamPartitioner<?>>> virtualPartitionNodes;
+ private Map<Integer, Tuple3<Integer, StreamPartitioner<?>, ShuffleMode>> virtualPartitionNodes;
protected Map<Integer, String> vertexIDtoBrokerID;
protected Map<Integer, Long> vertexIDtoLoopTimeout;
@@ -340,14 +342,17 @@ public class StreamGraph extends StreamingPlan {
* @param virtualId ID of the virtual node.
* @param partitioner The partitioner
*/
- public void addVirtualPartitionNode(Integer originalId, Integer virtualId, StreamPartitioner<?> partitioner) {
+ public void addVirtualPartitionNode(
+ Integer originalId,
+ Integer virtualId,
+ StreamPartitioner<?> partitioner,
+ ShuffleMode shuffleMode) {
if (virtualPartitionNodes.containsKey(virtualId)) {
throw new IllegalStateException("Already has virtual partition node with id " + virtualId);
}
- virtualPartitionNodes.put(virtualId,
- new Tuple2<Integer, StreamPartitioner<?>>(originalId, partitioner));
+ virtualPartitionNodes.put(virtualId, new Tuple3<>(originalId, partitioner, shuffleMode));
}
/**
@@ -375,6 +380,7 @@ public class StreamGraph extends StreamingPlan {
typeNumber,
null,
new ArrayList<String>(),
+ null,
null);
}
@@ -384,7 +390,8 @@ public class StreamGraph extends StreamingPlan {
int typeNumber,
StreamPartitioner<?> partitioner,
List<String> outputNames,
- OutputTag outputTag) {
+ OutputTag outputTag,
+ ShuffleMode shuffleMode) {
if (virtualSideOutputNodes.containsKey(upStreamVertexID)) {
int virtualId = upStreamVertexID;
@@ -392,7 +399,7 @@ public class StreamGraph extends StreamingPlan {
if (outputTag == null) {
outputTag = virtualSideOutputNodes.get(virtualId).f1;
}
- addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, null, outputTag);
+ addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, null, outputTag, shuffleMode);
} else if (virtualSelectNodes.containsKey(upStreamVertexID)) {
int virtualId = upStreamVertexID;
upStreamVertexID = virtualSelectNodes.get(virtualId).f0;
@@ -400,14 +407,15 @@ public class StreamGraph extends StreamingPlan {
// selections that happen downstream override earlier selections
outputNames = virtualSelectNodes.get(virtualId).f1;
}
- addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag);
+ addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag, shuffleMode);
} else if (virtualPartitionNodes.containsKey(upStreamVertexID)) {
int virtualId = upStreamVertexID;
upStreamVertexID = virtualPartitionNodes.get(virtualId).f0;
if (partitioner == null) {
partitioner = virtualPartitionNodes.get(virtualId).f1;
}
- addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag);
+ shuffleMode = virtualPartitionNodes.get(virtualId).f2;
+ addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames, outputTag, shuffleMode);
} else {
StreamNode upstreamNode = getStreamNode(upStreamVertexID);
StreamNode downstreamNode = getStreamNode(downStreamVertexID);
@@ -429,7 +437,11 @@ public class StreamGraph extends StreamingPlan {
}
}
- StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner, outputTag);
+ if (shuffleMode == null) {
+ shuffleMode = ShuffleMode.PIPELINED;
+ }
+
+ StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner, outputTag, shuffleMode);
getStreamNode(edge.getSourceId()).addOutEdge(edge);
getStreamNode(edge.getTargetId()).addInEdge(edge);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
index ef3162b..792b719 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
@@ -249,7 +249,8 @@ public class StreamGraphGenerator {
Collection<Integer> transformedIds = transform(input);
for (Integer transformedId: transformedIds) {
int virtualId = StreamTransformation.getNewNodeId();
- streamGraph.addVirtualPartitionNode(transformedId, virtualId, partition.getPartitioner());
+ streamGraph.addVirtualPartitionNode(
+ transformedId, virtualId, partition.getPartitioner(), partition.getShuffleMode());
resultIds.add(virtualId);
}
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 7bd49d0..99a6f2a 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
@@ -51,6 +51,7 @@ import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import org.apache.flink.streaming.api.operators.UdfStreamOperatorFactory;
+import org.apache.flink.streaming.api.transformations.ShuffleMode;
import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
import org.apache.flink.streaming.runtime.partitioner.RescalePartitioner;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
@@ -506,17 +507,31 @@ public class StreamingJobGraphGenerator {
downStreamConfig.setNumberOfInputs(downStreamConfig.getNumberOfInputs() + 1);
StreamPartitioner<?> partitioner = edge.getPartitioner();
+
+ ResultPartitionType resultPartitionType;
+ switch (edge.getShuffleMode()) {
+ case PIPELINED:
+ resultPartitionType = ResultPartitionType.PIPELINED_BOUNDED;
+ break;
+ case BATCH:
+ resultPartitionType = ResultPartitionType.BLOCKING;
+ break;
+ default:
+ throw new UnsupportedOperationException("Data exchange mode " +
+ edge.getShuffleMode() + " is not supported yet.");
+ }
+
JobEdge jobEdge;
if (partitioner instanceof ForwardPartitioner || partitioner instanceof RescalePartitioner) {
jobEdge = downStreamVertex.connectNewDataSetAsInput(
headVertex,
DistributionPattern.POINTWISE,
- ResultPartitionType.PIPELINED_BOUNDED);
+ resultPartitionType);
} else {
jobEdge = downStreamVertex.connectNewDataSetAsInput(
headVertex,
DistributionPattern.ALL_TO_ALL,
- ResultPartitionType.PIPELINED_BOUNDED);
+ resultPartitionType);
}
// set strategy name so that web interface can show it.
jobEdge.setShipStrategyName(partitioner.toString());
@@ -542,6 +557,7 @@ public class StreamingJobGraphGenerator {
&& (headOperator.getChainingStrategy() == ChainingStrategy.HEAD ||
headOperator.getChainingStrategy() == ChainingStrategy.ALWAYS)
&& (edge.getPartitioner() instanceof ForwardPartitioner)
+ && edge.getShuffleMode() != ShuffleMode.BATCH
&& upStreamVertex.getParallelism() == downStreamVertex.getParallelism()
&& streamGraph.isChainingEnabled();
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java
index 6f30e0f..73e8ab5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java
@@ -27,6 +27,8 @@ import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import java.util.Collection;
import java.util.List;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
* This transformation represents a change of partitioning of the input elements.
*
@@ -39,8 +41,11 @@ import java.util.List;
public class PartitionTransformation<T> extends StreamTransformation<T> {
private final StreamTransformation<T> input;
+
private final StreamPartitioner<T> partitioner;
+ private final ShuffleMode shuffleMode;
+
/**
* Creates a new {@code PartitionTransformation} from the given input and
* {@link StreamPartitioner}.
@@ -49,9 +54,25 @@ public class PartitionTransformation<T> extends StreamTransformation<T> {
* @param partitioner The {@code StreamPartitioner}
*/
public PartitionTransformation(StreamTransformation<T> input, StreamPartitioner<T> partitioner) {
+ this(input, partitioner, ShuffleMode.PIPELINED);
+ }
+
+ /**
+ * Creates a new {@code PartitionTransformation} from the given input and
+ * {@link StreamPartitioner}.
+ *
+ * @param input The input {@code StreamTransformation}
+ * @param partitioner The {@code StreamPartitioner}
+ * @param shuffleMode The {@code ShuffleMode}
+ */
+ public PartitionTransformation(
+ StreamTransformation<T> input,
+ StreamPartitioner<T> partitioner,
+ ShuffleMode shuffleMode) {
super("Partition", input.getOutputType(), input.getParallelism());
this.input = input;
this.partitioner = partitioner;
+ this.shuffleMode = checkNotNull(shuffleMode);
}
/**
@@ -69,6 +90,13 @@ public class PartitionTransformation<T> extends StreamTransformation<T> {
return partitioner;
}
+ /**
+ * Returns the {@link ShuffleMode} of this {@link PartitionTransformation}.
+ */
+ public ShuffleMode getShuffleMode() {
+ return shuffleMode;
+ }
+
@Override
public Collection<StreamTransformation<?>> getTransitivePredecessors() {
List<StreamTransformation<?>> result = Lists.newArrayList();
@@ -79,6 +107,6 @@ public class PartitionTransformation<T> extends StreamTransformation<T> {
@Override
public final void setChainingStrategy(ChainingStrategy strategy) {
- throw new UnsupportedOperationException("Cannot set chaining strategy on Union Transformation.");
+ throw new UnsupportedOperationException("Cannot set chaining strategy on Partition Transformation.");
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/ShuffleMode.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/ShuffleMode.java
new file mode 100644
index 0000000..583c57e
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/ShuffleMode.java
@@ -0,0 +1,39 @@
+/*
+ * 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.streaming.api.transformations;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * The shuffle mode defines the data exchange mode between operators.
+ */
+@PublicEvolving
+public enum ShuffleMode {
+ /**
+ * Producer and consumer are online at the same time.
+ * Produced data is received by consumer immediately.
+ */
+ PIPELINED,
+
+ /**
+ * The producer first produces its entire result and finishes.
+ * After that, the consumer is started and may consume the data.
+ */
+ BATCH
+}
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 996ac42..c909b32 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
@@ -45,6 +45,9 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
+import org.apache.flink.streaming.api.transformations.PartitionTransformation;
+import org.apache.flink.streaming.api.transformations.ShuffleMode;
+import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
import org.apache.flink.util.Collector;
import org.apache.flink.util.TestLogger;
@@ -368,4 +371,32 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
OutputFormat<?> sinkFormat2 = outputFormats.get(nameToOperatorIds.get("Sink: sink2")).getUserCodeObject();
assertTrue(sinkFormat2 instanceof DiscardingOutputFormat);
}
+
+ /**
+ * Test manually setting shuffle mode.
+ */
+ @Test
+ public void testShuffleMode() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ // fromElements -> Map -> Print, will not chain since the batch data exchange mode
+ DataStream<Integer> mapDataStream = env.fromElements(1, 2, 3)
+ .map((MapFunction<Integer, Integer>) value -> value).setParallelism(2);
+
+ DataStream<Integer> partitionDataStream = new DataStream<>(env, new PartitionTransformation<>(
+ mapDataStream.getTransformation(), new ForwardPartitioner<>(), ShuffleMode.BATCH));
+ partitionDataStream.print().setParallelism(2);
+
+ JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph());
+
+ List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ assertEquals(3, verticesSorted.size());
+
+ JobVertex sourceVertex = verticesSorted.get(0);
+ JobVertex mapVertex = verticesSorted.get(1);
+ JobVertex printVertex = verticesSorted.get(2);
+
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, mapVertex.getInputs().get(0).getSource().getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, printVertex.getInputs().get(0).getSource().getResultType());
+ }
}