You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2023/02/09 03:46:31 UTC
[flink] 02/03: [FLINK-30392][runtime] Change broadcast hybrid selective result partition type to hybrid full in compile phase instead of partition create phase. In the previous implementation, we will force the broadcast hybrid selective edge to be of hybrid full(re-consumable) in the ResultPartitionTypeFactory. But this is not the correct phase to do this replacement, because for scheduling topology, the result is not changed still. This inconsistency will cause many problems, such as increasing the cos [...]
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit dabc3ba57776dc31f476d7882e7ebcd973f3909d
Author: Weijie Guo <re...@163.com>
AuthorDate: Mon Feb 6 12:28:49 2023 +0800
[FLINK-30392][runtime] Change broadcast hybrid selective result partition type to hybrid full in compile phase instead of partition create phase.
In the previous implementation, we will force the broadcast hybrid selective edge to be of hybrid full(re-consumable) in the ResultPartitionTypeFactory. But this is not the correct phase to do this replacement, because for scheduling topology, the result is not changed still. This inconsistency will cause many problems, such as increasing the cost of fail-over.
---
.../network/partition/ResultPartitionFactory.java | 9 ------
.../partition/ResultPartitionFactoryTest.java | 12 --------
.../api/graph/StreamingJobGraphGenerator.java | 20 ++++++++++----
.../api/graph/StreamingJobGraphGeneratorTest.java | 32 ++++++++++++++++++++++
4 files changed, 47 insertions(+), 26 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
index 968fbcc80dc..3d8c27bc09d 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
@@ -227,15 +227,6 @@ public class ResultPartitionFactory {
}
} else if (type == ResultPartitionType.HYBRID_FULL
|| type == ResultPartitionType.HYBRID_SELECTIVE) {
- if (type == ResultPartitionType.HYBRID_SELECTIVE && isBroadcast) {
- // for broadcast result partition, it can be optimized to always use full spilling
- // strategy to significantly reduce shuffle data writing cost.
- LOG.info(
- "{} result partition has been replaced by {} result partition to reduce shuffle data writing cost.",
- type,
- ResultPartitionType.HYBRID_FULL);
- type = ResultPartitionType.HYBRID_FULL;
- }
partition =
new HsResultPartition(
taskNameWithSubtaskAndId,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
index 07d4a97f28b..7b4da59471d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
@@ -132,18 +132,6 @@ class ResultPartitionFactoryTest {
assertThat(resultPartition.isReleased()).isFalse();
}
- @Test
- public void testHybridBroadcastEdgeAlwaysUseFullResultPartition() {
- ResultPartition resultPartition =
- createResultPartition(
- ResultPartitionType.HYBRID_SELECTIVE, Integer.MAX_VALUE, true);
- assertThat(resultPartition.partitionType).isEqualTo(ResultPartitionType.HYBRID_FULL);
-
- resultPartition =
- createResultPartition(ResultPartitionType.HYBRID_FULL, Integer.MAX_VALUE, true);
- assertThat(resultPartition.partitionType).isEqualTo(ResultPartitionType.HYBRID_FULL);
- }
-
private static ResultPartition createResultPartition(ResultPartitionType partitionType) {
return createResultPartition(partitionType, Integer.MAX_VALUE);
}
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 cbb02e5a164..20e7dba85b3 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
@@ -1072,6 +1072,21 @@ public class StreamingJobGraphGenerator {
dataSetId = consumerEdge.getIntermediateDatasetIdToProduce();
}
+ if (partitionType.isHybridResultPartition()) {
+ hasHybridResultPartition = true;
+ if (partitioner.isBroadcast()
+ && partitionType == ResultPartitionType.HYBRID_SELECTIVE) {
+ // for broadcast result partition, it can be optimized to always use full
+ // spilling strategy to significantly reduce shuffle data writing cost.
+ LOG.info(
+ "{} result partition has been replaced by {} result partition to support "
+ + "broadcast optimization, which will reduce shuffle data writing cost.",
+ partitionType.name(),
+ ResultPartitionType.HYBRID_FULL.name());
+ partitionType = ResultPartitionType.HYBRID_FULL;
+ }
+ }
+
NonChainedOutput output =
new NonChainedOutput(
consumerEdge.supportsUnalignedCheckpoints(),
@@ -1183,11 +1198,6 @@ public class StreamingJobGraphGenerator {
StreamPartitioner<?> partitioner = output.getPartitioner();
ResultPartitionType resultPartitionType = output.getPartitionType();
- if (resultPartitionType == ResultPartitionType.HYBRID_FULL
- || resultPartitionType == ResultPartitionType.HYBRID_SELECTIVE) {
- hasHybridResultPartition = true;
- }
-
checkBufferTimeout(resultPartitionType, edge);
JobEdge jobEdge;
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 b2448fd103b..3dbecaaebb8 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
@@ -103,6 +103,7 @@ import org.apache.flink.streaming.api.transformations.MultipleInputTransformatio
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.api.transformations.StreamExchangeMode;
+import org.apache.flink.streaming.runtime.partitioner.BroadcastPartitioner;
import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
import org.apache.flink.streaming.runtime.partitioner.RebalancePartitioner;
import org.apache.flink.streaming.runtime.partitioner.RescalePartitioner;
@@ -121,6 +122,8 @@ import org.assertj.core.api.Assertions;
import org.assertj.core.data.Offset;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.io.ObjectOutputStream;
@@ -1779,6 +1782,35 @@ class StreamingJobGraphGeneratorTest {
.isEqualTo(cacheTransformation.getDatasetId());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testHybridBroadcastEdgeAlwaysUseFullResultPartition(boolean isSelective) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ env.disableOperatorChaining();
+
+ DataStream<Integer> sourceDataStream = env.fromElements(1, 2, 3);
+ DataStream<Integer> partitionAfterSourceDataStream =
+ new DataStream<>(
+ env,
+ new PartitionTransformation<>(
+ sourceDataStream.getTransformation(),
+ new BroadcastPartitioner<>(),
+ isSelective
+ ? StreamExchangeMode.HYBRID_SELECTIVE
+ : StreamExchangeMode.HYBRID_FULL));
+ partitionAfterSourceDataStream.addSink(new DiscardingSink<>());
+
+ JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph());
+
+ List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
+ assertThat(verticesSorted).hasSize(2);
+
+ JobVertex sourceAndMapVertex = verticesSorted.get(0);
+ assertThat(sourceAndMapVertex.getProducedDataSets().get(0).getResultType())
+ .isEqualTo(ResultPartitionType.HYBRID_FULL);
+ }
+
/**
* Tests that multiple downstream consumer vertices can reuse the same intermediate blocking
* dataset if they have the same parallelism and partitioner.