Skip to content

Commit bc455b2

Browse files
Dennis-MirceaDennis-Mircea
committed
[FLINK-39938] FLIP-586: Composable Parallelism Alignment Modes for Flink Autoscaler
1 parent 44027b9 commit bc455b2

29 files changed

Lines changed: 1612 additions & 382 deletions

File tree

‎docs/content.zh/docs/operations/plugins.md‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -278,6 +278,92 @@ The following steps demonstrate how to develop and use a custom metric evaluator
278278
o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom metric evaluator via ServiceLoader: org.apache.flink.autoscaler.custom.CustomEvaluator.
279279
```
280280
281+
## Custom Parallelism Alignment Modes
282+
283+
The autoscaler aligns a vertex's computed target parallelism to the number of key groups (keyBy) or source partitions to reduce data skew. The built-in modes (`BALANCED`, `EVENLY_SPREAD`, `OFF`) are selected by name through `job.autoscaler.scaling.parallelism-alignment.mode`. A custom alignment mode can also be provided as a plugin by implementing the `ParallelismAlignmentMode` SPI. Custom modes are discovered through the [Plugins](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins) mechanism when running inside the Kubernetes operator, and through the standard Java `ServiceLoader` mechanism when running with `flink-autoscaler-standalone`. In both cases the implementation class must be registered in `META-INF/services`.
284+
285+
All `job.autoscaler.*` keys related to custom alignment modes also support the legacy `kubernetes.operator.`-prefixed form as a fallback (for example, `kubernetes.operator.job.autoscaler.scaling.parallelism-alignment.mode`). The canonical key takes precedence on overlap.
286+
287+
The following steps demonstrate how to develop and use a custom alignment mode.
288+
289+
1. Implement `org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode`:
290+
```java
291+
package org.apache.flink.autoscaler.alignment;
292+
293+
/** Custom alignment mode snapping the computed parallelism down to the nearest divisor. */
294+
public class CustomAlignmentMode implements ParallelismAlignmentMode {
295+
296+
/** Apply to every vertex, not only the source and keyBy vertices the built-ins handle. */
297+
@Override
298+
public boolean isApplicable(Context ctx) {
299+
return true;
300+
}
301+
302+
@Override
303+
public int align(Context ctx) {
304+
// Number of key groups, or source partitions for a partitioned source.
305+
int n = ctx.getNumSourcePartitions() > 0
306+
? ctx.getNumSourcePartitions()
307+
: ctx.getMaxParallelism();
308+
// Optional mode-specific parameter, read from the per-mode configuration.
309+
int minParallelism = ctx.getModeConfiguration().getInteger("min-parallelism", 1);
310+
for (int p = ctx.getNewParallelism(); p >= minParallelism; p--) {
311+
if (n % p == 0) {
312+
return p;
313+
}
314+
}
315+
return ctx.getNewParallelism();
316+
}
317+
}
318+
```
319+
The `Context` exposes the current and computed target parallelism, the number of key groups or source partitions, the input ship strategies, the `JobAutoScalerContext`, the per-vertex evaluated metrics, the job topology, and the prefix-stripped per-mode `Configuration` (`getModeConfiguration()`). The autoscaler calls `align` only when `isApplicable(Context)` returns true. That method defaults to keyBy (hash) vertices and to partitioned sources that report a partition count (Kafka and Pulsar do by default), and a custom mode can override it to widen the scope, for example to align custom partitioned vertices.
320+
321+
2. Create the service definition file `org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode` in `META-INF/services`:
322+
```text
323+
org.apache.flink.autoscaler.alignment.CustomAlignmentMode
324+
```
325+
326+
3. Use the Maven tool to package the project and generate the custom alignment mode JAR.
327+
328+
4. Select the custom mode by name and point it at your implementation class. Any other `job.autoscaler.scaling.parallelism-alignment.mode.<name>.*` entries are passed to the mode (prefix-stripped) through `Context#getModeConfiguration()`:
329+
```yaml
330+
job.autoscaler.scaling.parallelism-alignment.mode: custom-mode
331+
job.autoscaler.scaling.parallelism-alignment.mode.custom-mode.class: org.apache.flink.autoscaler.alignment.CustomAlignmentMode
332+
job.autoscaler.scaling.parallelism-alignment.mode.custom-mode.min-parallelism: 4
333+
```
334+
{{< hint info >}}
335+
The `<name>` is any identifier you choose. `custom-mode` is used here, but `CUSTOM_MODE` or any other style works just as well. It is matched exactly, including case, so the value of `scaling.parallelism-alignment.mode` and the `<name>` in `scaling.parallelism-alignment.mode.<name>.class` must be identical. As a result, `CUSTOM_MODE` and `custom-mode` are two different modes, not aliases. The only reserved names are the built-ins (`BALANCED`, `EVENLY_SPREAD`, `OFF`), which always resolve to the built-in modes regardless of any configured class.
336+
{{< /hint >}}
337+
338+
5. Deploy the mode.
339+
340+
- **Operator deployment** - create a Dockerfile to build a custom image from the `apache/flink-kubernetes-operator` official image and copy the generated JAR to a custom alignment mode plugin directory under `/opt/flink/plugins` (the value of the `FLINK_PLUGINS_DIR` environment variable in the flink-kubernetes-operator helm chart). The structure of the custom alignment mode directory under `/opt/flink/plugins` is as follows:
341+
```text
342+
/opt/flink/plugins
343+
├── custom-alignment-mode
344+
│ ├── custom-alignment-mode.jar
345+
└── ...
346+
```
347+
348+
With the custom alignment mode directory location, the Dockerfile is defined as follows:
349+
```shell script
350+
FROM apache/flink-kubernetes-operator
351+
ENV FLINK_PLUGINS_DIR=/opt/flink/plugins
352+
ENV CUSTOM_ALIGNMENT_MODE_DIR=custom-alignment-mode
353+
RUN mkdir $FLINK_PLUGINS_DIR/$CUSTOM_ALIGNMENT_MODE_DIR
354+
COPY custom-alignment-mode.jar $FLINK_PLUGINS_DIR/$CUSTOM_ALIGNMENT_MODE_DIR/
355+
```
356+
357+
Install the flink-kubernetes-operator helm chart with the custom image and verify the `deploy/flink-kubernetes-operator` log has:
358+
```text
359+
o.a.f.k.o.u.AutoscalerUtils [INFO ] Discovered custom alignment mode for autoscaler from plugin directory[/opt/flink/plugins]: org.apache.flink.autoscaler.alignment.CustomAlignmentMode.
360+
```
361+
362+
- **Standalone autoscaler** - simply place the custom alignment mode JAR on the classpath of the `flink-autoscaler-standalone` process. It will be picked up automatically via Java's `ServiceLoader` and discovery will be logged:
363+
```text
364+
o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom alignment mode via ServiceLoader: org.apache.flink.autoscaler.alignment.CustomAlignmentMode.
365+
```
366+
281367
## Custom Scaling Executors
282368
283369
`ScalingExecutorPlugin` is a pluggable component that allows users to intercept, modify, or veto computed scaling decisions before they are applied. Custom scaling executors are discovered through the [Plugins](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins) mechanism when running inside the Kubernetes operator, and through the standard Java `ServiceLoader` mechanism when running with `flink-autoscaler-standalone`. In both cases the implementation class must be registered in `META-INF/services`.

‎docs/content/docs/operations/plugins.md‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -278,6 +278,92 @@ The following steps demonstrate how to develop and use a custom metric evaluator
278278
o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom metric evaluator via ServiceLoader: org.apache.flink.autoscaler.custom.CustomEvaluator.
279279
```
280280
281+
## Custom Parallelism Alignment Modes
282+
283+
The autoscaler aligns a vertex's computed target parallelism to the number of key groups (keyBy) or source partitions to reduce data skew. The built-in modes (`BALANCED`, `EVENLY_SPREAD`, `OFF`) are selected by name through `job.autoscaler.scaling.parallelism-alignment.mode`. A custom alignment mode can also be provided as a plugin by implementing the `ParallelismAlignmentMode` SPI. Custom modes are discovered through the [Plugins](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins) mechanism when running inside the Kubernetes operator, and through the standard Java `ServiceLoader` mechanism when running with `flink-autoscaler-standalone`. In both cases the implementation class must be registered in `META-INF/services`.
284+
285+
All `job.autoscaler.*` keys related to custom alignment modes also support the legacy `kubernetes.operator.`-prefixed form as a fallback (for example, `kubernetes.operator.job.autoscaler.scaling.parallelism-alignment.mode`). The canonical key takes precedence on overlap.
286+
287+
The following steps demonstrate how to develop and use a custom alignment mode.
288+
289+
1. Implement `org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode`:
290+
```java
291+
package org.apache.flink.autoscaler.alignment;
292+
293+
/** Custom alignment mode snapping the computed parallelism down to the nearest divisor. */
294+
public class CustomAlignmentMode implements ParallelismAlignmentMode {
295+
296+
/** Apply to every vertex, not only the source and keyBy vertices the built-ins handle. */
297+
@Override
298+
public boolean isApplicable(Context ctx) {
299+
return true;
300+
}
301+
302+
@Override
303+
public int align(Context ctx) {
304+
// Number of key groups, or source partitions for a partitioned source.
305+
int n = ctx.getNumSourcePartitions() > 0
306+
? ctx.getNumSourcePartitions()
307+
: ctx.getMaxParallelism();
308+
// Optional mode-specific parameter, read from the per-mode configuration.
309+
int minParallelism = ctx.getModeConfiguration().getInteger("min-parallelism", 1);
310+
for (int p = ctx.getNewParallelism(); p >= minParallelism; p--) {
311+
if (n % p == 0) {
312+
return p;
313+
}
314+
}
315+
return ctx.getNewParallelism();
316+
}
317+
}
318+
```
319+
The `Context` exposes the current and computed target parallelism, the number of key groups or source partitions, the input ship strategies, the `JobAutoScalerContext`, the per-vertex evaluated metrics, the job topology, and the prefix-stripped per-mode `Configuration` (`getModeConfiguration()`). The autoscaler calls `align` only when `isApplicable(Context)` returns true. That method defaults to keyBy (hash) vertices and to partitioned sources that report a partition count (Kafka and Pulsar do by default), and a custom mode can override it to widen the scope, for example to align custom partitioned vertices.
320+
321+
2. Create the service definition file `org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode` in `META-INF/services`:
322+
```text
323+
org.apache.flink.autoscaler.alignment.CustomAlignmentMode
324+
```
325+
326+
3. Use the Maven tool to package the project and generate the custom alignment mode JAR.
327+
328+
4. Select the custom mode by name and point it at your implementation class. Any other `job.autoscaler.scaling.parallelism-alignment.mode.<name>.*` entries are passed to the mode (prefix-stripped) through `Context#getModeConfiguration()`:
329+
```yaml
330+
job.autoscaler.scaling.parallelism-alignment.mode: custom-mode
331+
job.autoscaler.scaling.parallelism-alignment.mode.custom-mode.class: org.apache.flink.autoscaler.alignment.CustomAlignmentMode
332+
job.autoscaler.scaling.parallelism-alignment.mode.custom-mode.min-parallelism: 4
333+
```
334+
{{< hint info >}}
335+
The `<name>` is any identifier you choose. `custom-mode` is used here, but `CUSTOM_MODE` or any other style works just as well. It is matched exactly, including case, so the value of `scaling.parallelism-alignment.mode` and the `<name>` in `scaling.parallelism-alignment.mode.<name>.class` must be identical. As a result, `CUSTOM_MODE` and `custom-mode` are two different modes, not aliases. The only reserved names are the built-ins (`BALANCED`, `EVENLY_SPREAD`, `OFF`), which always resolve to the built-in modes regardless of any configured class.
336+
{{< /hint >}}
337+
338+
5. Deploy the mode.
339+
340+
- **Operator deployment** - create a Dockerfile to build a custom image from the `apache/flink-kubernetes-operator` official image and copy the generated JAR to a custom alignment mode plugin directory under `/opt/flink/plugins` (the value of the `FLINK_PLUGINS_DIR` environment variable in the flink-kubernetes-operator helm chart). The structure of the custom alignment mode directory under `/opt/flink/plugins` is as follows:
341+
```text
342+
/opt/flink/plugins
343+
├── custom-alignment-mode
344+
│ ├── custom-alignment-mode.jar
345+
└── ...
346+
```
347+
348+
With the custom alignment mode directory location, the Dockerfile is defined as follows:
349+
```shell script
350+
FROM apache/flink-kubernetes-operator
351+
ENV FLINK_PLUGINS_DIR=/opt/flink/plugins
352+
ENV CUSTOM_ALIGNMENT_MODE_DIR=custom-alignment-mode
353+
RUN mkdir $FLINK_PLUGINS_DIR/$CUSTOM_ALIGNMENT_MODE_DIR
354+
COPY custom-alignment-mode.jar $FLINK_PLUGINS_DIR/$CUSTOM_ALIGNMENT_MODE_DIR/
355+
```
356+
357+
Install the flink-kubernetes-operator helm chart with the custom image and verify the `deploy/flink-kubernetes-operator` log has:
358+
```text
359+
o.a.f.k.o.u.AutoscalerUtils [INFO ] Discovered custom alignment mode for autoscaler from plugin directory[/opt/flink/plugins]: org.apache.flink.autoscaler.alignment.CustomAlignmentMode.
360+
```
361+
362+
- **Standalone autoscaler** - simply place the custom alignment mode JAR on the classpath of the `flink-autoscaler-standalone` process. It will be picked up automatically via Java's `ServiceLoader` and discovery will be logged:
363+
```text
364+
o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom alignment mode via ServiceLoader: org.apache.flink.autoscaler.alignment.CustomAlignmentMode.
365+
```
366+
281367
## Custom Scaling Executors
282368
283369
`ScalingExecutorPlugin` is a pluggable component that allows users to intercept, modify, or veto computed scaling decisions before they are applied. Custom scaling executors are discovered through the [Plugins](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins) mechanism when running inside the Kubernetes operator, and through the standard Java `ServiceLoader` mechanism when running with `flink-autoscaler-standalone`. In both cases the implementation class must be registered in `META-INF/services`.

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

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -218,6 +218,24 @@
218218
<td>List&lt;String&gt;</td>
219219
<td>List of named custom scaling executors instances to register with the autoscaler. For each &lt;name&gt; entry, the implementation class to instantiate must be set via 'job.autoscaler.scaling.custom-executor.&lt;name&gt;.class', and additional per-instance options live under 'job.autoscaler.scaling.custom-executor.&lt;name&gt;.&lt;parameter&gt;'. Custom executors are executed in ascending order of their ScalingExecutorPlugin#priority() value, regardless of the order in which they appear in this list.</td>
220220
</tr>
221+
<tr>
222+
<td><h5>job.autoscaler.scaling.parallelism-alignment.mode</h5></td>
223+
<td style="word-wrap: break-word;">"BALANCED"</td>
224+
<td>String</td>
225+
<td>How the autoscaler aligns the parallelism of source vertices and keyBy (hash) vertices to the number of key groups or source partitions. One of the built-in modes (BALANCED reduces per-subtask load above the target, EVENLY_SPREAD aligns to an exact divisor for even data distribution, OFF disables alignment), or the name of a custom alignment mode whose class is configured via 'job.autoscaler.scaling.parallelism-alignment.mode.&lt;name&gt;.class'. Supersedes the deprecated scaling.key-group.partitions.adjust.mode.</td>
226+
</tr>
227+
<tr>
228+
<td><h5>job.autoscaler.scaling.parallelism-alignment.mode.&lt;name&gt;.&lt;parameter&gt;</h5></td>
229+
<td style="word-wrap: break-word;">(none)</td>
230+
<td>String</td>
231+
<td>An arbitrary parameter passed to the custom alignment mode named &lt;name&gt;. All such parameters are made available to the mode (with the prefix stripped) through Context.getModeConfiguration().</td>
232+
</tr>
233+
<tr>
234+
<td><h5>job.autoscaler.scaling.parallelism-alignment.mode.&lt;name&gt;.class</h5></td>
235+
<td style="word-wrap: break-word;">(none)</td>
236+
<td>String</td>
237+
<td>The fully-qualified class name of the custom alignment mode named &lt;name&gt;, discovered as a plugin and selectable via scaling.parallelism-alignment.mode=&lt;name&gt;.</td>
238+
</tr>
221239
<tr>
222240
<td><h5>job.autoscaler.scaling.effectiveness.detection.enabled</h5></td>
223241
<td style="word-wrap: break-word;">false</td>
@@ -242,12 +260,6 @@
242260
<td>Duration</td>
243261
<td>Time interval to resend the identical event</td>
244262
</tr>
245-
<tr>
246-
<td><h5>job.autoscaler.scaling.key-group.partitions.adjust.mode</h5></td>
247-
<td style="word-wrap: break-word;">EVENLY_SPREAD</td>
248-
<td><p>Enum</p></td>
249-
<td>How to adjust the parallelism of Source vertex or upstream shuffle is keyBy<br /><br />Possible values:<ul><li>"EVENLY_SPREAD": This mode ensures that the parallelism adjustment attempts to evenly distribute data across subtasks. It is particularly effective for source vertices that are aware of partition counts or vertices after 'keyBy' operation. The goal is to have the number of key groups or partitions be divisible by the set parallelism, ensuring even data distribution and reducing data skew.</li><li>"MAXIMIZE_UTILISATION": This model is to maximize resource utilization. In this mode, an attempt is made to set the parallelism that meets the current consumption rate requirements. It is not enforced that the number of key groups or partitions is divisible by the parallelism.</li></ul></td>
250-
</tr>
251263
<tr>
252264
<td><h5>job.autoscaler.stabilization.interval</h5></td>
253265
<td style="word-wrap: break-word;">5 min</td>

‎flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.apache.flink.autoscaler.ScalingExecutor;
2828
import org.apache.flink.autoscaler.ScalingExecutorPlugin;
2929
import org.apache.flink.autoscaler.ScalingMetricEvaluator;
30+
import org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode;
3031
import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
3132
import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
3233
import org.apache.flink.autoscaler.standalone.flinkcluster.FlinkClusterJobListFetcher;
@@ -97,11 +98,14 @@ JobAutoScaler<KEY, Context> createJobAutoscaler(
9798
AutoscalerUtils.discoverCustomEvaluators();
9899
Collection<ScalingExecutorPlugin<KEY, Context>> customExecutors =
99100
AutoscalerUtils.discoverCustomScalingExecutors();
101+
Collection<ParallelismAlignmentMode> customAlignmentModes =
102+
AutoscalerUtils.discoverCustomAlignmentModes();
100103

101104
return new JobAutoScalerImpl<>(
102105
new RestApiMetricsCollector<>(),
103106
new ScalingMetricEvaluator(customEvaluators),
104-
new ScalingExecutor<>(eventHandler, stateStore, null, customExecutors),
107+
new ScalingExecutor<>(
108+
eventHandler, stateStore, null, customExecutors, customAlignmentModes),
105109
eventHandler,
106110
new RescaleApiScalingRealizer<>(eventHandler),
107111
stateStore);

0 commit comments

Comments
 (0)