You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/04/23 16:50:53 UTC
[GitHub] [flink] GJL commented on a change in pull request #11774: [FLINK-17020][runtime] Introduce GlobalDataExchangeMode for JobGraph generation
GJL commented on a change in pull request #11774:
URL: https://github.com/apache/flink/pull/11774#discussion_r413926186
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
##########
@@ -898,6 +825,116 @@ public void testSlotSharingOnAllVerticesInSameSlotSharingGroupByDefaultDisabled(
assertDistinctSharingGroups(source1Vertex, source2Vertex, map2Vertex);
}
+ @Test
+ public void testDefaultGlobalDataExchangeModeIsAllEdgesPipelined() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ assertThat(streamGraph.getGlobalDataExchangeMode(), is(GlobalDataExchangeMode.ALL_EDGES_PIPELINED));
+ }
+
+ @Test
+ public void testAllEdgesBlockingMode() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_BLOCKING);
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+ final JobVertex map1Vertex = verticesSorted.get(1);
+ final JobVertex map2Vertex = verticesSorted.get(2);
+
+ assertEquals(ResultPartitionType.BLOCKING, sourceVertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map1Vertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map2Vertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ @Test
+ public void testAllEdgesPipelinedMode() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_PIPELINED);
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+ final JobVertex map1Vertex = verticesSorted.get(1);
+ final JobVertex map2Vertex = verticesSorted.get(2);
+
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, map1Vertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, map2Vertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ @Test
+ public void testForwardEdgesPipelinedMode() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED);
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+ final JobVertex map1Vertex = verticesSorted.get(1);
+ final JobVertex map2Vertex = verticesSorted.get(2);
+
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map1Vertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map2Vertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ @Test
+ public void testPointwiseEdgesPipelinedMode() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED);
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+ final JobVertex map1Vertex = verticesSorted.get(1);
+ final JobVertex map2Vertex = verticesSorted.get(2);
+
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, map1Vertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map2Vertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ @Test
+ public void testGlobalDataExchangeModeDoesNotOverrideSpecifiedShuffleMode() {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ final DataStream<Integer> source = env.fromElements(1, 2, 3).setParallelism(1);
+ final DataStream<Integer> forward = new DataStream<>(env, new PartitionTransformation<>(
+ source.getTransformation(), new ForwardPartitioner<>(), ShuffleMode.PIPELINED));
+ forward.map(i -> i).startNewChain().setParallelism(1);
+ final StreamGraph streamGraph = env.getStreamGraph();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_BLOCKING);
+
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ /**
+ * Topology: source(parallelism=1) --(forward)--> map1(parallelism=1)
+ * --(rescale)--> map2(parallelism=2) --(rebalance)--> sink(parallelism=2).
+ */
+ private StreamGraph createStreamGraphForGlobalDataExchangeModeTests() {
Review comment:
Maybe it makes sense to move the added tests to a new class.
Also, can be declared `static`.
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
##########
@@ -898,6 +825,116 @@ public void testSlotSharingOnAllVerticesInSameSlotSharingGroupByDefaultDisabled(
assertDistinctSharingGroups(source1Vertex, source2Vertex, map2Vertex);
}
+ @Test
+ public void testDefaultGlobalDataExchangeModeIsAllEdgesPipelined() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ assertThat(streamGraph.getGlobalDataExchangeMode(), is(GlobalDataExchangeMode.ALL_EDGES_PIPELINED));
+ }
+
+ @Test
+ public void testAllEdgesBlockingMode() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_BLOCKING);
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+ final JobVertex map1Vertex = verticesSorted.get(1);
+ final JobVertex map2Vertex = verticesSorted.get(2);
+
+ assertEquals(ResultPartitionType.BLOCKING, sourceVertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map1Vertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map2Vertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ @Test
+ public void testAllEdgesPipelinedMode() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_PIPELINED);
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+ final JobVertex map1Vertex = verticesSorted.get(1);
+ final JobVertex map2Vertex = verticesSorted.get(2);
+
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, map1Vertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, map2Vertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ @Test
+ public void testForwardEdgesPipelinedMode() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.FORWARD_EDGES_PIPELINED);
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+ final JobVertex map1Vertex = verticesSorted.get(1);
+ final JobVertex map2Vertex = verticesSorted.get(2);
+
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map1Vertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map2Vertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ @Test
+ public void testPointwiseEdgesPipelinedMode() {
+ final StreamGraph streamGraph = createStreamGraphForGlobalDataExchangeModeTests();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.POINTWISE_EDGES_PIPELINED);
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+ final JobVertex map1Vertex = verticesSorted.get(1);
+ final JobVertex map2Vertex = verticesSorted.get(2);
+
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, map1Vertex.getProducedDataSets().get(0).getResultType());
+ assertEquals(ResultPartitionType.BLOCKING, map2Vertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ @Test
+ public void testGlobalDataExchangeModeDoesNotOverrideSpecifiedShuffleMode() {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ final DataStream<Integer> source = env.fromElements(1, 2, 3).setParallelism(1);
+ final DataStream<Integer> forward = new DataStream<>(env, new PartitionTransformation<>(
+ source.getTransformation(), new ForwardPartitioner<>(), ShuffleMode.PIPELINED));
+ forward.map(i -> i).startNewChain().setParallelism(1);
+ final StreamGraph streamGraph = env.getStreamGraph();
+ streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_BLOCKING);
+
+ final JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ final List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ final JobVertex sourceVertex = verticesSorted.get(0);
+
+ assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
+ }
+
+ /**
+ * Topology: source(parallelism=1) --(forward)--> map1(parallelism=1)
+ * --(rescale)--> map2(parallelism=2) --(rebalance)--> sink(parallelism=2).
+ */
+ private StreamGraph createStreamGraphForGlobalDataExchangeModeTests() {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ final DataStream<Integer> source = env.fromElements(1, 2, 3).setParallelism(1);
+
+ final DataStream<Integer> forward = new DataStream<>(env, new PartitionTransformation<>(
+ source.getTransformation(), new ForwardPartitioner<>(), ShuffleMode.UNDEFINED));
+ final DataStream<Integer> map1 = forward.map(i -> i).startNewChain().setParallelism(1);
+
+ final DataStream<Integer> rescale = new DataStream<>(env, new PartitionTransformation<>(
+ map1.getTransformation(), new RescalePartitioner<>(), ShuffleMode.UNDEFINED));
+ final DataStream<Integer> map2 = rescale.map(i -> i).setParallelism(2);
Review comment:
Is there a benefit compared to calling `map1.rescale().map(i -> i).setParallelism(2)` ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org