-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[GOBBLIN-2185] Recommend GoT Dynamic Auto-Scaling using heuristics ba…
…sed on `WorkUnitsSizeSummary` (#4087) * Implement GoT Dynamic auto-scaling PoC of `WorkUnitsSizeSummary`-driven linear heuristic * Do not generate `@Setter`s for `@Data` POJOs, for which deserialization support prevents having `final` members * Align choice of directory between `FsScalingDirectivesRecipient` and `FsScalingDirectivesSource`, and ensure various handles get closed
- Loading branch information
Showing
43 changed files
with
1,085 additions
and
112 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
56 changes: 56 additions & 0 deletions
56
.../src/main/java/org/apache/gobblin/temporal/ddm/activity/RecommendScalingForWorkUnits.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.gobblin.temporal.ddm.activity; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Properties; | ||
|
|
||
| import io.temporal.activity.ActivityInterface; | ||
| import io.temporal.activity.ActivityMethod; | ||
|
|
||
| import org.apache.gobblin.source.workunit.WorkUnit; | ||
| import org.apache.gobblin.temporal.ddm.work.TimeBudget; | ||
| import org.apache.gobblin.temporal.ddm.work.WorkUnitsSizeSummary; | ||
| import org.apache.gobblin.temporal.dynamic.ScalingDirective; | ||
|
|
||
|
|
||
|
|
||
| /** | ||
| * Activity to suggest the Dynamic Scaling warranted to complete processing of some amount of {@link org.apache.gobblin.source.workunit.WorkUnit}s | ||
| * within {@link TimeBudget}, through a combination of Workforce auto-scaling and Worker right-sizing. | ||
| * | ||
| * As with all {@link ActivityInterface}s, this is stateless, so the {@link ScalingDirective}(s) returned "stand alone", presuming nothing of current | ||
| * {@link org.apache.gobblin.temporal.dynamic.WorkforceStaffing}. It thus falls to the caller to coordinate whether to apply the directive(s) as-is, | ||
| * or first to adjust in light of scaling levels already in the current {@link org.apache.gobblin.temporal.dynamic.WorkforcePlan}. | ||
| */ | ||
| @ActivityInterface | ||
| public interface RecommendScalingForWorkUnits { | ||
|
|
||
| /** | ||
| * Recommend the {@link ScalingDirective}s to process the {@link WorkUnit}s of {@link WorkUnitsSizeSummary} within {@link TimeBudget}. | ||
| * | ||
| * @param remainingWork may characterize a newly-generated batch of `WorkUnit`s for which no processing has yet begun - or be the sub-portion | ||
| * of an in-progress job that still awaits processing | ||
| * @param sourceClass contextualizes the `WorkUnitsSizeSummary` and should name a {@link org.apache.gobblin.source.Source} | ||
| * @param timeBudget the remaining target duration for processing the summarized `WorkUnit`s | ||
| * @param jobProps all job props, to either guide the recommendation or better contextualize the nature of the `remainingWork` | ||
| * @return the {@link ScalingDirective}s to process the summarized {@link WorkUnit}s within {@link TimeBudget} | ||
| */ | ||
| @ActivityMethod | ||
| List<ScalingDirective> recommendScaling(WorkUnitsSizeSummary remainingWork, String sourceClass, TimeBudget timeBudget, Properties jobProps); | ||
| } |
77 changes: 77 additions & 0 deletions
77
...g/apache/gobblin/temporal/ddm/activity/impl/AbstractRecommendScalingForWorkUnitsImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.gobblin.temporal.ddm.activity.impl; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.Properties; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import org.apache.gobblin.runtime.JobState; | ||
| import org.apache.gobblin.temporal.ddm.activity.RecommendScalingForWorkUnits; | ||
| import org.apache.gobblin.temporal.ddm.work.TimeBudget; | ||
| import org.apache.gobblin.temporal.ddm.work.WorkUnitsSizeSummary; | ||
| import org.apache.gobblin.temporal.dynamic.ProfileDerivation; | ||
| import org.apache.gobblin.temporal.dynamic.ProfileOverlay; | ||
| import org.apache.gobblin.temporal.dynamic.ScalingDirective; | ||
| import org.apache.gobblin.temporal.dynamic.WorkforceProfiles; | ||
|
|
||
|
|
||
| /** | ||
| * Skeletal impl handling all foundational concerns, but leaving it to a concrete impl to actually choose the auto-scaling | ||
| * {@link ScalingDirective#getSetPoint()} for the exactly one {@link ScalingDirective} being recommended. | ||
| */ | ||
| @Slf4j | ||
| public abstract class AbstractRecommendScalingForWorkUnitsImpl implements RecommendScalingForWorkUnits { | ||
|
|
||
| // TODO: decide whether this name ought to be configurable - or instead a predictable name that callers may expect (and possibly adjust) | ||
| public static final String DEFAULT_PROFILE_DERIVATION_NAME = "workUnitsProc"; | ||
|
|
||
| @Override | ||
| public List<ScalingDirective> recommendScaling(WorkUnitsSizeSummary remainingWork, String sourceClass, TimeBudget timeBudget, Properties jobProps) { | ||
| // NOTE: no attempt to determine the current scaling - per `RecommendScalingForWorkUnits` javadoc, the `ScalingDirective`(s) returned must "stand alone", | ||
| // presuming nothing of the current `WorkforcePlan`'s `WorkforceStaffing` | ||
| JobState jobState = new JobState(jobProps); | ||
| ScalingDirective procWUsWorkerScaling = new ScalingDirective( | ||
| calcProfileDerivationName(jobState), | ||
| calcDerivationSetPoint(remainingWork, sourceClass, timeBudget, jobState), | ||
| System.currentTimeMillis(), | ||
| Optional.of(calcProfileDerivation(calcBasisProfileName(jobState), remainingWork, sourceClass, jobState)) | ||
| ); | ||
| log.info("Recommended re-scaling to process work units: {}", procWUsWorkerScaling); | ||
| return Arrays.asList(procWUsWorkerScaling); | ||
| } | ||
|
|
||
| protected abstract int calcDerivationSetPoint(WorkUnitsSizeSummary remainingWork, String sourceClass, TimeBudget timeBudget, JobState jobState); | ||
|
|
||
| protected ProfileDerivation calcProfileDerivation(String basisProfileName, WorkUnitsSizeSummary remainingWork, String sourceClass, JobState jobState) { | ||
| // TODO: implement right-sizing!!! (for now just return unchanged) | ||
| return new ProfileDerivation(basisProfileName, ProfileOverlay.unchanged()); | ||
| } | ||
|
|
||
| protected String calcProfileDerivationName(JobState jobState) { | ||
| // TODO: if we ever return > 1 directive, append a monotonically increasing number to avoid collisions | ||
| return DEFAULT_PROFILE_DERIVATION_NAME; | ||
| } | ||
|
|
||
| protected String calcBasisProfileName(JobState jobState) { | ||
| return WorkforceProfiles.BASELINE_NAME; // always build upon baseline | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
87 changes: 87 additions & 0 deletions
87
...e/gobblin/temporal/ddm/activity/impl/RecommendScalingForWorkUnitsLinearHeuristicImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.gobblin.temporal.ddm.activity.impl; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import org.apache.gobblin.runtime.JobState; | ||
| import org.apache.gobblin.temporal.GobblinTemporalConfigurationKeys; | ||
| import org.apache.gobblin.temporal.ddm.work.TimeBudget; | ||
| import org.apache.gobblin.temporal.ddm.work.WorkUnitsSizeSummary; | ||
| import org.apache.gobblin.temporal.ddm.worker.WorkFulfillmentWorker; | ||
|
|
||
|
|
||
| /** | ||
| * Simple config-driven linear recommendation for how many containers to use to complete the "remaining work" within a given {@link TimeBudget}, per: | ||
| * | ||
| * a. from {@link WorkUnitsSizeSummary}, find how many (remaining) "top-level" {@link org.apache.gobblin.source.workunit.MultiWorkUnit}s of some mean size | ||
| * b. from the configured {@link #AMORTIZED_NUM_BYTES_PER_MINUTE}, find the expected "processing rate" in bytes / minute | ||
| * 1. estimate the time required for processing a mean-sized `MultiWorkUnit` (MWU) | ||
| * c. from {@link JobState}, find per-container `MultiWorkUnit` parallelism capacity (aka. "worker-slots") to base the recommendation upon | ||
| * 2. calculate the per-container throughput of MWUs per minute | ||
| * 3. estimate the total per-container-minutes required to process all MWUs | ||
| * d. from the {@link TimeBudget}, find the target number of minutes in which to complete processing of all MWUs | ||
| * 4. recommend the number of containers so all MWU processing should finish within the target number of minutes | ||
| */ | ||
| @Slf4j | ||
| public class RecommendScalingForWorkUnitsLinearHeuristicImpl extends AbstractRecommendScalingForWorkUnitsImpl { | ||
|
|
||
| public static final String AMORTIZED_NUM_BYTES_PER_MINUTE = GobblinTemporalConfigurationKeys.DYNAMIC_SCALING_PREFIX + "heuristic.params.numBytesPerMinute"; | ||
| public static final long DEFAULT_AMORTIZED_NUM_BYTES_PER_MINUTE = 80 * 1000L * 1000L * 60L; // 80MB/sec | ||
|
|
||
| @Override | ||
| protected int calcDerivationSetPoint(WorkUnitsSizeSummary remainingWork, String sourceClass, TimeBudget jobTimeBudget, JobState jobState) { | ||
| // for simplicity, for now, consider only top-level work units (aka. `MultiWorkUnit`s - MWUs) | ||
| long numMWUs = remainingWork.getTopLevelWorkUnitsCount(); | ||
| double meanBytesPerMWU = remainingWork.getTopLevelWorkUnitsMeanSize(); | ||
| int numSimultaneousMWUsPerContainer = calcPerContainerWUCapacity(jobState); // (a worker-thread is a slot for top-level (MWUs) - not constituent sub-WUs) | ||
| long bytesPerMinuteProcRate = calcAmortizedBytesPerMinute(jobState); | ||
| log.info("Calculating auto-scaling (for {} remaining work units within {}) using: bytesPerMinuteProcRate = {}; meanBytesPerMWU = {}", | ||
| numMWUs, jobTimeBudget, bytesPerMinuteProcRate, meanBytesPerMWU); | ||
|
|
||
| // calc how many container*minutes to process all MWUs, based on mean MWU size | ||
| double minutesProcTimeForMeanMWU = meanBytesPerMWU / bytesPerMinuteProcRate; | ||
| double meanMWUsThroughputPerContainerMinute = numSimultaneousMWUsPerContainer / minutesProcTimeForMeanMWU; | ||
| double estContainerMinutesForAllMWUs = numMWUs / meanMWUsThroughputPerContainerMinute; | ||
|
|
||
| long targetNumMinutesForAllMWUs = jobTimeBudget.getMaxTargetDurationMinutes(); | ||
| // TODO: take into account `jobTimeBudget.getPermittedOverageMinutes()` - e.g. to decide whether to use `Math.ceil` vs. `Math.floor` | ||
|
|
||
| // TODO: decide how to account for container startup; working est. for GoT-on-YARN ~ 3 mins (req to alloc ~ 30s; alloc to workers ready ~ 2.5m) | ||
| // e.g. can we amortize away / ignore when `targetNumMinutesForAllMWUs >> workerRequestToReadyNumMinutes`? | ||
| // TODO take into account that MWUs are quantized into discrete chunks; this est. uses avg and presumes to divide partial MWUs amongst workers | ||
| // can we we mostly ignore if we keep MWU "chunk size" "small-ish", like maybe even just `duration(max(MWU)) <= targetNumMinutesForAllMWUs/2)`? | ||
|
|
||
| int recommendedNumContainers = (int) Math.floor(estContainerMinutesForAllMWUs / targetNumMinutesForAllMWUs); | ||
| log.info("Recommended auto-scaling: {} containers, given: minutesToProc(mean(MWUs)) = {}; throughput = {} (MWUs / container*minute); " | ||
| + "est. container*minutes to complete ALL ({}) MWUs = {}", | ||
| recommendedNumContainers, minutesProcTimeForMeanMWU, meanMWUsThroughputPerContainerMinute, numMWUs, estContainerMinutesForAllMWUs); | ||
| return recommendedNumContainers; | ||
| } | ||
|
|
||
| protected int calcPerContainerWUCapacity(JobState jobState) { | ||
| int numWorkersPerContainer = jobState.getPropAsInt(GobblinTemporalConfigurationKeys.TEMPORAL_NUM_WORKERS_PER_CONTAINER, | ||
| GobblinTemporalConfigurationKeys.DEFAULT_TEMPORAL_NUM_WORKERS_PER_CONTAINERS); | ||
| int numThreadsPerWorker = WorkFulfillmentWorker.MAX_EXECUTION_CONCURRENCY; // TODO: get from config, once that's implemented | ||
| return numWorkersPerContainer * numThreadsPerWorker; | ||
| } | ||
|
|
||
| protected long calcAmortizedBytesPerMinute(JobState jobState) { | ||
| return jobState.getPropAsLong(AMORTIZED_NUM_BYTES_PER_MINUTE, DEFAULT_AMORTIZED_NUM_BYTES_PER_MINUTE); | ||
| } | ||
| } |
Oops, something went wrong.