You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2023/02/16 13:51:11 UTC
[flink] branch master updated: [FLINK-30895][coordination] Dynamically adjust slot distribution
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 97e1dbc90cc [FLINK-30895][coordination] Dynamically adjust slot distribution
97e1dbc90cc is described below
commit 97e1dbc90cc3ea6e6e48e901813ab3bfbb693c6a
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Thu Feb 16 14:51:03 2023 +0100
[FLINK-30895][coordination] Dynamically adjust slot distribution
---
docs/content.zh/docs/deployment/elastic_scaling.md | 1 -
docs/content/docs/deployment/elastic_scaling.md | 1 -
.../scheduler/adaptive/JobGraphJobInformation.java | 1 +
.../adaptive/allocator/JobInformation.java | 2 +
.../allocator/SlotSharingSlotAllocator.java | 55 +++++++++++++++++++---
.../allocator/SlotSharingSlotAllocatorTest.java | 40 ++++++++++++++++
6 files changed, 91 insertions(+), 9 deletions(-)
diff --git a/docs/content.zh/docs/deployment/elastic_scaling.md b/docs/content.zh/docs/deployment/elastic_scaling.md
index c65c974a35b..1df55d738db 100644
--- a/docs/content.zh/docs/deployment/elastic_scaling.md
+++ b/docs/content.zh/docs/deployment/elastic_scaling.md
@@ -145,7 +145,6 @@ Adaptive 调度器可以通过[所有在名字包含 `adaptive-scheduler` 的配
- **只支持流式 Job**：Adaptive 调度器的第一个版本仅支持流式 Job。当提交的是一个批处理 Job 时，我们会自动换回默认调度器。
- **不支持[本地恢复]({{< ref "docs/ops/state/large_state_tuning">}}#task-local-recovery)**：本地恢复是将 Task 调度到状态尽可能的被重用的机器上的功能。不支持这个功能意味着 Adaptive 调度器需要每次从 Checkpoint 的存储中下载整个 State。
- **不支持部分故障恢复**: 部分故障恢复意味着调度器可以只重启失败 Job 其中某一部分（在 Flink 的内部结构中被称之为 Region）而不是重启整个 Job。这个限制只会影响那些独立并行（Embarrassingly Parallel）Job的恢复时长，默认的调度器可以重启失败的部分，然而 Adaptive 将需要重启整个 Job。
-- **空闲 Slot**: 如果 Slot 共享组的最大并行度不相等，提供给 Adaptive 调度器所使用的的 Slot 可能不会被使用。
- 扩缩容事件会触发 Job 和 Task 重启，Task 重试的次数也会增加。
## Adaptive Batch Scheduler
diff --git a/docs/content/docs/deployment/elastic_scaling.md b/docs/content/docs/deployment/elastic_scaling.md
index b11e459f294..0acf86ad910 100644
--- a/docs/content/docs/deployment/elastic_scaling.md
+++ b/docs/content/docs/deployment/elastic_scaling.md
@@ -147,7 +147,6 @@ The behavior of Adaptive Scheduler is configured by [all configuration options c
- **Streaming jobs only**: The first version of Adaptive Scheduler runs with streaming jobs only. When submitting a batch job, we will automatically fall back to the default scheduler.
- **No support for [local recovery]({{< ref "docs/ops/state/large_state_tuning">}}#task-local-recovery)**: Local recovery is a feature that schedules tasks to machines so that the state on that machine gets re-used if possible. The lack of this feature means that Adaptive Scheduler will always need to download the entire state from the checkpoint storage.
- **No support for partial failover**: Partial failover means that the scheduler is able to restart parts ("regions" in Flink's internals) of a failed job, instead of the entire job. This limitation impacts only recovery time of embarrassingly parallel jobs: Flink's default scheduler can restart failed parts, while Adaptive Scheduler will restart the entire job.
-- **Unused slots**: If the max parallelism for slot sharing groups is not equal, slots offered to Adaptive Scheduler might be unused.
- Scaling events trigger job and task restarts, which will increase the number of Task attempts.
## Adaptive Batch Scheduler
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobGraphJobInformation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobGraphJobInformation.java
index 1e265fb9352..1b0bb60e7e6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobGraphJobInformation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobGraphJobInformation.java
@@ -73,6 +73,7 @@ public class JobGraphJobInformation implements JobInformation {
return jobGraph.getCheckpointingSettings();
}
+ @Override
public Iterable<JobInformation.VertexInformation> getVertices() {
return Iterables.transform(
jobGraph.getVertices(), (vertex) -> getVertexInformation(vertex.getID()));
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
index 074dd6d15c5..99b8d48c051 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
@@ -36,6 +36,8 @@ public interface JobInformation {
VertexInformation getVertexInformation(JobVertexID jobVertexId);
+ Iterable<VertexInformation> getVertices();
+
/** Information about a single vertex. */
interface VertexInformation {
JobVertexID getJobVertexID();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java
index edc3b7f05c8..ce4c6bfb1af 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java
@@ -34,6 +34,7 @@ import javax.annotation.Nonnull;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -94,15 +95,15 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
@Override
public Optional<VertexParallelismWithSlotSharing> determineParallelism(
JobInformation jobInformation, Collection<? extends SlotInfo> freeSlots) {
- // TODO: This can waste slots if the max parallelism for slot sharing groups is not equal
- final int slotsPerSlotSharingGroup =
- freeSlots.size() / jobInformation.getSlotSharingGroups().size();
- if (slotsPerSlotSharingGroup == 0) {
- // => less slots than slot-sharing groups
+ // => less slots than slot-sharing groups
+ if (jobInformation.getSlotSharingGroups().size() > freeSlots.size()) {
return Optional.empty();
}
+ final Map<SlotSharingGroupId, Integer> slotSharingGroupParallelism =
+ determineSlotsPerSharingGroup(jobInformation, freeSlots.size());
+
final Iterator<? extends SlotInfo> slotIterator = freeSlots.iterator();
final Collection<ExecutionSlotSharingGroupAndSlot> assignments = new ArrayList<>();
@@ -115,7 +116,10 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
.collect(Collectors.toList());
final Map<JobVertexID, Integer> vertexParallelism =
- determineParallelism(containedJobVertices, slotsPerSlotSharingGroup);
+ determineVertexParallelism(
+ containedJobVertices,
+ slotSharingGroupParallelism.get(
+ slotSharingGroup.getSlotSharingGroupId()));
final Iterable<ExecutionSlotSharingGroup> sharedSlotToVertexAssignment =
createExecutionSlotSharingGroups(vertexParallelism);
@@ -133,7 +137,44 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
return Optional.of(new VertexParallelismWithSlotSharing(allVertexParallelism, assignments));
}
- private static Map<JobVertexID, Integer> determineParallelism(
+ /**
+ * Distributes free slots across the slot-sharing groups of the job. Slots are distributed as
+ * evenly as possible. If a group requires less than an even share of slots the remainder is
+ * distributed over the remaining groups.
+ */
+ private static Map<SlotSharingGroupId, Integer> determineSlotsPerSharingGroup(
+ JobInformation jobInformation, int freeSlots) {
+ int numUnassignedSlots = freeSlots;
+ int numUnassignedSlotSharingGroups = jobInformation.getSlotSharingGroups().size();
+
+ final Map<SlotSharingGroupId, Integer> slotSharingGroupParallelism = new HashMap<>();
+
+ for (Map.Entry<SlotSharingGroupId, Integer> slotSharingGroup :
+ sortSlotSharingGroupsByDesiredParallelism(jobInformation)) {
+ final int groupParallelism =
+ Math.min(
+ slotSharingGroup.getValue(),
+ numUnassignedSlots / numUnassignedSlotSharingGroups);
+
+ slotSharingGroupParallelism.put(slotSharingGroup.getKey(), groupParallelism);
+
+ numUnassignedSlots -= groupParallelism;
+ numUnassignedSlotSharingGroups--;
+ }
+
+ return slotSharingGroupParallelism;
+ }
+
+ private static List<Map.Entry<SlotSharingGroupId, Integer>>
+ sortSlotSharingGroupsByDesiredParallelism(JobInformation jobInformation) {
+
+ return getMaxParallelismForSlotSharingGroups(jobInformation.getVertices()).entrySet()
+ .stream()
+ .sorted(Comparator.comparingInt(Map.Entry::getValue))
+ .collect(Collectors.toList());
+ }
+
+ private static Map<JobVertexID, Integer> determineVertexParallelism(
Collection<JobInformation.VertexInformation> containedJobVertices, int availableSlots) {
final Map<JobVertexID, Integer> vertexParallelism = new HashMap<>();
for (JobInformation.VertexInformation jobVertex : containedJobVertices) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java
index a8dc02fa49a..8be500919f1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.util.ResourceCounter;
import org.apache.flink.util.TestLogger;
+import org.assertj.core.api.Assertions;
import org.junit.Test;
import java.util.ArrayList;
@@ -134,6 +135,40 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
is(vertex3.getParallelism()));
}
+ @Test
+ public void testDetermineParallelismWithVariedParallelism() {
+ final SlotSharingSlotAllocator slotAllocator =
+ SlotSharingSlotAllocator.createSlotSharingSlotAllocator(
+ TEST_RESERVE_SLOT_FUNCTION,
+ TEST_FREE_SLOT_FUNCTION,
+ TEST_IS_SLOT_FREE_FUNCTION);
+ final SlotSharingGroup slotSharingGroup1 = new SlotSharingGroup();
+ final JobInformation.VertexInformation vertex11 =
+ new TestVertexInformation(new JobVertexID(), 4, slotSharingGroup1);
+ final JobInformation.VertexInformation vertex12 =
+ new TestVertexInformation(new JobVertexID(), 1, slotSharingGroup1);
+ final JobInformation.VertexInformation vertex2 =
+ new TestVertexInformation(new JobVertexID(), 2, new SlotSharingGroup());
+
+ TestJobInformation testJobInformation =
+ new TestJobInformation(Arrays.asList(vertex11, vertex12, vertex2));
+
+ Map<JobVertexID, Integer> maxParallelismForVertices =
+ slotAllocator
+ .determineParallelism(
+ testJobInformation,
+ getSlots(vertex11.getParallelism() + vertex2.getParallelism()))
+ .get()
+ .getMaxParallelismForVertices();
+
+ Assertions.assertThat(maxParallelismForVertices.get(vertex11.getJobVertexID()))
+ .isEqualTo(vertex11.getParallelism());
+ Assertions.assertThat(maxParallelismForVertices.get(vertex12.getJobVertexID()))
+ .isEqualTo(vertex12.getParallelism());
+ Assertions.assertThat(maxParallelismForVertices.get(vertex2.getJobVertexID()))
+ .isEqualTo(vertex2.getParallelism());
+ }
+
@Test
public void testDetermineParallelismUnsuccessfulWithLessSlotsThanSlotSharingGroups() {
final SlotSharingSlotAllocator slotAllocator =
@@ -243,6 +278,11 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
public VertexInformation getVertexInformation(JobVertexID jobVertexId) {
return vertexIdToInformation.get(jobVertexId);
}
+
+ @Override
+ public Iterable<VertexInformation> getVertices() {
+ return vertexIdToInformation.values();
+ }
}
private static class TestVertexInformation implements JobInformation.VertexInformation {