[Feature][CONNECTORS-V2-Paimon] flink task parallelism

apache · Sep 12, 2024 · ad8281b · ad8281b
1 parent cbe8a22
commit ad8281b
Show file tree

Hide file tree

Showing 6 changed files with 38 additions and 11 deletions.
diff --git a/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java b/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java
@@ -56,13 +56,16 @@ public class SinkExecuteProcessor
         extends FlinkAbstractPluginExecuteProcessor<Optional<? extends Factory>> {
 
     private static final String PLUGIN_TYPE = PluginType.SINK.getType();
+    private final int parallelism;
 
     protected SinkExecuteProcessor(
             List<URL> jarPaths,
             Config envConfig,
             List<? extends Config> pluginConfigs,
-            JobContext jobContext) {
+            JobContext jobContext,
+            int parallelism) {
         super(jarPaths, envConfig, pluginConfigs, jobContext);
+        this.parallelism = parallelism;
     }
 
     @Override
@@ -139,10 +142,11 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
             }
             DataStreamSink<Row> dataStreamSink =
                     stream.getDataStream()
-                            .sinkTo(new FlinkSink<>(sink, stream.getCatalogTables().get(0)))
+                            .sinkTo(
+                                    new FlinkSink<>(
+                                            sink, stream.getCatalogTables().get(0), parallelism))
                             .name(String.format("%s-Sink", sink.getPluginName()));
             if (sinkConfig.hasPath(CommonOptions.PARALLELISM.key())) {
-                int parallelism = sinkConfig.getInt(CommonOptions.PARALLELISM.key());
                 dataStreamSink.setParallelism(parallelism);
             }
         }

diff --git a/...ommon/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java b/...ommon/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java
@@ -49,6 +49,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -85,7 +86,7 @@ public FlinkExecution(Config config) {
         registerPlugin(envConfig);
         JobContext jobContext = new JobContext();
         jobContext.setJobMode(RuntimeEnvironment.getJobMode(config));
-
+        int parallelism = Optional.of(envConfig.getInt("parallelism")).orElse(1);
         this.sourcePluginExecuteProcessor =
                 new SourceExecuteProcessor(
                         jarPaths, envConfig, config.getConfigList(Constants.SOURCE), jobContext);
@@ -98,7 +99,11 @@ public FlinkExecution(Config config) {
                         jobContext);
         this.sinkPluginExecuteProcessor =
                 new SinkExecuteProcessor(
-                        jarPaths, envConfig, config.getConfigList(Constants.SINK), jobContext);
+                        jarPaths,
+                        envConfig,
+                        config.getConfigList(Constants.SINK),
+                        jobContext,
+                        parallelism);
 
         this.flinkRuntimeEnvironment =
                 FlinkRuntimeEnvironment.getInstance(this.registerPlugin(config, jarPaths));

diff --git a/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java b/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java
@@ -57,13 +57,16 @@ public class SinkExecuteProcessor
         extends FlinkAbstractPluginExecuteProcessor<Optional<? extends Factory>> {
 
     private static final String PLUGIN_TYPE = PluginType.SINK.getType();
+    private final int parallelism;
 
     protected SinkExecuteProcessor(
             List<URL> jarPaths,
             Config envConfig,
             List<? extends Config> pluginConfigs,
-            JobContext jobContext) {
+            JobContext jobContext,
+            int parallelism) {
         super(jarPaths, envConfig, pluginConfigs, jobContext);
+        this.parallelism = parallelism;
     }
 
     @Override
@@ -143,7 +146,9 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
                             .sinkTo(
                                     SinkV1Adapter.wrap(
                                             new FlinkSink<>(
-                                                    sink, stream.getCatalogTables().get(0))))
+                                                    sink,
+                                                    stream.getCatalogTables().get(0),
+                                                    parallelism)))
                             .name(String.format("%s-Sink", sink.getPluginName()));
             if (sinkConfig.hasPath(CommonOptions.PARALLELISM.key())) {
                 int parallelism = sinkConfig.getInt(CommonOptions.PARALLELISM.key());

diff --git a/...-13/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriterContext.java b/...-13/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriterContext.java
@@ -39,17 +39,24 @@ public class FlinkSinkWriterContext implements SinkWriter.Context {
 
     private final InitContext writerContext;
     private final EventListener eventListener;
+    private final int parallelism;
 
-    public FlinkSinkWriterContext(InitContext writerContext) {
+    public FlinkSinkWriterContext(InitContext writerContext, int parallelism) {
         this.writerContext = writerContext;
         this.eventListener = new DefaultEventProcessor(getJobIdForV14(writerContext));
+        this.parallelism = parallelism;
     }
 
     @Override
     public int getIndexOfSubtask() {
         return writerContext.getSubtaskId();
     }
 
+    @Override
+    public int getNumberOfParallelSubtasks() {
+        return parallelism;
+    }
+
     @Override
     public MetricsContext getMetricsContext() {
         try {

diff --git a/...ion-flink-common/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSink.java b/...ion-flink-common/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSink.java
@@ -50,19 +50,23 @@ public class FlinkSink<InputT, CommT, WriterStateT, GlobalCommT>
 
     private final CatalogTable catalogTable;
 
+    private final int parallelism;
+
     public FlinkSink(
             SeaTunnelSink<SeaTunnelRow, WriterStateT, CommT, GlobalCommT> sink,
-            CatalogTable catalogTable) {
+            CatalogTable catalogTable,
+            int parallelism) {
         this.sink = sink;
         this.catalogTable = catalogTable;
+        this.parallelism = parallelism;
     }
 
     @Override
     public SinkWriter<InputT, CommitWrapper<CommT>, FlinkWriterState<WriterStateT>> createWriter(
             Sink.InitContext context, List<FlinkWriterState<WriterStateT>> states)
             throws IOException {
         org.apache.seatunnel.api.sink.SinkWriter.Context stContext =
-                new FlinkSinkWriterContext(context);
+                new FlinkSinkWriterContext(context, parallelism);
         if (states == null || states.isEmpty()) {
             return new FlinkSinkWriter<>(
                     sink.createWriter(stContext), 1, catalogTable.getSeaTunnelRowType(), stContext);

diff --git a/...mon/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriterContext.java b/...mon/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriterContext.java
@@ -36,10 +36,12 @@ public class FlinkSinkWriterContext implements SinkWriter.Context {
 
     private final Sink.InitContext writerContext;
     private final EventListener eventListener;
+    private final int parallelism;
 
-    public FlinkSinkWriterContext(InitContext writerContext) {
+    public FlinkSinkWriterContext(InitContext writerContext, int parallelism) {
         this.writerContext = writerContext;
         this.eventListener = new DefaultEventProcessor(getFlinkJobId(writerContext));
+        this.parallelism = parallelism;
     }
 
     @Override