[FLINK-31706] [runtime] The default source parallelism should be the same as ex… #22555

xiaochen-zhou · 2023-05-09T15:45:52Z

What is the purpose of the change

Currently, the sources need to set execution.batch.adaptive.auto-parallelism.default-source-parallelism in the adaptive batch scheduler mode, otherwise, the source parallelism is only 1 by default. A better solution might be set as the default execution parallelism if no user configured.

Brief change log

Modified the default value of source-parallelism to be more reasonable.

Verifying this change

This change is already covered by existing tests, such as AdaptiveBatchSchedulerTest

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)

…ecution's default parallelism

flinkbot · 2023-05-09T15:54:27Z

CI report:

a9cf6e6 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

xiaochen-zhou · 2023-05-10T01:08:50Z

can you spare some precious time to review? Thanks very much @Myasuka

JunRuiLee · 2023-05-10T12:10:56Z

@clownxc Thanks for creating this pr! I'm curious why you don't fallback the value of globalDefaultSourceParallelism to parallelism.default if necessary outside the constructor of DefaultVertexParallelismAndInputInfosDecider? This change will be much cleaner.
In addition, we do not require that parallelism.default must be a positive number. On the contrary, before version 1.17, users need to configure parallelism.default = -1 to enable parallelism derivation. WDYT?

JunRuiLee

@clownxc I've left some comments, PTAL!

JunRuiLee · 2023-05-10T12:37:13Z

flink-core/src/main/java/org/apache/flink/configuration/BatchExecutionOptions.java

@@ -99,7 +99,7 @@ public class BatchExecutionOptions {
    public static final ConfigOption<Integer> ADAPTIVE_AUTO_PARALLELISM_DEFAULT_SOURCE_PARALLELISM =
            key("execution.batch.adaptive.auto-parallelism.default-source-parallelism")
                    .intType()
-                    .defaultValue(1)
+                    .defaultValue(0)
                    .withDeprecatedKeys(


Why do you need to change the default value? I think a better way is through configuration.getOptional(BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_DEFAULT_SOURCE_PARALLELISM))
to determine whether the configuration item has been configured

JunRuiLee · 2023-05-10T12:37:55Z

...ache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java

        checkNotNull(dataVolumePerTask);
+        checkArgument(
+                defaultExecutionParallelism > 0,
+                "The default execution parallelism must be larger than 0.");



We do not require that the default parallelism must be greater than 0

JunRuiLee · 2023-05-10T12:38:57Z

...ache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java

@@ -84,21 +85,28 @@ private DefaultVertexParallelismAndInputInfosDecider(
            int globalMaxParallelism,
            int globalMinParallelism,
            MemorySize dataVolumePerTask,
-            int globalDefaultSourceParallelism) {
+            int globalDefaultSourceParallelism,
+            int defaultExecutionParallelism) {



Maybe we can determine the value of source parallelism externally without touching this method.

JunRuiLee · 2023-05-10T12:40:08Z

...ache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java

-                        BatchExecutionOptions
-                                .ADAPTIVE_AUTO_PARALLELISM_DEFAULT_SOURCE_PARALLELISM));
+                        BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_DEFAULT_SOURCE_PARALLELISM),
+                configuration.get(CoreOptions.DEFAULT_PARALLELISM));
    }


parallelism.default should be obtained from ExecutionConfig instead of jobmaster Configuration

parallelism.default should be obtained from ExecutionConfig instead of jobmaster Configuration

Thank you very much for review. I want to get the parallelism of ExecutionConfig as the default value.

public int getParallelism() { return configuration.get(CoreOptions.DEFAULT_PARALLELISM); }

parallelism.default should be obtained from ExecutionConfig instead of jobmaster Configuration

I don't know if my idea is correct, looking forward to your reply.

xiaochen-zhou · 2023-05-10T15:02:38Z

@clownxc Thanks for creating this pr! I'm curious why you don't fallback the value of globalDefaultSourceParallelism to parallelism.default if necessary outside the constructor of DefaultVertexParallelismAndInputInfosDecider? This change will be much cleaner. In addition, we do not require that parallelism.default must be a positive number. On the contrary, before version 1.17, users need to configure parallelism.default = -1 to enable parallelism derivation. WDYT?

Thank you very much for review, I will try to modify the code as you say

xiaochen-zhou · 2023-05-10T15:34:35Z

@clownxc I've left some comments, PTAL!

Thank you very much for your review, I have modified the code, can you re-review the code when you are free, and make some comments.

JunRuiLee · 2023-05-11T09:06:57Z

@clownxc Thanks for updating. I'm sorry that I overlooked that this pr will lead to a change in the default behavior of the configuration ADAPTIVE_AUTO_PARALLELISM_DEFAULT_SOURCE_PARALLELISM, and this kind of change needs to initiate a discussion in the dev mail list.
@wanglijie95 Could you help to have a look? WDYT?

Myasuka · 2023-05-11T12:37:16Z

@clownxc Please move to the original ticket. Any changes to break the previous behavior needs a discussion in the JIRA, that's why we did not assign anyone to take the ticket.
BTW, we think this change should be reasonable indeed, but a discussion is necessary.

ClownXC added 2 commits May 9, 2023 23:18

[FLINK-31706] The default source parallelism should be the same as ex…

f510468

…ecution's default parallelism

update test

49c1f04

format violations

f99baa0

flinkbot added the component=Runtime/Coordination label May 9, 2023

JunRuiLee suggested changes May 10, 2023

View reviewed changes

ClownXC added 3 commits May 10, 2023 23:17

update code

e858358

update code

de0bf7d

update code

b20c8cc

ClownXC added 2 commits May 10, 2023 23:48

update code

cd5f5cb

update code

9e58e80

style

a9cf6e6

xiaochen-zhou closed this May 15, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-31706] [runtime] The default source parallelism should be the same as ex… #22555

[FLINK-31706] [runtime] The default source parallelism should be the same as ex… #22555

xiaochen-zhou commented May 9, 2023

flinkbot commented May 9, 2023 •

edited

xiaochen-zhou commented May 10, 2023

JunRuiLee commented May 10, 2023 •

edited

JunRuiLee left a comment

JunRuiLee May 10, 2023

JunRuiLee May 10, 2023

JunRuiLee May 10, 2023

JunRuiLee May 10, 2023

xiaochen-zhou May 10, 2023 •

edited

xiaochen-zhou May 10, 2023

xiaochen-zhou commented May 10, 2023

xiaochen-zhou commented May 10, 2023

JunRuiLee commented May 11, 2023

Myasuka commented May 11, 2023

[FLINK-31706] [runtime] The default source parallelism should be the same as ex… #22555

[FLINK-31706] [runtime] The default source parallelism should be the same as ex… #22555

Conversation

xiaochen-zhou commented May 9, 2023

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented May 9, 2023 • edited

CI report:

xiaochen-zhou commented May 10, 2023

JunRuiLee commented May 10, 2023 • edited

JunRuiLee left a comment

Choose a reason for hiding this comment

JunRuiLee May 10, 2023

Choose a reason for hiding this comment

JunRuiLee May 10, 2023

Choose a reason for hiding this comment

JunRuiLee May 10, 2023

Choose a reason for hiding this comment

JunRuiLee May 10, 2023

Choose a reason for hiding this comment

xiaochen-zhou May 10, 2023 • edited

Choose a reason for hiding this comment

xiaochen-zhou May 10, 2023

Choose a reason for hiding this comment

xiaochen-zhou commented May 10, 2023

xiaochen-zhou commented May 10, 2023

JunRuiLee commented May 11, 2023

Myasuka commented May 11, 2023

flinkbot commented May 9, 2023 •

edited

JunRuiLee commented May 10, 2023 •

edited

xiaochen-zhou May 10, 2023 •

edited