Skip to content

Commit 6e16bac

Browse files
1996fanrui1996fanrui
committed
[FLINK-36863][autoscaler] Use the maximum parallelism in the past scale-down.interval window when scaling down
1 parent 6c6c44e commit 6e16bac

File tree

7 files changed

+544
-237
lines changed

7 files changed

+544
-237
lines changed

‎flink-autoscaler/src/main/java/org/apache/flink/autoscaler/DelayedScaleDown.java

Lines changed: 89 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -30,23 +30,96 @@
3030

3131
import java.time.Instant;
3232
import java.util.HashMap;
33+
import java.util.LinkedList;
3334
import java.util.Map;
3435

36+
import static org.apache.flink.util.Preconditions.checkState;
37+
3538
/** All delayed scale down requests. */
3639
public class DelayedScaleDown {
3740

41+
/** Details of the recommended parallelism. */
42+
@Data
43+
public static class RecommendedParallelism {
44+
@Nonnull private final Instant triggerTime;
45+
private final int parallelism;
46+
private final boolean outsideUtilizationBound;
47+
48+
@JsonCreator
49+
public RecommendedParallelism(
50+
@Nonnull @JsonProperty("triggerTime") Instant triggerTime,
51+
@JsonProperty("parallelism") int parallelism,
52+
@JsonProperty("outsideUtilizationBound") boolean outsideUtilizationBound) {
53+
this.triggerTime = triggerTime;
54+
this.parallelism = parallelism;
55+
this.outsideUtilizationBound = outsideUtilizationBound;
56+
}
57+
}
58+
3859
/** The delayed scale down info for vertex. */
3960
@Data
4061
public static class VertexDelayedScaleDownInfo {
4162
private final Instant firstTriggerTime;
42-
private int maxRecommendedParallelism;
63+
64+
/**
65+
* In theory, it maintains all recommended parallelisms at each time within the past
66+
* `scale-down.interval` window period, so all recommended parallelisms before the window
67+
* start time will be evicted.
68+
*
69+
* <p>Also, if latest parallelism is greater than the past parallelism, all smaller
70+
* parallelism in the past never be the max recommended parallelism, so we could evict all
71+
* smaller parallelism in the past. It's a general optimization for calculating max value
72+
* for sliding window. So we only need to maintain a list with monotonically decreasing
73+
* parallelism within the past window, and the first parallelism will be the max recommended
74+
* parallelism within the past `scale-down.interval` window period.
75+
*/
76+
private final LinkedList<RecommendedParallelism> recommendedParallelisms;
77+
78+
public VertexDelayedScaleDownInfo(Instant firstTriggerTime) {
79+
this.firstTriggerTime = firstTriggerTime;
80+
this.recommendedParallelisms = new LinkedList<>();
81+
}
4382

4483
@JsonCreator
4584
public VertexDelayedScaleDownInfo(
4685
@JsonProperty("firstTriggerTime") Instant firstTriggerTime,
47-
@JsonProperty("maxRecommendedParallelism") int maxRecommendedParallelism) {
86+
@JsonProperty("recommendedParallelisms")
87+
LinkedList<RecommendedParallelism> recommendedParallelisms) {
4888
this.firstTriggerTime = firstTriggerTime;
49-
this.maxRecommendedParallelism = maxRecommendedParallelism;
89+
this.recommendedParallelisms = recommendedParallelisms;
90+
}
91+
92+
/** Record current recommended parallelism. */
93+
public void recordRecommendedParallelism(
94+
Instant triggerTime, int parallelism, boolean outsideUtilizationBound) {
95+
// Evict all recommended parallelisms that are lower than or equal to the latest
96+
// parallelism. When the past parallelism is equal to the latest parallelism,
97+
// triggerTime needs to be updated, so it also needs to be evicted.
98+
while (!recommendedParallelisms.isEmpty()
99+
&& recommendedParallelisms.peekLast().getParallelism() <= parallelism) {
100+
recommendedParallelisms.pollLast();
101+
}
102+
103+
recommendedParallelisms.addLast(
104+
new RecommendedParallelism(triggerTime, parallelism, outsideUtilizationBound));
105+
}
106+
107+
@JsonIgnore
108+
public RecommendedParallelism getMaxRecommendedParallelism(Instant windowStartTime) {
109+
// Evict all recommended parallelisms before the window start time.
110+
while (!recommendedParallelisms.isEmpty()
111+
&& recommendedParallelisms
112+
.peekFirst()
113+
.getTriggerTime()
114+
.isBefore(windowStartTime)) {
115+
recommendedParallelisms.pollFirst();
116+
}
117+
118+
var maxRecommendedParallelism = recommendedParallelisms.peekFirst();
119+
checkState(
120+
maxRecommendedParallelism != null,
121+
"The getMaxRecommendedParallelism should be called after triggering a scale down, it may be a bug.");
122+
return maxRecommendedParallelism;
50123
}
51124
}
52125

@@ -63,18 +136,19 @@ public DelayedScaleDown() {
63136
/** Trigger a scale down, and return the corresponding {@link VertexDelayedScaleDownInfo}. */
64137
@Nonnull
65138
public VertexDelayedScaleDownInfo triggerScaleDown(
66-
JobVertexID vertex, Instant triggerTime, int parallelism) {
67-
var vertexDelayedScaleDownInfo = delayedVertices.get(vertex);
68-
if (vertexDelayedScaleDownInfo == null) {
69-
// It's the first trigger
70-
vertexDelayedScaleDownInfo = new VertexDelayedScaleDownInfo(triggerTime, parallelism);
71-
delayedVertices.put(vertex, vertexDelayedScaleDownInfo);
72-
updated = true;
73-
} else if (parallelism > vertexDelayedScaleDownInfo.getMaxRecommendedParallelism()) {
74-
// Not the first trigger, but the maxRecommendedParallelism needs to be updated.
75-
vertexDelayedScaleDownInfo.setMaxRecommendedParallelism(parallelism);
76-
updated = true;
77-
}
139+
JobVertexID vertex,
140+
Instant triggerTime,
141+
int parallelism,
142+
boolean outsideUtilizationBound) {
143+
// The vertexDelayedScaleDownInfo is updated once scale down is triggered due to we need
144+
// update the triggerTime each time.
145+
updated = true;
146+
147+
var vertexDelayedScaleDownInfo =
148+
delayedVertices.computeIfAbsent(
149+
vertex, k -> new VertexDelayedScaleDownInfo(triggerTime));
150+
vertexDelayedScaleDownInfo.recordRecommendedParallelism(
151+
triggerTime, parallelism, outsideUtilizationBound);
78152

79153
return vertexDelayedScaleDownInfo;
80154
}

‎flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java

Lines changed: 62 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,8 @@
5656
import static org.apache.flink.autoscaler.metrics.ScalingMetric.MAX_PARALLELISM;
5757
import static org.apache.flink.autoscaler.metrics.ScalingMetric.NUM_SOURCE_PARTITIONS;
5858
import static org.apache.flink.autoscaler.metrics.ScalingMetric.PARALLELISM;
59+
import static org.apache.flink.autoscaler.metrics.ScalingMetric.SCALE_DOWN_RATE_THRESHOLD;
60+
import static org.apache.flink.autoscaler.metrics.ScalingMetric.SCALE_UP_RATE_THRESHOLD;
5961
import static org.apache.flink.autoscaler.metrics.ScalingMetric.TRUE_PROCESSING_RATE;
6062
import static org.apache.flink.autoscaler.topology.ShipStrategy.HASH;
6163
import static org.apache.flink.configuration.description.TextElement.text;
@@ -92,12 +94,15 @@ public JobVertexScaler(AutoScalerEventHandler<KEY, Context> autoScalerEventHandl
9294
@Getter
9395
public static class ParallelismChange {
9496

95-
private static final ParallelismChange NO_CHANGE = new ParallelismChange(-1);
97+
private static final ParallelismChange NO_CHANGE = new ParallelismChange(-1, false);
9698

9799
private final int newParallelism;
98100

99-
private ParallelismChange(int newParallelism) {
101+
private final boolean outsideUtilizationBound;
102+
103+
private ParallelismChange(int newParallelism, boolean outsideUtilizationBound) {
100104
this.newParallelism = newParallelism;
105+
this.outsideUtilizationBound = outsideUtilizationBound;
101106
}
102107

103108
public boolean isNoChange() {
@@ -113,24 +118,29 @@ public boolean equals(Object o) {
113118
return false;
114119
}
115120
ParallelismChange that = (ParallelismChange) o;
116-
return newParallelism == that.newParallelism;
121+
return newParallelism == that.newParallelism
122+
&& outsideUtilizationBound == that.outsideUtilizationBound;
117123
}
118124

119125
@Override
120126
public int hashCode() {
121-
return Objects.hash(newParallelism);
127+
return Objects.hash(newParallelism, outsideUtilizationBound);
122128
}
123129

124130
@Override
125131
public String toString() {
126132
return isNoChange()
127133
? "NoParallelismChange"
128-
: "ParallelismChange{newParallelism=" + newParallelism + '}';
134+
: "ParallelismChange{newParallelism="
135+
+ newParallelism
136+
+ ", outsideUtilizationBound="
137+
+ outsideUtilizationBound
138+
+ "}";
129139
}
130140

131-
public static ParallelismChange build(int newParallelism) {
141+
public static ParallelismChange build(int newParallelism, boolean outsideUtilizationBound) {
132142
checkArgument(newParallelism > 0, "The parallelism should be greater than 0.");
133-
return new ParallelismChange(newParallelism);
143+
return new ParallelismChange(newParallelism, outsideUtilizationBound);
134144
}
135145

136146
public static ParallelismChange noChange() {
@@ -239,6 +249,8 @@ private ParallelismChange detectBlockScaling(
239249
currentParallelism != newParallelism,
240250
"The newParallelism is equal to currentParallelism, no scaling is needed. This is probably a bug.");
241251

252+
var outsideUtilizationBound = outsideUtilizationBound(vertex, evaluatedMetrics);
253+
242254
var scaledUp = currentParallelism < newParallelism;
243255

244256
if (scaledUp) {
@@ -248,7 +260,7 @@ private ParallelismChange detectBlockScaling(
248260

249261
// If we don't have past scaling actions for this vertex, don't block scale up.
250262
if (history.isEmpty()) {
251-
return ParallelismChange.build(newParallelism);
263+
return ParallelismChange.build(newParallelism, outsideUtilizationBound);
252264
}
253265

254266
var lastSummary = history.get(history.lastKey());
@@ -260,28 +272,59 @@ && detectIneffectiveScaleUp(
260272
return ParallelismChange.noChange();
261273
}
262274

263-
return ParallelismChange.build(newParallelism);
275+
return ParallelismChange.build(newParallelism, outsideUtilizationBound);
276+
} else {
277+
return applyScaleDownInterval(
278+
delayedScaleDown, vertex, conf, newParallelism, outsideUtilizationBound);
279+
}
280+
}
281+
282+
private static boolean outsideUtilizationBound(
283+
JobVertexID vertex, Map<ScalingMetric, EvaluatedScalingMetric> metrics) {
284+
double trueProcessingRate = metrics.get(TRUE_PROCESSING_RATE).getAverage();
285+
double scaleUpRateThreshold = metrics.get(SCALE_UP_RATE_THRESHOLD).getCurrent();
286+
double scaleDownRateThreshold = metrics.get(SCALE_DOWN_RATE_THRESHOLD).getCurrent();
287+
288+
if (trueProcessingRate < scaleUpRateThreshold
289+
|| trueProcessingRate > scaleDownRateThreshold) {
290+
LOG.debug(
291+
"Vertex {} processing rate {} is outside ({}, {})",
292+
vertex,
293+
trueProcessingRate,
294+
scaleUpRateThreshold,
295+
scaleDownRateThreshold);
296+
return true;
264297
} else {
265-
return applyScaleDownInterval(delayedScaleDown, vertex, conf, newParallelism);
298+
LOG.debug(
299+
"Vertex {} processing rate {} is within target ({}, {})",
300+
vertex,
301+
trueProcessingRate,
302+
scaleUpRateThreshold,
303+
scaleDownRateThreshold);
266304
}
305+
return false;
267306
}
268307

269308
private ParallelismChange applyScaleDownInterval(
270309
DelayedScaleDown delayedScaleDown,
271310
JobVertexID vertex,
272311
Configuration conf,
273-
int newParallelism) {
312+
int newParallelism,
313+
boolean outsideUtilizationBound) {
274314
var scaleDownInterval = conf.get(SCALE_DOWN_INTERVAL);
275315
if (scaleDownInterval.toMillis() <= 0) {
276316
// The scale down interval is disable, so don't block scaling.
277-
return ParallelismChange.build(newParallelism);
317+
return ParallelismChange.build(newParallelism, outsideUtilizationBound);
278318
}
279319

280320
var now = clock.instant();
281-
var delayedScaleDownInfo = delayedScaleDown.triggerScaleDown(vertex, now, newParallelism);
321+
var windowStartTime = now.minus(scaleDownInterval);
322+
var delayedScaleDownInfo =
323+
delayedScaleDown.triggerScaleDown(
324+
vertex, now, newParallelism, outsideUtilizationBound);
282325

283326
// Never scale down within scale down interval
284-
if (now.isBefore(delayedScaleDownInfo.getFirstTriggerTime().plus(scaleDownInterval))) {
327+
if (windowStartTime.isBefore(delayedScaleDownInfo.getFirstTriggerTime())) {
285328
if (now.equals(delayedScaleDownInfo.getFirstTriggerTime())) {
286329
LOG.info("The scale down of {} is delayed by {}.", vertex, scaleDownInterval);
287330
} else {
@@ -293,7 +336,11 @@ private ParallelismChange applyScaleDownInterval(
293336
} else {
294337
// Using the maximum parallelism within the scale down interval window instead of the
295338
// latest parallelism when scaling down
296-
return ParallelismChange.build(delayedScaleDownInfo.getMaxRecommendedParallelism());
339+
var maxRecommendedParallelism =
340+
delayedScaleDownInfo.getMaxRecommendedParallelism(windowStartTime);
341+
return ParallelismChange.build(
342+
maxRecommendedParallelism.getParallelism(),
343+
maxRecommendedParallelism.isOutsideUtilizationBound());
297344
}
298345
}
299346

‎flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java

Lines changed: 7 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -47,8 +47,8 @@
4747
import java.util.Collections;
4848
import java.util.HashMap;
4949
import java.util.Map;
50-
import java.util.Set;
5150
import java.util.SortedMap;
51+
import java.util.concurrent.atomic.AtomicBoolean;
5252

5353
import static org.apache.flink.autoscaler.config.AutoScalerOptions.EXCLUDED_PERIODS;
5454
import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_ENABLED;
@@ -57,9 +57,6 @@
5757
import static org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_SUMMARY_HEADER_SCALING_EXECUTION_DISABLED;
5858
import static org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_SUMMARY_HEADER_SCALING_EXECUTION_ENABLED;
5959
import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.addToScalingHistoryAndStore;
60-
import static org.apache.flink.autoscaler.metrics.ScalingMetric.SCALE_DOWN_RATE_THRESHOLD;
61-
import static org.apache.flink.autoscaler.metrics.ScalingMetric.SCALE_UP_RATE_THRESHOLD;
62-
import static org.apache.flink.autoscaler.metrics.ScalingMetric.TRUE_PROCESSING_RATE;
6360

6461
/** Class responsible for executing scaling decisions. */
6562
public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
@@ -178,44 +175,6 @@ private void updateRecommendedParallelism(
178175
scalingSummary.getNewParallelism())));
179176
}
180177

181-
@VisibleForTesting
182-
static boolean allChangedVerticesWithinUtilizationTarget(
183-
Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedMetrics,
184-
Set<JobVertexID> changedVertices) {
185-
// No vertices with changed parallelism.
186-
if (changedVertices.isEmpty()) {
187-
return true;
188-
}
189-
190-
for (JobVertexID vertex : changedVertices) {
191-
var metrics = evaluatedMetrics.get(vertex);
192-
193-
double trueProcessingRate = metrics.get(TRUE_PROCESSING_RATE).getAverage();
194-
double scaleUpRateThreshold = metrics.get(SCALE_UP_RATE_THRESHOLD).getCurrent();
195-
double scaleDownRateThreshold = metrics.get(SCALE_DOWN_RATE_THRESHOLD).getCurrent();
196-
197-
if (trueProcessingRate < scaleUpRateThreshold
198-
|| trueProcessingRate > scaleDownRateThreshold) {
199-
LOG.debug(
200-
"Vertex {} processing rate {} is outside ({}, {})",
201-
vertex,
202-
trueProcessingRate,
203-
scaleUpRateThreshold,
204-
scaleDownRateThreshold);
205-
return false;
206-
} else {
207-
LOG.debug(
208-
"Vertex {} processing rate {} is within target ({}, {})",
209-
vertex,
210-
trueProcessingRate,
211-
scaleUpRateThreshold,
212-
scaleDownRateThreshold);
213-
}
214-
}
215-
LOG.info("All vertex processing rates are within target.");
216-
return true;
217-
}
218-
219178
@VisibleForTesting
220179
Map<JobVertexID, ScalingSummary> computeScalingSummary(
221180
Context context,
@@ -235,6 +194,7 @@ Map<JobVertexID, ScalingSummary> computeScalingSummary(
235194

236195
var excludeVertexIdList =
237196
context.getConfiguration().get(AutoScalerOptions.VERTEX_EXCLUDE_IDS);
197+
AtomicBoolean anyVertexOutsideBound = new AtomicBoolean(false);
238198
evaluatedMetrics
239199
.getVertexMetrics()
240200
.forEach(
@@ -260,6 +220,9 @@ Map<JobVertexID, ScalingSummary> computeScalingSummary(
260220
if (parallelismChange.isNoChange()) {
261221
return;
262222
}
223+
if (parallelismChange.isOutsideUtilizationBound()) {
224+
anyVertexOutsideBound.set(true);
225+
}
263226
out.put(
264227
v,
265228
new ScalingSummary(
@@ -270,8 +233,8 @@ Map<JobVertexID, ScalingSummary> computeScalingSummary(
270233
});
271234

272235
// If the Utilization of all tasks is within range, we can skip scaling.
273-
if (allChangedVerticesWithinUtilizationTarget(
274-
evaluatedMetrics.getVertexMetrics(), out.keySet())) {
236+
if (!anyVertexOutsideBound.get()) {
237+
LOG.info("All vertex processing rates are within target.");
275238
return Map.of();
276239
}
277240

0 commit comments

Comments
 (0)