Druid Segment Ops Segment & Ingestion Orchestration

Historical Tier Balancing Strategies in Apache Druid

Once load rules decide how many replicas a segment needs and which tier they belong to, the balancer decides which specific Historical in that tier receives each one — and whether to relocate already-loaded segments to keep disk and query load even. That choice, governed by a pluggable balancing strategy and a small set of move-budget knobs, is the difference between a tier where every node carries its share and one where a single Historical is a latency hotspot. This guide, part of Apache Druid segment replication, balancing & high availability, covers the cost, cachingCost, and random strategies, the maxSegmentsToMove budget, round-robin assignment, and how smartSegmentLoading now automates most of it.

The cost balancer scoring candidate Historicals for a new segment A new segment to place enters the cost balancer, which scores each Historical in the tier. Historical A already holds segments from the same datasource and adjacent time intervals, giving it a high placement cost. Historical B holds unrelated segments, giving a low cost. Historical C is nearly full, giving a high cost. The balancer selects Historical B, the lowest-cost host, spreading the datasource across the tier. New segment to place cost balancer scores each host picks the minimum Historical A same datasource + adjacent time Historical B unrelated segments Historical C nearly full high cost low cost → chosen high cost

Mechanics & Internals

Balancing runs as a Coordinator duty every run, and it does two related jobs: assignment (which Historical receives a segment that must be newly loaded) and moving (relocating an already-loaded segment from one Historical to another in the same tier to even out utilization). Both are driven by the active balancer strategy, historically set with the runtime property druid.coordinator.balancer.strategy, though in current Druid the default smartSegmentLoading mode pins it to cost and manages the surrounding knobs for you.

The cost strategy is the production default and the one to understand. For a candidate placement it computes a joint cost that sums, over the segments already on that Historical, a function that grows when the candidate segment shares a datasource with, and lies close in time to, the resident segments. The intuition: two segments from the same datasource covering adjacent intervals are likely to be scanned by the same query, so co-locating them concentrates that query's work on one node. By assigning a high cost to such co-location, the balancer naturally spreads a datasource's segments — and therefore its query fan-out — across the whole tier, which is what keeps scatter/gather parallel and avoids hotspots. The balancer picks the lowest-cost host for each assignment and, for moving, looks for segments whose relocation most reduces total cost.

The other strategies trade that quality for speed or simplicity:

  • cachingCost computes the same cost function but caches intermediate per-segment costs so it scales to very large clusters where the naive cost computation becomes CPU-bound. It approximates cost closely and is the right choice when coordinator/time is dominated by balancing math rather than I/O.
  • random assigns and moves segments to uniformly random hosts. It is cheap and spreads count evenly, but ignores datasource/time locality, so it can co-locate query-correlated segments; use it only for debugging or truly homogeneous workloads.

Two further mechanisms shape assignment. Round-robin assignment (useRoundRobinSegmentAssignment, auto-enabled by smartSegmentLoading on large clusters) skips the cost computation for initial placement — it drops each new segment on the next host in rotation — and relies on later balancing moves to correct any imbalance. This decouples the expensive cost math from the hot path of loading many new segments at once, a large speedup during bulk ingestion or recovery. Separately, the batched segment sampler governs how the balancer picks candidate segments to consider moving each run; in older Druid this was the useBatchedSegmentSampler dynamic-config flag (default on), but recent versions removed the toggle and made the batched sampler the only, always-on behaviour — so on current clusters there is nothing to set.

The move budget is what keeps balancing from thrashing. maxSegmentsToMove caps relocations per run; every move copies a segment to a new host and drops the old copy only after the new one confirms load, so each move costs deep-storage read bandwidth and a moment of extra replication. Too high a budget on a busy or flapping cluster produces a rebalancing storm — the failure dissected in diagnosing segment balancing storms. With smartSegmentLoading on, maxSegmentsToMove is auto-sized; pin it manually only after measuring a specific convergence problem. Balancing also interacts with placement: it only ever moves a replica to a different host in the same tier, never violating the distinct-host invariant that makes replication fault-tolerant per segment replication and fault tolerance.

Validated Configuration Spec

Balancing is configured almost entirely through the Coordinator dynamic config at GET/POST /druid/coordinator/v1/config. The document below shows the balancing-relevant fields with smartSegmentLoading left on, the recommended default:

{
  "smartSegmentLoading": true,
  "maxSegmentsToMove": 100,
  "balancerComputeThreads": 2,
  "useRoundRobinSegmentAssignment": true,
  "maxSegmentsInNodeLoadingQueue": 500,
  "decommissioningNodes": [],
  "decommissioningMaxPercentOfMaxSegmentsToMove": 70
}
  • smartSegmentLoading — when true, auto-computes maxSegmentsToMove, useRoundRobinSegmentAssignment, and the load-queue caps from cluster size, and forces the cost strategy. Any explicit values you set for those fields are ignored while it is on; the fields are shown here for the manual case.
  • maxSegmentsToMove — the per-run relocation cap. Effective only with smartSegmentLoading: false; raise it to converge a lopsided tier faster, lower it to calm a storm.
  • balancerComputeThreads — parallelism for the cost/cachingCost computation. Increase on large clusters where balancing is CPU-bound; irrelevant for random.
  • useRoundRobinSegmentAssignment — cheap round-robin initial placement, corrected later by moves. A major speedup for bulk loads.
  • decommissioningNodes / decommissioningMaxPercentOfMaxSegmentsToMove — mark nodes to drain and bound how much of the move budget the drain consumes, so evacuating a node never starves normal balancing.

To pin the strategy manually — for example to switch a very large cluster to cachingCost — set smartSegmentLoading off and declare the strategy in the Coordinator's runtime.properties:

druid.coordinator.balancer.strategy=cachingCost
druid.coordinator.period=PT60S

The random strategy is available the same way (=random) but is for diagnostics only. Full field semantics are maintained in the Apache Druid coordinator documentation.

Sizing Heuristics & Formulas

The balancer's target is even utilization. For a tier of $N$ Historicals holding a total of $B$ bytes across $S$ segments, the ideal per-node load is $B / N$ bytes, and the imbalance you are trying to eliminate is the spread of actual loads around that mean. A useful health metric is the coefficient of variation of per-node used bytes:

$$\text{CV} = \frac{\sigma(\text{usedBytes})}{\bar{x}(\text{usedBytes})}$$

A well-balanced tier sits near $\text{CV} \lesssim 0.05$ (loads within ~5% of each other); a CV climbing toward 0.2 signals a hotspot the balancer has not yet corrected — often because the move budget is too small or a node was recently added.

Convergence time after an imbalance (a new node, a recovered node, a large ingest) is bounded by the move budget. To relocate the $M$ segments needed to level the tier at maxSegmentsToMove $= m$ per run of period $p$:

$$t_{\text{level}} \approx \left\lceil \frac{M}{m} \right\rceil \times p$$

Adding one empty node to a tier of $N$ nodes requires moving roughly $B / (N + 1)$ bytes onto it to reach balance; at average segment size $b$ that is about $M \approx B / \big((N+1),b\big)$ segments. For a 10 TB tier growing from 5 to 6 nodes with 500 MB segments, $M \approx 10,000,000 / (6 \times 500) \approx 3333$ segments; at $m = 100$ and $p = 60\text{s}$ that levels in $\lceil 3333/100 \rceil \times 60 \approx 2000\text{s}$, about 33 minutes. Raising $m$ shortens this linearly until deep-storage read throughput becomes the binding constraint — the same dual bottleneck that governs re-replication.

The cost of the cost strategy itself scales with segment count. Because it scores candidate placements against resident segments, its per-run compute grows roughly with $S \times \log S$ for the sampled candidate set; once coordinator/time is dominated by this term rather than by load-queue I/O, switch to cachingCost or raise balancerComputeThreads.

Python Orchestration Snippet

The driver below evaluates tier balance from the live server view, and — when the coefficient of variation exceeds a threshold — temporarily raises the move budget to accelerate levelling, then polls until balanced and restores the original config. It reads and writes the dynamic config with capped exponential backoff, using only the standard library plus requests.

import time
import statistics
import requests

COORDINATOR = "http://coordinator:8081"


def tier_used_bytes(tier: str) -> list[int]:
    r = requests.get(f"{COORDINATOR}/druid/coordinator/v1/servers?simple", timeout=30)
    r.raise_for_status()
    return [s["currSize"] for s in r.json() if s.get("tier") == tier]


def coeff_of_variation(values: list[int]) -> float:
    if len(values) < 2:
        return 0.0
    mean = statistics.mean(values)
    return statistics.pstdev(values) / mean if mean else 0.0


def get_config() -> dict:
    r = requests.get(f"{COORDINATOR}/druid/coordinator/v1/config", timeout=30)
    r.raise_for_status()
    return r.json()


def set_config(patch: dict) -> None:
    cfg = get_config()
    cfg.update(patch)
    r = requests.post(
        f"{COORDINATOR}/druid/coordinator/v1/config",
        json=cfg,
        headers={"Content-Type": "application/json",
                 "X-Druid-Author": "balance-bot",
                 "X-Druid-Comment": "adjust move budget"},
        timeout=30,
    )
    r.raise_for_status()


def rebalance_tier(tier: str, target_cv: float = 0.05,
                   boosted_moves: int = 500, max_wait: float = 3600.0) -> None:
    if coeff_of_variation(tier_used_bytes(tier)) <= target_cv:
        print(f"tier {tier} already balanced")
        return
    saved = get_config()
    original = {"smartSegmentLoading": saved["smartSegmentLoading"],
                "maxSegmentsToMove": saved["maxSegmentsToMove"]}
    set_config({"smartSegmentLoading": False, "maxSegmentsToMove": boosted_moves})
    try:
        delay, waited = 30.0, 0.0
        while waited < max_wait:
            cv = coeff_of_variation(tier_used_bytes(tier))
            print(f"t+{int(waited):>4}s  tier {tier} CV={cv:.3f}")
            if cv <= target_cv:
                print(f"tier {tier} balanced (CV={cv:.3f})")
                return
            time.sleep(delay)
            waited += delay
            delay = min(delay * 2, 120.0)  # capped exponential backoff
        raise RuntimeError(f"tier {tier} did not balance within budget")
    finally:
        set_config(original)  # always restore, even on failure


if __name__ == "__main__":
    rebalance_tier("hot")

Two safeguards matter here. The finally block always restores the original config, so a boosted move budget never leaks into steady state and becomes a latent storm risk. And the driver only intervenes when CV actually exceeds the target — it never moves segments the balancer would have left alone, respecting the same "don't act on assumed state" discipline the replication health gate applies to availability.

Failure Modes & Diagnostics

1. One Historical is a latency hotspot while peers idle. The tier is imbalanced — check per-node used bytes and compute the spread:

curl -s "http://coordinator:8081/druid/coordinator/v1/servers?simple" \
  | jq '[.[] | select(.tier=="hot") | {host, gb: ((.currSize/1073741824)|floor)}] | sort_by(.gb)'

A wide spread with a small maxSegmentsToMove means the balancer is under-budgeted; raise it (with smartSegmentLoading off) or let it converge over more runs.

2. Coordinator run time climbing, dominated by balancing. The cost computation is CPU-bound on segment count. Inspect the duty timing and segment volume:

curl -s "http://coordinator:8081/druid/coordinator/v1/config" \
  | jq '{smartSegmentLoading, maxSegmentsToMove, balancerComputeThreads}'

If coordinator/time is high and segment count large, switch to cachingCost or raise balancerComputeThreads. Runaway moves specifically are the storm scenario in diagnosing segment balancing storms.

3. Newly added node stays nearly empty. Round-robin assignment plus a small move budget fills it slowly. Watch its load queue drain:

curl -s "http://coordinator:8081/druid/coordinator/v1/loadqueue?simple" \
  | jq 'to_entries[] | select(.key | test("hist-hot-new")) | {host: .key, queued: .value}'

A steadily shrinking queue is healthy; a flat empty queue with the node still under-filled means the balancer sees no low-cost moves onto it — usually fine, it fills as new segments arrive.

4. Segments constantly moving with no net improvement. A flapping Historical repeatedly leaves and rejoins, forcing re-evaluation each cycle. Correlate segment/moved/count with node announcements and stabilize the flapping node before touching balancer config.

Automation Checklist

Wire these into the pipeline that manages balancing so the tier stays even without thrashing.

Up one level: Apache Druid segment replication, balancing & high availability frames balancing as the placement half of the reconciliation loop.

Diagnosing Segment Balancing Storms Read more

Back to Apache Druid Segment Replication, Balancing & High Availability