[Feature][Zeta] Support Parallelism Inference #10102
Open
+1,299
−14
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Comment on lines
+211
to
+234
| **服务器级别配置示例** | ||
|
|
||
| 在 `seatunnel.yaml` 中配置： | ||
|
|
||
| ```yaml | ||
| seatunnel: | ||
| engine: | ||
| parallelism-inference: | ||
| enabled: true | ||
| max-parallelism: 100 | ||
| ``` | ||
|
|
||
| **作业级别配置示例** | ||
|
|
||
| 在作业配置文件的 `env` 块中配置： | ||
|
|
||
| ```hocon | ||
| env { | ||
| # 启用并行度推断 | ||
| parallelism.inference.enabled = true | ||
| # 设置最大并行度 | ||
| parallelism.inference.max-parallelism = 50 | ||
| } | ||
Comment on lines
187
to
211
| @Override | ||
| public int inferParallelism() { | ||
| try { | ||
| for (Map.Entry<String, ReadBuilder> entry : readBuilders.entrySet()) { | ||
| String tableKey = entry.getKey(); | ||
| ReadBuilder readBuilder = entry.getValue(); | ||
| try { | ||
| List<PartitionEntry> partitionEntries = | ||
| readBuilder.newScan().listPartitionEntries(); | ||
| return !partitionEntries.isEmpty() ? partitionEntries.size() : 1; | ||
| } catch (Exception e) { | ||
| log.warn( | ||
| "Failed to get partition info for table {}, skipping parallelism inference", | ||
| tableKey, | ||
| e); | ||
| return -1; | ||
| } | ||
| } | ||
|
|
||
| } catch (Exception e) { | ||
| log.warn("Failed to infer parallelism for Paimon source", e); | ||
| return -1; | ||
| } | ||
| return 1; | ||
| } |
Member
There was a problem hiding this comment.
Only take the first table? Could you explain the design?
Comment on lines
126
to
128
| "Enable automatic parallelism inference based on data volume. " | ||
| + "When enabled, operators with parallelism=-1 will have their parallelism " | ||
| + "automatically determined based on the size of consumed data."); |
Member
There was a problem hiding this comment.
The described logic has not been implemented
Comment on lines
135
to
136
| "The maximum parallelism for operators when using automatic inference. " | ||
| + "Must be a power of 2 to ensure even distribution of subpartitions."); |
| .orElse(parallelismInferenceConfig.isEnabled()); | ||
|
|
||
| if (inferenceEnabled && source instanceof SupportParallelismInference) { | ||
| int inferredParallelism = ((SupportParallelismInference) source).inferParallelism(); |
Member
There was a problem hiding this comment.
What if we roll back the parallelism to the default if it has issues?
| public static Option<Integer> PARALLELISM_INFERENCE_MAX_PARALLELISM = | ||
| Options.key("parallelism.inference.max-parallelism") | ||
| .intType() | ||
| .defaultValue(128) |
Member
There was a problem hiding this comment.
Why is it 128 here? The document describes it as 64.
|
|
||
| // ==================== Parallelism Inference Options ==================== | ||
|
|
||
| public static Option<Boolean> PARALLELISM_INFERENCE_ENABLED = |
Member
There was a problem hiding this comment.
New parameters need to be added in EnvOptionRule
Contributor
Author
|
I have updated the code based on the review suggestions. Please take a look when you have time. @zhangshenghang |
Comment on lines
+188
to
+212
| public int inferParallelism() { | ||
| int inferParallelism = 0; | ||
| try { | ||
| for (Map.Entry<String, ReadBuilder> entry : readBuilders.entrySet()) { | ||
| String tableKey = entry.getKey(); | ||
| ReadBuilder readBuilder = entry.getValue(); | ||
| try { | ||
| List<PartitionEntry> partitionEntries = | ||
| readBuilder.newScan().listPartitionEntries(); | ||
| inferParallelism += partitionEntries.size(); | ||
| } catch (Exception e) { | ||
| log.warn( | ||
| "Failed to get partition info for table {}, skipping parallelism inference", | ||
| tableKey, | ||
| e); | ||
| return -1; | ||
| } | ||
| } | ||
|
|
||
| } catch (Exception e) { | ||
| log.warn("Failed to infer parallelism for Paimon source", e); | ||
| return -1; | ||
| } | ||
| return inferParallelism <= 0 ? 1 : inferParallelism; | ||
| } |
Member
There was a problem hiding this comment.
Is there a problem with this design? Is there any recommended calculation rule for the parallelism of our Sink under normal circumstances? @Hisoka-X
| public class ParallelismInferenceConfig implements Serializable { | ||
| private static final long serialVersionUID = 1L; | ||
| private boolean enabled = false; | ||
| private int maxParallelism = 128; |
Member
There was a problem hiding this comment.
Should it be consistent with the above here? It should be 64
Contributor
Author
There was a problem hiding this comment.
Should it be consistent with the above here? It should be 64
Done.
Comment on lines
134
to
135
| // Expected: 2 (table1) + 3 (table2) + 1 (table3) = 7 | ||
| Assertions.assertEquals(6, parallelism); |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Purpose of this pull request
Supports automatic parallelism inference for sources that implement the
SupportParallelismInferenceinterface (e.g., Paimon connector). When enabled, the engine will automatically determine the optimal parallelism based on data characteristics instead of using default parallelism value: 1Does this PR introduce any user-facing change?
Yes.
How was this patch tested?
Add new tests
Check list
New License Guide
incompatible-changes.mdto describe the incompatibility caused by this PR.