[
https://issues.apache.org/jira/browse/FLINK-35285?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17868546#comment-17868546
]
Maximilian Michels commented on FLINK-35285:
--------------------------------------------
Hi! Great discussion. Thanks for lopping me in! I think it makes sense to add
an option to disable the key group alignment logic. I created a JIRA for this a
while ago: FLINK-32120.
Key group alignment is an important pillar to ensure stability of the
algorithm. I understand your concerns about sticky parallelisms, but it is
likely that the number of scalings will increase without key group alignment,
at least in some situations.
Estimating the partition skew in the absence of key group alignment and
compensating for it by only a slightly higher parallelism, is an interesting
idea that we might want to test. I was contemplating something like that (see:
FLINK-32119). We should make sure though the adjusted parallelism still matches
the desired target utilization and its throughput. Without this, the resulting
scaling decision may turn out to be less stable because the lower throughput
bound will be hit faster.
In short, I think we can independently work on:
# FLINK-32120
# FLINK-32119 (or use this issue): Add a new skew estimation logic which
adjusts the parallelism just enough to match the desired target utilization for
scenarios where we don’t use the key group alignment hammer. This would also
apply to scenarios where key group alignment is not possible.
> Autoscaler key group optimization can interfere with scale-down.max-factor
> --------------------------------------------------------------------------
>
> Key: FLINK-35285
> URL: https://issues.apache.org/jira/browse/FLINK-35285
> Project: Flink
> Issue Type: Bug
> Components: Kubernetes Operator
> Reporter: Trystan
> Priority: Minor
>
> When setting a less aggressive scale down limit, the key group optimization
> can prevent a vertex from scaling down at all. It will hunt from target
> upwards to maxParallelism/2, and will always find currentParallelism again.
>
> A simple test trying to scale down from a parallelism of 60 with a
> scale-down.max-factor of 0.2:
> {code:java}
> assertEquals(48, JobVertexScaler.scale(60, inputShipStrategies, 360, .8, 8,
> 360)); {code}
>
> It seems reasonable to make a good attempt to spread data across subtasks,
> but not at the expense of total deadlock. The problem is that during scale
> down it doesn't actually ensure that newParallelism will be <
> currentParallelism. The only workaround is to set a scale down factor large
> enough such that it finds the next lowest divisor of the maxParallelism.
>
> Clunky, but something to ensure it can make at least some progress. There is
> another test that now fails, but just to illustrate the point:
> {code:java}
> for (int p = newParallelism; p <= maxParallelism / 2 && p <= upperBound; p++)
> {
> if ((scaleFactor < 1 && p < currentParallelism) || (scaleFactor > 1 && p
> > currentParallelism)) {
> if (maxParallelism % p == 0) {
> return p;
> }
> }
> } {code}
>
> Perhaps this is by design and not a bug, but total failure to scale down in
> order to keep optimized key groups does not seem ideal.
>
> Key group optimization block:
> [https://github.com/apache/flink-kubernetes-operator/blob/fe3d24e4500d6fcaed55250ccc816546886fd1cf/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java#L296C1-L303C10]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
