You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/10/14 05:14:13 UTC
(paimon) branch master updated: [flink] fix compact unaware bucket table in combine mode the writer parallelism is always 1 (#4309)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3a7f79024 [flink] fix compact unaware bucket table in combine mode the writer parallelism is always 1 (#4309)
3a7f79024 is described below
commit 3a7f790244d6cfd57f4d0d3f8135ab71449ac33b
Author: Tan-JiaLiang <ta...@gmail.com>
AuthorDate: Mon Oct 14 13:14:08 2024 +0800
[flink] fix compact unaware bucket table in combine mode the writer parallelism is always 1 (#4309)
---
.../paimon/flink/action/CompactDatabaseAction.java | 18 +++++++++++++-----
.../paimon/flink/sink/FlinkStreamPartitioner.java | 11 +++++++++++
.../operator/CombinedUnawareBatchSourceFunction.java | 8 +-------
.../CombinedUnawareStreamingSourceFunction.java | 3 +--
4 files changed, 26 insertions(+), 14 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
index ef6772e36..fda9ff695 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
@@ -54,6 +54,7 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import static org.apache.paimon.flink.sink.FlinkStreamPartitioner.partition;
+import static org.apache.paimon.flink.sink.FlinkStreamPartitioner.rebalance;
/** Database compact action for Flink. */
public class CompactDatabaseAction extends ActionBase {
@@ -208,6 +209,11 @@ public class CompactDatabaseAction extends ActionBase {
.toMillis())
.withPartitionIdleTime(partitionIdleTime);
+ Integer parallelism =
+ tableOptions.get(FlinkConnectorOptions.SINK_PARALLELISM) == null
+ ? env.getParallelism()
+ : tableOptions.get(FlinkConnectorOptions.SINK_PARALLELISM);
+
// multi bucket table which has multi bucket in a partition like fix bucket and dynamic
// bucket
DataStream<RowData> awareBucketTableSource =
@@ -217,14 +223,16 @@ public class CompactDatabaseAction extends ActionBase {
.withContinuousMode(isStreaming)
.buildAwareBucketTableSource(),
new BucketsRowChannelComputer(),
- tableOptions.get(FlinkConnectorOptions.SINK_PARALLELISM));
+ parallelism);
// unaware bucket table
DataStream<MultiTableUnawareAppendCompactionTask> unawareBucketTableSource =
- sourceBuilder
- .withEnv(env)
- .withContinuousMode(isStreaming)
- .buildForUnawareBucketsTableSource();
+ rebalance(
+ sourceBuilder
+ .withEnv(env)
+ .withContinuousMode(isStreaming)
+ .buildForUnawareBucketsTableSource(),
+ parallelism);
new CombinedTableCompactorSink(catalogLoader(), tableOptions)
.sinkFrom(awareBucketTableSource, unawareBucketTableSource);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkStreamPartitioner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkStreamPartitioner.java
index 78e532053..f9b81760c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkStreamPartitioner.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkStreamPartitioner.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.io.network.api.writer.SubtaskStateMapper;
import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
+import org.apache.flink.streaming.runtime.partitioner.RebalancePartitioner;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -77,4 +78,14 @@ public class FlinkStreamPartitioner<T> extends StreamPartitioner<T> {
}
return new DataStream<>(input.getExecutionEnvironment(), partitioned);
}
+
+ public static <T> DataStream<T> rebalance(DataStream<T> input, Integer parallelism) {
+ RebalancePartitioner<T> partitioner = new RebalancePartitioner<>();
+ PartitionTransformation<T> partitioned =
+ new PartitionTransformation<>(input.getTransformation(), partitioner);
+ if (parallelism != null) {
+ partitioned.setParallelism(parallelism);
+ }
+ return new DataStream<>(input.getExecutionEnvironment(), partitioned);
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSourceFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSourceFunction.java
index b0b5a7784..8ec8d5f2c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSourceFunction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSourceFunction.java
@@ -36,8 +36,6 @@ import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.StreamSource;
-import org.apache.flink.streaming.api.transformations.PartitionTransformation;
-import org.apache.flink.streaming.runtime.partitioner.RebalancePartitioner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -135,11 +133,7 @@ public class CombinedUnawareBatchSourceFunction
new MultiUnawareTablesReadOperator(catalogLoader, partitionIdleTime));
}
- PartitionTransformation<MultiTableUnawareAppendCompactionTask> transformation =
- new PartitionTransformation<>(
- source.getTransformation(), new RebalancePartitioner<>());
-
- return new DataStream<>(env, transformation);
+ return source;
}
private static Long getPartitionInfo(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSourceFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSourceFunction.java
index 54a90ac36..e398e09a8 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSourceFunction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSourceFunction.java
@@ -110,8 +110,7 @@ public class CombinedUnawareStreamingSourceFunction
isParallel,
name,
Boundedness.CONTINUOUS_UNBOUNDED)
- .forceNonParallel()
- .rebalance();
+ .forceNonParallel();
}
@Override