Skip to content

Commit c9ab70e

Browse files
aplyusninaplyusnin
committed
[FLINK-31215] Backpropagate processing rate limits from non-scalable bottlenecks to upstream operators
1 parent 4d5b2c9 commit c9ab70e

File tree

10 files changed

+871
-4
lines changed

10 files changed

+871
-4
lines changed

‎docs/layouts/shortcodes/generated/auto_scaler_configuration.html

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,18 @@
116116
<td>Double</td>
117117
<td>Percentage threshold for switching to observed from busy time based true processing rate if the measurement is off by at least the configured fraction. For example 0.15 means we switch to observed if the busy time based computation is at least 15% higher during catchup.</td>
118118
</tr>
119+
<tr>
120+
<td><h5>job.autoscaler.processing.rate.backpropagation.enabled</h5></td>
121+
<td style="word-wrap: break-word;">false</td>
122+
<td>Boolean</td>
123+
<td>Enable backpropagation of processing rate during autoscaling to reduce resources usage.</td>
124+
</tr>
125+
<tr>
126+
<td><h5>job.autoscaler.processing.rate.backpropagation.impact</h5></td>
127+
<td style="word-wrap: break-word;">0.0</td>
128+
<td>Double</td>
129+
<td>How strong should backpropagated values affect scaling. 0 - means no effect, 1 - use backpropagated values. It is not recommended to set this factor greater than 0.8</td>
130+
</tr>
119131
<tr>
120132
<td><h5>job.autoscaler.quota.cpu</h5></td>
121133
<td style="word-wrap: break-word;">(none)</td>
@@ -210,7 +222,7 @@
210222
<td><h5>job.autoscaler.vertex.exclude.ids</h5></td>
211223
<td style="word-wrap: break-word;"></td>
212224
<td>List&lt;String&gt;</td>
213-
<td>A (semicolon-separated) list of vertex ids in hexstring for which to disable scaling. Caution: For non-sink vertices this will still scale their downstream operators until https://issues.apache.org/jira/browse/FLINK-31215 is implemented.</td>
225+
<td>A (semicolon-separated) list of vertex ids in hexstring for which to disable scaling.</td>
214226
</tr>
215227
<tr>
216228
<td><h5>job.autoscaler.vertex.max-parallelism</h5></td>

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

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,10 @@
2020
import org.apache.flink.annotation.VisibleForTesting;
2121
import org.apache.flink.autoscaler.config.AutoScalerOptions;
2222
import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
23+
import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
2324
import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
2425
import org.apache.flink.autoscaler.metrics.ScalingMetric;
26+
import org.apache.flink.autoscaler.topology.JobTopology;
2527
import org.apache.flink.autoscaler.topology.ShipStrategy;
2628
import org.apache.flink.autoscaler.utils.AutoScalerUtils;
2729
import org.apache.flink.configuration.Configuration;
@@ -37,6 +39,7 @@
3739
import java.time.Instant;
3840
import java.time.ZoneId;
3941
import java.util.Collection;
42+
import java.util.List;
4043
import java.util.Map;
4144
import java.util.Objects;
4245
import java.util.SortedMap;
@@ -52,8 +55,10 @@
5255
import static org.apache.flink.autoscaler.metrics.ScalingMetric.MAX_PARALLELISM;
5356
import static org.apache.flink.autoscaler.metrics.ScalingMetric.NUM_SOURCE_PARTITIONS;
5457
import static org.apache.flink.autoscaler.metrics.ScalingMetric.PARALLELISM;
58+
import static org.apache.flink.autoscaler.metrics.ScalingMetric.TARGET_DATA_RATE;
5559
import static org.apache.flink.autoscaler.metrics.ScalingMetric.TRUE_PROCESSING_RATE;
5660
import static org.apache.flink.autoscaler.topology.ShipStrategy.HASH;
61+
import static org.apache.flink.autoscaler.utils.AutoScalerUtils.getTargetDataRateFromUpstream;
5762
import static org.apache.flink.util.Preconditions.checkArgument;
5863

5964
/** Component responsible for computing vertex parallelism based on the scaling metrics. */
@@ -148,6 +153,67 @@ public static ParallelismChange noChange() {
148153
}
149154
}
150155

156+
public void backpropagateRate(
157+
Configuration conf,
158+
JobVertexID vertex,
159+
JobTopology topology,
160+
EvaluatedMetrics evaluatedMetrics,
161+
Map<JobVertexID, Double> backpropagationRate,
162+
List<String> excludedVertices) {
163+
164+
if (excludedVertices.contains(vertex.toHexString())) {
165+
return;
166+
}
167+
168+
var vertexMetrics = evaluatedMetrics.getVertexMetrics().get(vertex);
169+
170+
// vertex scale factor is limited by max scale factor
171+
double scaleFactor = 1 + conf.get(MAX_SCALE_UP_FACTOR);
172+
173+
// vertex scale factor is limited by max parallelism scale factor
174+
scaleFactor =
175+
Math.min(
176+
scaleFactor,
177+
vertexMetrics.get(MAX_PARALLELISM).getCurrent()
178+
/ vertexMetrics.get(PARALLELISM).getCurrent());
179+
180+
double maxProcessingRateAfterScale =
181+
Math.min(
182+
vertexMetrics.get(TARGET_DATA_RATE).getAverage()
183+
* backpropagationRate.getOrDefault(vertex, 1.0),
184+
vertexMetrics.get(TRUE_PROCESSING_RATE).getAverage() * scaleFactor);
185+
186+
// evaluating partially updated target data rate from upstream
187+
double targetDataRate =
188+
getTargetDataRateFromUpstream(
189+
evaluatedMetrics, topology, vertex, backpropagationRate);
190+
191+
// if cannot derive finite value, then assume full processing
192+
if (Double.isNaN(targetDataRate) || Double.isInfinite(targetDataRate)) {
193+
return;
194+
}
195+
196+
// if cannot derive finite value, then assume full processing
197+
if (Double.isNaN(maxProcessingRateAfterScale)
198+
|| Double.isInfinite(maxProcessingRateAfterScale)) {
199+
return;
200+
}
201+
202+
// if all input stream can be processed, skip propagation
203+
if (targetDataRate < maxProcessingRateAfterScale) {
204+
return;
205+
}
206+
207+
// propagation coefficient
208+
double adjustmentRate = maxProcessingRateAfterScale / targetDataRate;
209+
210+
// update rate of direct upstream vertices
211+
for (var v : topology.getVertexInfos().get(vertex).getInputs().keySet()) {
212+
double vertexRate = backpropagationRate.getOrDefault(v, 1.0);
213+
backpropagationRate.put(v, vertexRate * adjustmentRate);
214+
}
215+
}
216+
151217
public ParallelismChange computeScaleTargetParallelism(
152218
Context context,
153219
JobVertexID vertex,

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

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,8 @@
3636
import org.apache.flink.runtime.instance.SlotSharingGroupId;
3737
import org.apache.flink.runtime.jobgraph.JobVertexID;
3838

39+
import org.apache.flink.shaded.curator5.com.google.common.base.Preconditions;
40+
3941
import org.slf4j.Logger;
4042
import org.slf4j.LoggerFactory;
4143

@@ -53,6 +55,7 @@
5355
import static org.apache.flink.autoscaler.JobVertexScaler.ParallelismChangeType.NO_CHANGE;
5456
import static org.apache.flink.autoscaler.JobVertexScaler.ParallelismChangeType.REQUIRED_CHANGE;
5557
import static org.apache.flink.autoscaler.config.AutoScalerOptions.EXCLUDED_PERIODS;
58+
import static org.apache.flink.autoscaler.config.AutoScalerOptions.PROCESSING_RATE_BACKPROPAGATION_ENABLED;
5659
import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_ENABLED;
5760
import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_EVENT_INTERVAL;
5861
import static org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_EXECUTION_DISABLED_REASON;
@@ -61,6 +64,7 @@
6164
import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.addToScalingHistoryAndStore;
6265
import static org.apache.flink.autoscaler.metrics.ScalingMetric.SCALE_DOWN_RATE_THRESHOLD;
6366
import static org.apache.flink.autoscaler.metrics.ScalingMetric.SCALE_UP_RATE_THRESHOLD;
67+
import static org.apache.flink.autoscaler.metrics.ScalingMetric.TARGET_DATA_RATE;
6468
import static org.apache.flink.autoscaler.metrics.ScalingMetric.TRUE_PROCESSING_RATE;
6569

6670
/** Class responsible for executing scaling decisions. */
@@ -238,6 +242,11 @@ Map<JobVertexID, ScalingSummary> computeScalingSummary(
238242

239243
var excludeVertexIdList =
240244
context.getConfiguration().get(AutoScalerOptions.VERTEX_EXCLUDE_IDS);
245+
246+
if (context.getConfiguration().get(PROCESSING_RATE_BACKPROPAGATION_ENABLED)) {
247+
backpropagateProcessingRate(context, evaluatedMetrics, jobTopology);
248+
}
249+
241250
evaluatedMetrics
242251
.getVertexMetrics()
243252
.forEach(
@@ -284,6 +293,73 @@ Map<JobVertexID, ScalingSummary> computeScalingSummary(
284293
return out;
285294
}
286295

296+
private void backpropagateProcessingRate(
297+
Context context, EvaluatedMetrics evaluatedMetrics, JobTopology jobTopology) {
298+
var conf = context.getConfiguration();
299+
double impact = conf.get(AutoScalerOptions.PROCESSING_RATE_BACKPROPAGATION_IMPACT);
300+
Preconditions.checkState(
301+
0 <= impact && impact <= 1.0, "Backpropagation impact should be in range [0, 1]");
302+
var propagationRate = new HashMap<JobVertexID, Double>();
303+
var excludeVertexIdList =
304+
context.getConfiguration().get(AutoScalerOptions.VERTEX_EXCLUDE_IDS);
305+
var vertexIterator =
306+
jobTopology
307+
.getStrongTopologicalOrder()
308+
.listIterator(jobTopology.getStrongTopologicalOrder().size());
309+
310+
// backpropagate scale factors
311+
while (vertexIterator.hasPrevious()) {
312+
var vertex = vertexIterator.previous();
313+
jobVertexScaler.backpropagateRate(
314+
conf,
315+
vertex,
316+
jobTopology,
317+
evaluatedMetrics,
318+
propagationRate,
319+
excludeVertexIdList);
320+
}
321+
322+
// use an extra map to not lose precision
323+
Map<JobVertexID, Double> adjustedDataRate = new HashMap<>();
324+
325+
// re-evaluating vertices capacity
326+
// Target data rate metric is rewritten for parallelism evaluation
327+
for (var vertex : jobTopology.getVerticesInTopologicalOrder()) {
328+
double newTargetDataRate = 0.0;
329+
330+
if (jobTopology.isSource(vertex)) {
331+
double targetDateRate =
332+
evaluatedMetrics
333+
.getVertexMetrics()
334+
.get(vertex)
335+
.get(TARGET_DATA_RATE)
336+
.getAverage();
337+
338+
// linear interpolation between adjusted value and initial
339+
newTargetDataRate =
340+
targetDateRate
341+
* (impact * propagationRate.getOrDefault(vertex, 1.0)
342+
+ 1.0
343+
- impact);
344+
} else {
345+
for (var input : jobTopology.getVertexInfos().get(vertex).getInputs().keySet()) {
346+
newTargetDataRate +=
347+
adjustedDataRate.get(input)
348+
* jobTopology
349+
.getVertexInfos()
350+
.get(vertex)
351+
.getInputRatios()
352+
.get(input);
353+
}
354+
}
355+
adjustedDataRate.put(vertex, newTargetDataRate);
356+
evaluatedMetrics
357+
.getVertexMetrics()
358+
.get(vertex)
359+
.put(TARGET_DATA_RATE, EvaluatedScalingMetric.avg(newTargetDataRate));
360+
}
361+
}
362+
287363
private boolean isJobUnderMemoryPressure(
288364
Context ctx, Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics) {
289365

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -352,6 +352,7 @@ private void computeTargetDataRate(
352352
var inputTargetRate = inputEvaluatedMetrics.get(TARGET_DATA_RATE);
353353
var outputRatio =
354354
computeEdgeOutputRatio(inputVertex, vertex, topology, metricsHistory);
355+
topology.get(vertex).getInputRatios().put(inputVertex, outputRatio);
355356
LOG.debug(
356357
"Computed output ratio for edge ({} -> {}) : {}",
357358
inputVertex,

‎flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,24 @@ private static ConfigOptions.OptionBuilder autoScalerConfig(String key) {
5858
.withDescription(
5959
"Enable vertex scaling execution by the autoscaler. If disabled, the autoscaler will only collect metrics and evaluate the suggested parallelism for each vertex but will not upgrade the jobs.");
6060

61+
public static final ConfigOption<Boolean> PROCESSING_RATE_BACKPROPAGATION_ENABLED =
62+
autoScalerConfig("processing.rate.backpropagation.enabled")
63+
.booleanType()
64+
.defaultValue(false)
65+
.withFallbackKeys(
66+
oldOperatorConfigKey("processing.rate.backpropagation.enabled"))
67+
.withDescription(
68+
"Enable backpropagation of processing rate during autoscaling to reduce resources usage.");
69+
70+
public static final ConfigOption<Double> PROCESSING_RATE_BACKPROPAGATION_IMPACT =
71+
autoScalerConfig("processing.rate.backpropagation.impact")
72+
.doubleType()
73+
.defaultValue(0.0)
74+
.withFallbackKeys(
75+
oldOperatorConfigKey("processing.rate.backpropagation.impact"))
76+
.withDescription(
77+
"How strong should backpropagated values affect scaling. 0 - means no effect, 1 - use backpropagated values. It is not recommended to set this factor greater than 0.8");
78+
6179
public static final ConfigOption<Duration> METRICS_WINDOW =
6280
autoScalerConfig("metrics.window")
6381
.durationType()
@@ -320,7 +338,7 @@ private static ConfigOptions.OptionBuilder autoScalerConfig(String key) {
320338
.defaultValues()
321339
.withFallbackKeys(oldOperatorConfigKey("vertex.exclude.ids"))
322340
.withDescription(
323-
"A (semicolon-separated) list of vertex ids in hexstring for which to disable scaling. Caution: For non-sink vertices this will still scale their downstream operators until https://issues.apache.org/jira/browse/FLINK-31215 is implemented.");
341+
"A (semicolon-separated) list of vertex ids in hexstring for which to disable scaling.");
324342

325343
public static final ConfigOption<Duration> SCALING_EVENT_INTERVAL =
326344
autoScalerConfig("scaling.event.interval")

‎flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/JobTopology.java

Lines changed: 70 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import java.util.List;
4040
import java.util.Map;
4141
import java.util.Set;
42+
import java.util.TreeMap;
4243
import java.util.stream.Collectors;
4344

4445
/** Structure representing information about the jobgraph that is relevant for scaling. */
@@ -52,6 +53,7 @@ public class JobTopology {
5253
@Getter private final Map<SlotSharingGroupId, Set<JobVertexID>> slotSharingGroupMapping;
5354
@Getter private final Set<JobVertexID> finishedVertices;
5455
@Getter private final List<JobVertexID> verticesInTopologicalOrder;
56+
@Getter private final List<JobVertexID> strongTopologicalOrder;
5557

5658
public JobTopology(Collection<VertexInfo> vertexInfo) {
5759
this(new HashSet<>(vertexInfo));
@@ -99,6 +101,7 @@ public JobTopology(Set<VertexInfo> vertexInfo) {
99101
this.slotSharingGroupMapping = ImmutableMap.copyOf(vertexSlotSharingGroupMapping);
100102
this.finishedVertices = finishedVertices.build();
101103
this.verticesInTopologicalOrder = returnVerticesInTopologicalOrder();
104+
this.strongTopologicalOrder = returnStrongTopologicalOrder();
102105
}
103106

104107
public VertexInfo get(JobVertexID jvi) {
@@ -112,9 +115,9 @@ public boolean isSource(JobVertexID jobVertexID) {
112115
private List<JobVertexID> returnVerticesInTopologicalOrder() {
113116
List<JobVertexID> sorted = new ArrayList<>(vertexInfos.size());
114117

115-
Map<JobVertexID, List<JobVertexID>> remainingInputs = new HashMap<>(vertexInfos.size());
118+
Map<JobVertexID, Set<JobVertexID>> remainingInputs = new HashMap<>(vertexInfos.size());
116119
vertexInfos.forEach(
117-
(id, v) -> remainingInputs.put(id, new ArrayList<>(v.getInputs().keySet())));
120+
(id, v) -> remainingInputs.put(id, new HashSet<>(v.getInputs().keySet())));
118121

119122
while (!remainingInputs.isEmpty()) {
120123
List<JobVertexID> verticesWithZeroIndegree = new ArrayList<>();
@@ -140,6 +143,71 @@ private List<JobVertexID> returnVerticesInTopologicalOrder() {
140143
return sorted;
141144
}
142145

146+
/**
147+
* Strong topological order is a topological order, where vertices are also sorted by their
148+
* distance to the most distant sources.
149+
*
150+
* @return vertices in the enhanced topological order
151+
*/
152+
public List<JobVertexID> returnStrongTopologicalOrder() {
153+
List<JobVertexID> sorted = new ArrayList<>(vertexInfos.size());
154+
155+
Map<JobVertexID, Set<JobVertexID>> remainingInputs = new HashMap<>(vertexInfos.size());
156+
vertexInfos.forEach(
157+
(id, v) -> remainingInputs.put(id, new HashSet<>(v.getInputs().keySet())));
158+
159+
Map<JobVertexID, Integer> distances = new HashMap<>(vertexInfos.size());
160+
TreeMap<Integer, List<JobVertexID>> order = new TreeMap<>();
161+
162+
while (!remainingInputs.isEmpty()) {
163+
List<JobVertexID> verticesWithZeroIndegree = new ArrayList<>();
164+
165+
// storing
166+
remainingInputs.forEach(
167+
(v, inputs) -> {
168+
if (inputs.isEmpty()) {
169+
int dist = distances.getOrDefault(v, 0);
170+
if (!order.containsKey(dist)) {
171+
order.put(dist, new ArrayList<>());
172+
}
173+
order.get(dist).add(v);
174+
verticesWithZeroIndegree.add(v);
175+
}
176+
});
177+
178+
verticesWithZeroIndegree.forEach(
179+
v -> {
180+
remainingInputs.remove(v);
181+
vertexInfos
182+
.get(v)
183+
.getOutputs()
184+
.keySet()
185+
.forEach(o -> remainingInputs.get(o).remove(v));
186+
});
187+
188+
List<JobVertexID> layer = order.firstEntry().getValue();
189+
order.remove(order.firstKey());
190+
191+
layer.forEach(
192+
v -> {
193+
final int dist = distances.getOrDefault(v, 0);
194+
vertexInfos
195+
.get(v)
196+
.getOutputs()
197+
.keySet()
198+
.forEach(
199+
o -> {
200+
remainingInputs.get(o).remove(v);
201+
int dist1 = distances.getOrDefault(o, 0);
202+
distances.put(o, Math.max(dist1, dist + 1));
203+
});
204+
});
205+
206+
sorted.addAll(layer);
207+
}
208+
return sorted;
209+
}
210+
143211
public static JobTopology fromJsonPlan(
144212
String jsonPlan,
145213
Map<JobVertexID, SlotSharingGroupId> slotSharingGroupIdMap,

0 commit comments

Comments
 (0)