[Fix][Spark] Fix source parallelism not working with Spark engine#9319
Conversation
| envOption.put(EnvCommonOptions.PARALLELISM.key(), String.valueOf(parallelism)); | ||
| Dataset<Row> dataset = | ||
| sparkRuntimeEnvironment | ||
| .getSparkSession() | ||
| .read() | ||
| .format(SeaTunnelSource.class.getSimpleName()) | ||
| .option(EnvCommonOptions.PARALLELISM.key(), parallelism) | ||
| .option( |
There was a problem hiding this comment.
Thanks @joexjx . But in seatunnel, we recommend configuring parallelism in env rather than source configuration. This helps improve configuration consistency across different engines. Maybe you can help update the documentation to make the description clearer.
There was a problem hiding this comment.
I understand. However, this code does contain a very simple and obvious error. envOption is a HashMap that already includes EnvCommonOptions.PARALLELISM.key(). The .option(EnvCommonOptions.PARALLELISM.key(), parallelism) call is being overwritten by .options(envOption), which is why the properly configured parallelism (determined earlier based on whether it comes from env or source) isn't taking effect.
Old Code
New Code
There was a problem hiding this comment.
I removed .option(EnvCommonOptions.PARALLELISM.key(), ...) and updated envOptions with EnvCommonOptions.PARALLELISM in advance. This ensures that the parallelism value (which was already determined earlier based on whether it should be taken from env or source) takes effect correctly.
|
Seem like the commit history mess. Please rebase on dev. |
6aaaba8 to
a27d20c
Compare
| @Order(1) | ||
| public class SingleSplitSourceParallelismTest { | ||
| @Test | ||
| public void testSourceParallelismIsOneButEnvParallelismIsNotOne() |
There was a problem hiding this comment.
The current test case is a bit overly complex. We just need to ensure that the DataSet returned by the execute method of SourceExecuteProcessor has the expected parallelism.
SourceExecuteProcessor processor = new SourceExecuteProcessor();
... init
Assertions.equals(10, processor.execute(xxx).get(0).getDataset().rdd().getNumPartitions());
There was a problem hiding this comment.
@Hisoka-X Thank you for the guidance, I have added a simpler unit test.
|
waiting test case passes. |
…ng with Spark Engine (apache#9302) This bug was caused by EnvCommonOptions overriding SourceCommonOptions when setting the parallelism in the sparkRuntimeEnvironment.
…sm Overriding Environment Parallelism in Spark Engine.
…lelism Config Works And Overrides Env Config
close: #9302
This bug was caused by EnvCommonOptions overriding SourceCommonOptions when setting the parallelism in the sparkRuntimeEnvironment.
Purpose of this pull request
Does this PR introduce any user-facing change?
How was this patch tested?
Check list
New License Guide