[flink] Improve the parallelism of Flink database compaction commit o…

…perator (#3789)
apache · Jul 24, 2024 · 6f4aa51 · 6f4aa51
1 parent 094e925
commit 6f4aa51
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 5 deletions.
diff --git a/...imon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java b/...imon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
@@ -25,6 +25,7 @@
 import org.apache.paimon.flink.sink.FlinkSink;
 import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
 import org.apache.paimon.flink.sink.MultiTableCommittable;
+import org.apache.paimon.flink.sink.MultiTableCommittableChannelComputer;
 import org.apache.paimon.flink.sink.MultiTableCommittableTypeInfo;
 import org.apache.paimon.flink.sink.RestoreAndFailCommittableStateManager;
 import org.apache.paimon.flink.sink.StoreMultiCommitter;

diff --git a/...n-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java b/...n-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java
@@ -142,8 +142,15 @@ protected DataStreamSink<?> doCommit(
         if (streamingCheckpointEnabled) {
             assertStreamingConfiguration(env);
         }
+
+        DataStream<MultiTableCommittable> partitioned =
+                FlinkStreamPartitioner.partition(
+                        written,
+                        new MultiTableCommittableChannelComputer(),
+                        written.getParallelism());
         SingleOutputStreamOperator<?> committed =
-                written.transform(
+                partitioned
+                        .transform(
                                 GLOBAL_COMMITTER_NAME,
                                 new MultiTableCommittableTypeInfo(),
                                 new CommitterOperator<>(
@@ -154,8 +161,7 @@ protected DataStreamSink<?> doCommit(
                                         createCommitterFactory(),
                                         createCommittableStateManager(),
                                         options.get(END_INPUT_WATERMARK)))
-                        .setParallelism(1)
-                        .setMaxParallelism(1);
+                        .setParallelism(written.getParallelism());
         return committed.addSink(new DiscardingSink<>()).name("end").setParallelism(1);
     }
 

diff --git a/...MultiTableCommittableChannelComputer.java → ...MultiTableCommittableChannelComputer.java b/...MultiTableCommittableChannelComputer.java → ...MultiTableCommittableChannelComputer.java
@@ -16,9 +16,8 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.sink.cdc;
+package org.apache.paimon.flink.sink;
 
-import org.apache.paimon.flink.sink.MultiTableCommittable;
 import org.apache.paimon.table.sink.ChannelComputer;
 
 import java.util.Objects;