kfaraz commented on code in PR #14584: URL: https://github.com/apache/druid/pull/14584#discussion_r1294144871
########## server/src/main/java/org/apache/druid/server/coordinator/balancer/SegmentToMoveCalculator.java: ########## @@ -0,0 +1,309 @@ +/* + * 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.druid.server.coordinator.balancer; + +import com.google.common.base.Preconditions; +import it.unimi.dsi.fastutil.objects.Object2IntMap; +import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.server.coordinator.SegmentCountsPerInterval; +import org.apache.druid.server.coordinator.ServerHolder; + +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.stream.Collectors; + +/** + * Calculates the maximum, minimum and required number of segments to move in a + * Coordinator run for balancing. + */ +public class SegmentToMoveCalculator +{ + /** + * At least this number of segments must be picked for moving in every cycle + * to keep the cluster well balanced. + */ + private static final int MIN_SEGMENTS_TO_MOVE = 100; + + private static final Logger log = new Logger(SegmentToMoveCalculator.class); + + /** + * Calculates the number of segments to be picked for moving in the given tier, + * based on the level of skew between the historicals in the tier. + * + * @param tier Name of tier used for logging purposes + * @param historicals Active historicals in tier + * @param maxSegmentsToMoveInTier Maximum number of segments allowed to be moved + * in the tier. + * @return Number of segments to move in the tier in the range + * [{@link #MIN_SEGMENTS_TO_MOVE}, {@code maxSegmentsToMoveInTier}]. + */ + public static int computeNumSegmentsToMoveInTier( + String tier, + List<ServerHolder> historicals, + int maxSegmentsToMoveInTier + ) + { + final int totalSegments = historicals.stream().mapToInt( + server -> server.getProjectedSegments().getTotalSegmentCount() + ).sum(); + + // Move at least some segments to ensure that the cluster is always balancing itself + final int minSegmentsToMove = SegmentToMoveCalculator + .computeMinSegmentsToMoveInTier(totalSegments); + final int segmentsToMoveToFixDeviation = SegmentToMoveCalculator + .computeNumSegmentsToMoveToBalanceTier(tier, historicals); + log.info( + "Need to move [%,d] segments in tier[%s] to attain balance. Allowed values are [min=%d, max=%d].", + segmentsToMoveToFixDeviation, tier, minSegmentsToMove, maxSegmentsToMoveInTier + ); + + final int activeSegmentsToMove = Math.max(minSegmentsToMove, segmentsToMoveToFixDeviation); + return Math.min(activeSegmentsToMove, maxSegmentsToMoveInTier); + } + + /** + * Calculates the minimum number of segments that should be considered for + * moving in a tier, so that the cluster is always balancing itself. + * <p> + * This value must be calculated separately for every tier. + * + * @param totalSegmentsInTier Total number of all replicas of all segments + * loaded or queued across all historicals in the tier. + * @return {@code minSegmentsToMoveInTier} in the range + * [{@link #MIN_SEGMENTS_TO_MOVE}, {@code ~0.6% of totalSegmentsInTier}]. + */ + public static int computeMinSegmentsToMoveInTier(int totalSegmentsInTier) + { + // Divide by 2^14 and multiply by 100 so that the value increases + // in steps of 100 for every 2^14 = ~16k segments + int upperBound = (totalSegmentsInTier >> 14) * 100; + int lowerBound = Math.min(MIN_SEGMENTS_TO_MOVE, totalSegmentsInTier); + return Math.max(lowerBound, upperBound); + } Review Comment: Yeah, that would make sense. The only problem is that such a threshold would have to be dynamic too. A cluster with fewer segments results in smaller costs. So the threshold would have to increase as the number of segments in the cluster increases. As a compromise, I can reduce the `minSegmentsToMove` even further. I tested these changes on a cluster with 900k segments and `minSegmentsToMove` was about 6500. The good thing is that, even out of 6500, the cluster would decide to move only about 100 segments as the rest were already well placed. <img width="988" alt="Screenshot 2023-08-15 at 8 43 26 AM" src="https://github.com/apache/druid/assets/18635897/8f214727-6f28-4194-8a26-bd5446d94514";> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
