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:32 UTC
[flink] 03/03: [FLINK-30902][runtime] Change reusable hybrid result partition to hybrid full to enable partition reuse optimization.
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 4fb5f5e48aeeedba724dd1b44f2b819b40011944
Author: Weijie Guo <re...@163.com>
AuthorDate: Mon Feb 6 12:39:21 2023 +0800
[FLINK-30902][runtime] Change reusable hybrid result partition to hybrid full to enable partition reuse optimization.
This closes #21856
---
.../streaming/api/graph/NonChainedOutput.java | 6 +-
.../api/graph/StreamingJobGraphGenerator.java | 116 ++++++++++++++-------
.../api/graph/StreamingJobGraphGeneratorTest.java | 110 +++++++++++++++++++
3 files changed, 195 insertions(+), 37 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/NonChainedOutput.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/NonChainedOutput.java
index 1042b396276..f1d08d4b7df 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/NonChainedOutput.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/NonChainedOutput.java
@@ -62,7 +62,7 @@ public class NonChainedOutput implements Serializable {
private StreamPartitioner<?> partitioner;
/** Target {@link ResultPartitionType}. */
- private final ResultPartitionType partitionType;
+ private ResultPartitionType partitionType;
public NonChainedOutput(
boolean supportsUnalignedCheckpoints,
@@ -123,6 +123,10 @@ public class NonChainedOutput implements Serializable {
this.partitioner = partitioner;
}
+ public void setPartitionType(ResultPartitionType partitionType) {
+ this.partitionType = partitionType;
+ }
+
public StreamPartitioner<?> getPartitioner() {
return partitioner;
}
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 20e7dba85b3..3eceb27dc6a 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
@@ -758,6 +758,19 @@ public class StreamingJobGraphGenerator {
}
}
+ private void checkAndReplaceReusableHybridPartitionType(NonChainedOutput reusableOutput) {
+ if (reusableOutput.getPartitionType() == ResultPartitionType.HYBRID_SELECTIVE) {
+ // for can be reused hybrid output, it can be optimized to always use full
+ // spilling strategy to significantly reduce shuffle data writing cost.
+ reusableOutput.setPartitionType(ResultPartitionType.HYBRID_FULL);
+ LOG.info(
+ "{} result partition has been replaced by {} result partition to support partition reuse,"
+ + " which will reduce shuffle data writing cost.",
+ reusableOutput.getPartitionType().name(),
+ ResultPartitionType.HYBRID_FULL.name());
+ }
+ }
+
private InputOutputFormatContainer getOrCreateFormatContainer(Integer startNodeId) {
return chainedInputOutputFormats.computeIfAbsent(
startNodeId,
@@ -1057,11 +1070,6 @@ public class StreamingJobGraphGenerator {
opIntermediateOutputs.computeIfAbsent(vertexId, ignored -> new HashMap<>());
for (StreamEdge consumerEdge : consumerEdges) {
checkState(vertexId == consumerEdge.getSourceId(), "Vertex id must be the same.");
- int consumerParallelism =
- streamGraph.getStreamNode(consumerEdge.getTargetId()).getParallelism();
- int consumerMaxParallelism =
- streamGraph.getStreamNode(consumerEdge.getTargetId()).getMaxParallelism();
- StreamPartitioner<?> partitioner = consumerEdge.getPartitioner();
ResultPartitionType partitionType = getResultPartitionType(consumerEdge);
IntermediateDataSetID dataSetId = new IntermediateDataSetID();
@@ -1074,7 +1082,7 @@ public class StreamingJobGraphGenerator {
if (partitionType.isHybridResultPartition()) {
hasHybridResultPartition = true;
- if (partitioner.isBroadcast()
+ if (consumerEdge.getPartitioner().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.
@@ -1087,6 +1095,55 @@ public class StreamingJobGraphGenerator {
}
}
+ createOrReuseOutput(
+ outputs,
+ outputsConsumedByEdge,
+ consumerEdge,
+ isPersistentDataSet,
+ dataSetId,
+ partitionType);
+ }
+ return outputs;
+ }
+
+ private void createOrReuseOutput(
+ List<NonChainedOutput> outputs,
+ Map<StreamEdge, NonChainedOutput> outputsConsumedByEdge,
+ StreamEdge consumerEdge,
+ boolean isPersistentDataSet,
+ IntermediateDataSetID dataSetId,
+ ResultPartitionType partitionType) {
+ int consumerParallelism =
+ streamGraph.getStreamNode(consumerEdge.getTargetId()).getParallelism();
+ int consumerMaxParallelism =
+ streamGraph.getStreamNode(consumerEdge.getTargetId()).getMaxParallelism();
+ NonChainedOutput reusableOutput = null;
+ if (isPartitionTypeCanBeReuse(partitionType)) {
+ for (NonChainedOutput outputCandidate : outputsConsumedByEdge.values()) {
+ // Reusing the same output can improve performance. The target output can be reused
+ // if meeting the following conditions:
+ // 1. all is hybrid partition or are same re-consumable partition.
+ // 2. have the same partitioner, consumer parallelism, persistentDataSetId,
+ // outputTag.
+ if (allHybridOrSameReconsumablePartitionType(
+ outputCandidate.getPartitionType(), partitionType)
+ && consumerParallelism == outputCandidate.getConsumerParallelism()
+ && consumerMaxParallelism == outputCandidate.getConsumerMaxParallelism()
+ && Objects.equals(
+ outputCandidate.getPersistentDataSetId(),
+ consumerEdge.getIntermediateDatasetIdToProduce())
+ && Objects.equals(
+ outputCandidate.getOutputTag(), consumerEdge.getOutputTag())
+ && Objects.equals(
+ consumerEdge.getPartitioner(), outputCandidate.getPartitioner())) {
+ reusableOutput = outputCandidate;
+ outputsConsumedByEdge.put(consumerEdge, reusableOutput);
+ checkAndReplaceReusableHybridPartitionType(reusableOutput);
+ break;
+ }
+ }
+ }
+ if (reusableOutput == null) {
NonChainedOutput output =
new NonChainedOutput(
consumerEdge.supportsUnalignedCheckpoints(),
@@ -1097,38 +1154,25 @@ public class StreamingJobGraphGenerator {
isPersistentDataSet,
dataSetId,
consumerEdge.getOutputTag(),
- partitioner,
+ consumerEdge.getPartitioner(),
partitionType);
- if (!partitionType.isReconsumable()) {
- outputs.add(output);
- outputsConsumedByEdge.put(consumerEdge, output);
- } else {
- NonChainedOutput reusableOutput = null;
- for (NonChainedOutput outputCandidate : outputsConsumedByEdge.values()) {
- // the target output can be reused if they have the same partitioner and
- // consumer parallelism, reusing the same output can improve performance
- if (outputCandidate.getPartitionType().isReconsumable()
- && consumerParallelism == outputCandidate.getConsumerParallelism()
- && consumerMaxParallelism == outputCandidate.getConsumerMaxParallelism()
- && outputCandidate.getPartitionType() == partitionType
- && Objects.equals(
- outputCandidate.getPersistentDataSetId(),
- consumerEdge.getIntermediateDatasetIdToProduce())
- && Objects.equals(
- outputCandidate.getOutputTag(), consumerEdge.getOutputTag())
- && Objects.equals(partitioner, outputCandidate.getPartitioner())) {
- reusableOutput = outputCandidate;
- outputsConsumedByEdge.put(consumerEdge, reusableOutput);
- break;
- }
- }
- if (reusableOutput == null) {
- outputs.add(output);
- outputsConsumedByEdge.put(consumerEdge, output);
- }
- }
+ outputs.add(output);
+ outputsConsumedByEdge.put(consumerEdge, output);
}
- return outputs;
+ }
+
+ private boolean isPartitionTypeCanBeReuse(ResultPartitionType partitionType) {
+ // for non-hybrid partition, partition reuse only works when its re-consumable.
+ // for hybrid selective partition, it still has the opportunity to be converted to
+ // hybrid full partition to support partition reuse.
+ return partitionType.isReconsumable() || partitionType.isHybridResultPartition();
+ }
+
+ private boolean allHybridOrSameReconsumablePartitionType(
+ ResultPartitionType partitionType1, ResultPartitionType partitionType2) {
+ return (partitionType1.isReconsumable() && partitionType1 == partitionType2)
+ || (partitionType1.isHybridResultPartition()
+ && partitionType2.isHybridResultPartition());
}
private void tryConvertPartitionerForDynamicGraph(
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 3dbecaaebb8..a58a52dc536 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
@@ -1811,6 +1811,116 @@ class StreamingJobGraphGeneratorTest {
.isEqualTo(ResultPartitionType.HYBRID_FULL);
}
+ @Test
+ void testHybridPartitionReuse() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ DataStream<Integer> source = env.fromElements(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+
+ DataStream<Integer> partition1 =
+ new DataStream<>(
+ env,
+ new PartitionTransformation<>(
+ source.getTransformation(),
+ new RebalancePartitioner<>(),
+ StreamExchangeMode.HYBRID_FULL));
+ DataStream<Integer> partition2 =
+ new DataStream<>(
+ env,
+ new PartitionTransformation<>(
+ source.getTransformation(),
+ new RebalancePartitioner<>(),
+ StreamExchangeMode.HYBRID_SELECTIVE));
+ DataStream<Integer> partition3 =
+ new DataStream<>(
+ env,
+ new PartitionTransformation<>(
+ source.getTransformation(),
+ new RebalancePartitioner<>(),
+ StreamExchangeMode.HYBRID_SELECTIVE));
+ DataStream<Integer> partition4 =
+ new DataStream<>(
+ env,
+ new PartitionTransformation<>(
+ source.getTransformation(),
+ new RescalePartitioner<>(),
+ StreamExchangeMode.HYBRID_FULL));
+ DataStream<Integer> partition5 =
+ new DataStream<>(
+ env,
+ new PartitionTransformation<>(
+ source.getTransformation(),
+ new RescalePartitioner<>(),
+ StreamExchangeMode.BATCH));
+ DataStream<Integer> partition7 =
+ new DataStream<>(
+ env,
+ new PartitionTransformation<>(
+ source.getTransformation(),
+ new ForwardPartitioner<>(),
+ StreamExchangeMode.HYBRID_FULL));
+ // these two vertices can reuse the same intermediate dataset
+ partition1.addSink(new DiscardingSink<>()).setParallelism(2).name("sink1");
+ partition2.addSink(new DiscardingSink<>()).setParallelism(2).name("sink2");
+
+ // this can not reuse the same intermediate dataset because of different parallelism
+ partition3.addSink(new DiscardingSink<>()).setParallelism(3);
+
+ // this can not reuse the same intermediate dataset because of different partitioner
+ partition4.addSink(new DiscardingSink<>()).setParallelism(2);
+
+ // this can not reuse the same intermediate dataset because of different result partition
+ // type
+ partition5.addSink(new DiscardingSink<>()).setParallelism(2);
+
+ SingleOutputStreamOperator<Integer> mapStream =
+ partition7.map(value -> value).setParallelism(1);
+ DataStream<Integer> mapPartition =
+ new DataStream<>(
+ env,
+ new PartitionTransformation<>(
+ mapStream.getTransformation(),
+ new RescalePartitioner<>(),
+ StreamExchangeMode.HYBRID_FULL));
+ mapPartition.addSink(new DiscardingSink<>()).name("sink3");
+
+ StreamGraph streamGraph = env.getStreamGraph();
+ JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+
+ List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
+ assertThat(vertices).hasSize(7);
+
+ JobVertex sourceVertex = vertices.get(0);
+ List<IntermediateDataSetID> producedDataSet =
+ sourceVertex.getProducedDataSets().stream()
+ .map(IntermediateDataSet::getId)
+ .collect(Collectors.toList());
+ assertThat(producedDataSet).hasSize(5);
+
+ JobVertex sinkVertex1 = checkNotNull(findJobVertexWithName(vertices, "sink1"));
+ JobVertex sinkVertex2 = checkNotNull(findJobVertexWithName(vertices, "sink2"));
+ JobVertex sinkVertex3 = checkNotNull(findJobVertexWithName(vertices, "sink3"));
+
+ assertThat(sinkVertex2.getInputs().get(0).getSource().getId())
+ .isEqualTo(sinkVertex1.getInputs().get(0).getSource().getId());
+ assertThat(sinkVertex3.getInputs().get(0).getSource().getId())
+ .isNotEqualTo(sinkVertex1.getInputs().get(0).getSource().getId());
+
+ StreamConfig streamConfig = new StreamConfig(sourceVertex.getConfiguration());
+ List<IntermediateDataSetID> nonChainedOutputs =
+ streamConfig.getOperatorNonChainedOutputs(getClass().getClassLoader()).stream()
+ .map(NonChainedOutput::getDataSetId)
+ .collect(Collectors.toList());
+ assertThat(nonChainedOutputs).hasSize(4);
+
+ List<IntermediateDataSetID> streamOutputsInOrder =
+ streamConfig.getVertexNonChainedOutputs(getClass().getClassLoader()).stream()
+ .map(NonChainedOutput::getDataSetId)
+ .collect(Collectors.toList());
+ assertThat(streamOutputsInOrder).hasSize(5);
+ assertThat(streamOutputsInOrder).isEqualTo(producedDataSet);
+ }
+
/**
* Tests that multiple downstream consumer vertices can reuse the same intermediate blocking
* dataset if they have the same parallelism and partitioner.