Hi Xia and Jinrui, Filed https://github.com/apache/flink/pull/24736 to address the above described issue. Please take a look whenever you can.
Thanks Venkat On Thu, Apr 18, 2024 at 12:16 PM Venkatakrishnan Sowrirajan < vsowr...@asu.edu> wrote: > Filed https://issues.apache.org/jira/browse/FLINK-35165 to address the > above described issue. Will share the PR here once it is ready for review. > > Regards > Venkata krishnan > > > On Wed, Apr 17, 2024 at 5:32 AM Junrui Lee <jrlee....@gmail.com> wrote: > >> Thanks Venkata and Xia for providing further clarification. I think your >> example illustrates the significance of this proposal very well. Please >> feel free go ahead and address the concerns. >> >> Best, >> Junrui >> >> Venkatakrishnan Sowrirajan <vsowr...@asu.edu> 于2024年4月16日周二 07:01写道： >> >> > Thanks for adding your thoughts to this discussion. >> > >> > If we all agree that the source vertex parallelism shouldn't be bound by >> > the downstream max parallelism >> > (jobmanager.adaptive-batch-scheduler.max-parallelism) >> > based on the rationale and the issues described above, I can take a >> stab at >> > addressing the issue. >> > >> > Let me file a ticket to track this issue. Otherwise, I'm looking >> forward to >> > hearing more thoughts from others as well, especially Lijie and Junrui >> who >> > have more context on the AdaptiveBatchScheduler. >> > >> > Regards >> > Venkata krishnan >> > >> > >> > On Mon, Apr 15, 2024 at 12:54 AM Xia Sun <xingbe...@gmail.com> wrote: >> > >> > > Hi Venkat, >> > > I agree that the parallelism of source vertex should not be upper >> bounded >> > > by the job's global max parallelism. The case you mentioned, >> High >> > filter >> > > selectivity with huge amounts of data to read excellently supports >> this >> > > viewpoint. (In fact, in the current implementation, if the source >> > > parallelism is pre-specified at job create stage, rather than relying >> on >> > > the dynamic parallelism inference of the AdaptiveBatchScheduler, the >> > source >> > > vertex's parallelism can indeed exceed the job's global max >> parallelism.) >> > > >> > > As Lijie and Junrui pointed out, the key issue is "semantic >> consistency." >> > > Currently, if a vertex has not set maxParallelism, the >> > > AdaptiveBatchScheduler will use >> > > `execution.batch.adaptive.auto-parallelism.max-parallelism` as the >> > vertex's >> > > maxParallelism. Since the current implementation does not distinguish >> > > between source vertices and downstream vertices, source vertices are >> also >> > > subject to this limitation. >> > > >> > > Therefore, I believe that if the issue of "semantic consistency" can >> be >> > > well explained in the code and configuration documentation, the >> > > AdaptiveBatchScheduler should support that the parallelism of source >> > > vertices can exceed the job's global max parallelism. >> > > >> > > Best, >> > > Xia >> > > >> > > Venkatakrishnan Sowrirajan <vsowr...@asu.edu> 于2024年4月14日周日 10:31写道： >> > > >> > > > Let me state why I think "*jobmanager.adaptive-batch-sche* >> > > > *duler.default-source-parallelism*" should not be bound by the " >> > > > *jobmanager.adaptive-batch-sche**duler.max-parallelism*". >> > > > >> > > > - Source vertex is unique and does not have any upstream vertices >> > > > - Downstream vertices read shuffled data partitioned by key, >> which >> > is >> > > > not the case for the Source vertex >> > > > - Limiting source parallelism by downstream vertices' max >> > parallelism >> > > is >> > > > incorrect >> > > > >> > > > If we say for ""semantic consistency" the source vertex parallelism >> has >> > > to >> > > > be bound by the overall job's max parallelism, it can lead to >> following >> > > > issues: >> > > > >> > > > - High filter selectivity with huge amounts of data to read - >> > setting >> > > > high "*jobmanager.adaptive-batch-scheduler.max-parallelism*" so >> that >> > > > source parallelism can be set higher can lead to small blocks and >> > > > sub-optimal performance. >> > > > - Setting high >> > "*jobmanager.adaptive-batch-scheduler.max-parallelism*" >> > > > requires careful tuning of network buffer configurations which is >> > > > unnecessary in cases where it is not required just so that the >> > source >> > > > parallelism can be set high. >> > > > >> > > > Regards >> > > > Venkata krishnan >> > > > >> > > > On Thu, Apr 11, 2024 at 9:30 PM Junrui Lee <jrlee....@gmail.com> >> > wrote: >> > > > >> > > > > Hello Venkata krishnan, >> > > > > >> > > > > I think the term "semantic inconsistency" defined by >> > > > > jobmanager.adaptive-batch-scheduler.max-parallelism refers to >> > > > maintaining a >> > > > > uniform upper limit on parallelism across all vertices within a >> job. >> > As >> > > > the >> > > > > source vertices are part of the global execution graph, they >> should >> > > also >> > > > > respect this rule to ensure consistent application of parallelism >> > > > > constraints. >> > > > > >> > > > > Best, >> > > > > Junrui >> > > > > >> > > > > Venkatakrishnan Sowrirajan <vsowr...@asu.edu> 于2024年4月12日周五 >> 02:10写道： >> > > > > >> > > > > > Gentle bump on this question. cc @Becket Qin < >> becket....@gmail.com >> > > >> > > as >> > > > > > well. >> > > > > > >> > > > > > Regards >> > > > > > Venkata krishnan >> > > > > > >> > > > > > >> > > > > > On Tue, Mar 12, 2024 at 10:11 PM Venkatakrishnan Sowrirajan < >> > > > > > vsowr...@asu.edu> wrote: >> > > > > > >> > > > > > > Thanks for the response Lijie and Junrui. Sorry for the late >> > reply. >> > > > Few >> > > > > > > follow up questions. >> > > > > > > >> > > > > > > > Source can actually ignore this limit >> > > > > > > because it has no upstream, but this will lead to semantic >> > > > > inconsistency. >> > > > > > > >> > > > > > > Lijie, can you please elaborate on the above comment further? >> > What >> > > do >> > > > > you >> > > > > > > mean when you say it will lead to "semantic inconsistency"? >> > > > > > > >> > > > > > > > Secondly, we first need to limit the max parallelism of >> > > > (downstream) >> > > > > > > vertex, and then we can decide how many subpartitions >> (upstream >> > > > vertex) >> > > > > > > should produce. The limit should be effective, otherwise some >> > > > > downstream >> > > > > > > tasks will have no data to process. >> > > > > > > >> > > > > > > This makes sense in the context of any other vertices other >> than >> > > the >> > > > > > > source vertex. As you mentioned above ("Source can actually >> > ignore >> > > > this >> > > > > > > limit because it has no upstream"), therefore I feel " >> > > > > > > >> jobmanager.adaptive-batch-scheduler.default-source-parallelism" >> > > need >> > > > > not >> > > > > > > be upper bounded by >> > > > > > "jobmanager.adaptive-batch-scheduler.max-parallelism". >> > > > > > > >> > > > > > > Regards >> > > > > > > Venkata krishnan >> > > > > > > >> > > > > > > >> > > > > > > On Thu, Feb 29, 2024 at 2:11 AM Junrui Lee < >> jrlee....@gmail.com> >> > > > > wrote: >> > > > > > > >> > > > > > >> Hi Venkat, >> > > > > > >> >> > > > > > >> As Lijie mentioned, in Flink, the parallelism is required >> to be >> > > > less >> > > > > > than >> > > > > > >> or equal to the maximum parallelism. The config option >> > > > > > >> jobmanager.adaptive-batch-scheduler.max-parallelism and >> > > > > > >> >> jobmanager.adaptive-batch-scheduler.default-source-parallelism >> > > will >> > > > be >> > > > > > set >> > > > > > >> as the source's parallelism and max-parallelism, >> respectively. >> > > > > > Therefore, >> > > > > > >> the check failed situation you encountered is in line with >> the >> > > > > > >> expectations. >> > > > > > >> >> > > > > > >> Best, >> > > > > > >> Junrui >> > > > > > >> >> > > > > > >> Lijie Wang <wangdachui9...@gmail.com> 于2024年2月29日周四 17:35写道： >> > > > > > >> >> > > > > > >> > Hi Venkat, >> > > > > > >> > >> > > > > > >> > >> default-source-parallelism config should be independent >> > from >> > > > the >> > > > > > >> > max-parallelism >> > > > > > >> > >> > > > > > >> > Actually, it's not. >> > > > > > >> > >> > > > > > >> > Firstly, it's obvious that the parallelism should be less >> than >> > > or >> > > > > > equal >> > > > > > >> to >> > > > > > >> > the max parallelism(both literally and execution). The >> > > > > > >> > "jobmanager.adaptive-batch-scheduler.max-parallelism" will >> be >> > > used >> > > > > as >> > > > > > >> the >> > > > > > >> > max parallelism for a vertex if you don't set max >> parallelism >> > > for >> > > > it >> > > > > > >> > individually (Just like the source in your case). >> > > > > > >> > >> > > > > > >> > Secondly, we first need to limit the max parallelism of >> > > > (downstream) >> > > > > > >> > vertex, and then we can decide how many subpartitions >> > (upstream >> > > > > > vertex) >> > > > > > >> > should produce. The limit should be effective, otherwise >> some >> > > > > > downstream >> > > > > > >> > tasks will have no data to process. Source can actually >> ignore >> > > > this >> > > > > > >> limit >> > > > > > >> > because it has no upstream, but this will lead to semantic >> > > > > > >> inconsistency. >> > > > > > >> > >> > > > > > >> > Best, >> > > > > > >> > Lijie >> > > > > > >> > >> > > > > > >> > Venkatakrishnan Sowrirajan <vsowr...@asu.edu> >> 于2024年2月29日周四 >> > > > > 05:49写道： >> > > > > > >> > >> > > > > > >> > > Hi Flink devs, >> > > > > > >> > > >> > > > > > >> > > With Flink's AdaptiveBatchScheduler >> > > > > > >> > > < >> > > > > > >> > > >> > > > > > >> > >> > > > > > >> >> > > > > > >> > > > > >> > > > >> > > >> > >> https://urldefense.com/v3/__https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/elastic_scaling/*adaptive-batch-scheduler__;Iw!!IKRxdwAv5BmarQ!fwD4qP-fTEiqJH9CC3AHgXbR5MJPGm7ll1dYwElK-zrtujWDWio6_yvBa4rHlaZHP_lefLs4bZQAISrg5BrHLw$ >> > > > > > >> > > > >> > > > > > >> > > (Note: >> > > > > > >> > > this is different from AdaptiveScheduler >> > > > > > >> > > < >> > > > > > >> > > >> > > > > > >> > >> > > > > > >> >> > > > > > >> > > > > >> > > > >> > > >> > >> https://urldefense.com/v3/__https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/elastic_scaling/*adaptive-scheduler__;Iw!!IKRxdwAv5BmarQ!fwD4qP-fTEiqJH9CC3AHgXbR5MJPGm7ll1dYwElK-zrtujWDWio6_yvBa4rHlaZHP_lefLs4bZQAISqUzURivw$ >> > > > > > >> > > >), >> > > > > > >> > > the scheduler automatically determines the correct >> number of >> > > > > > >> downstream >> > > > > > >> > > tasks required to process the shuffle generated by the >> > > upstream >> > > > > > >> vertex. >> > > > > > >> > > >> > > > > > >> > > I have a question regarding the current behavior. There >> are >> > 2 >> > > > > > configs >> > > > > > >> > which >> > > > > > >> > > are in interplay here. >> > > > > > >> > > 1. >> > > > jobmanager.adaptive-batch-scheduler.default-source-parallelism >> > > > > > >> > > < >> > > > > > >> > > >> > > > > > >> > >> > > > > > >> >> > > > > > >> > > > > >> > > > >> > > >> > >> https://urldefense.com/v3/__https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/config/*jobmanager-adaptive-batch-scheduler-default-source-parallelism__;Iw!!IKRxdwAv5BmarQ!fwD4qP-fTEiqJH9CC3AHgXbR5MJPGm7ll1dYwElK-zrtujWDWio6_yvBa4rHlaZHP_lefLs4bZQAISoOTMiiCA$ >> > > > > > >> > > > >> > > > > > >> > > - The default parallelism of data source. >> > > > > > >> > > 2. jobmanager.adaptive-batch-scheduler.max-parallelism >> > > > > > >> > > < >> > > > > > >> > > >> > > > > > >> > >> > > > > > >> >> > > > > > >> > > > > >> > > > >> > > >> > >> https://urldefense.com/v3/__https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/config/*jobmanager-adaptive-batch-scheduler-max-parallelism__;Iw!!IKRxdwAv5BmarQ!fwD4qP-fTEiqJH9CC3AHgXbR5MJPGm7ll1dYwElK-zrtujWDWio6_yvBa4rHlaZHP_lefLs4bZQAISpOw_L_Eg$ >> > > > > > >> > > > >> > > > > > >> > > - >> > > > > > >> > > Upper bound of allowed parallelism to set adaptively. >> > > > > > >> > > >> > > > > > >> > > Currently, if " >> > > > > > >> > > >> > jobmanager.adaptive-batch-scheduler.default-source-parallelism >> > > > > > >> > > < >> > > > > > >> > > >> > > > > > >> > >> > > > > > >> >> > > > > > >> > > > > >> > > > >> > > >> > >> https://urldefense.com/v3/__https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/config/*jobmanager-adaptive-batch-scheduler-default-source-parallelism__;Iw!!IKRxdwAv5BmarQ!fwD4qP-fTEiqJH9CC3AHgXbR5MJPGm7ll1dYwElK-zrtujWDWio6_yvBa4rHlaZHP_lefLs4bZQAISoOTMiiCA$ >> > > > > > >> > > >" >> > > > > > >> > > is greater than >> > > > > "jobmanager.adaptive-batch-scheduler.max-parallelism >> > > > > > >> > > < >> > > > > > >> > > >> > > > > > >> > >> > > > > > >> >> > > > > > >> > > > > >> > > > >> > > >> > >> https://urldefense.com/v3/__https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/config/*jobmanager-adaptive-batch-scheduler-max-parallelism__;Iw!!IKRxdwAv5BmarQ!fwD4qP-fTEiqJH9CC3AHgXbR5MJPGm7ll1dYwElK-zrtujWDWio6_yvBa4rHlaZHP_lefLs4bZQAISpOw_L_Eg$ >> > > > > > >> > > >", >> > > > > > >> > > Flink application fails with the below message: >> > > > > > >> > > >> > > > > > >> > > "Vertex's parallelism should be smaller than or equal to >> > > > vertex's >> > > > > > max >> > > > > > >> > > parallelism." >> > > > > > >> > > >> > > > > > >> > > This is the corresponding code in Flink's >> > > > > > DefaultVertexParallelismInfo >> > > > > > >> > > < >> > > > > > >> > > >> > > > > > >> > >> > > > > > >> >> > > > > > >> > > > > >> > > > >> > > >> > >> https://urldefense.com/v3/__https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultVertexParallelismInfo.java*L110__;Iw!!IKRxdwAv5BmarQ!fwD4qP-fTEiqJH9CC3AHgXbR5MJPGm7ll1dYwElK-zrtujWDWio6_yvBa4rHlaZHP_lefLs4bZQAISqBRDEfwA$ >> > > > > > >> > > >. >> > > > > > >> > > My question is, "default-source-parallelism" config >> should >> > be >> > > > > > >> independent >> > > > > > >> > > from the "max-parallelism" flag. The former controls the >> > > default >> > > > > > >> source >> > > > > > >> > > parallelism while the latter controls the max number of >> > > > partitions >> > > > > > to >> > > > > > >> > write >> > > > > > >> > > the intermediate shuffle. >> > > > > > >> > > >> > > > > > >> > > If this is true, then the above check should be fixed. >> > > > Otherwise, >> > > > > > >> wanted >> > > > > > >> > to >> > > > > > >> > > understand why the "default-source-parallelism` should be >> > less >> > > > > than >> > > > > > >> the >> > > > > > >> > > "max-parallelism" >> > > > > > >> > > >> > > > > > >> > > Thanks >> > > > > > >> > > Venkat >> > > > > > >> > > >> > > > > > >> > >> > > > > > >> >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > >> >
