Tuning Compaction Input Segment Size

Auto-compaction can misbehave in two symmetric ways: it either refuses to touch a badly fragmented interval, or it churns endlessly, rewriting the same recent data every cycle and fighting live ingestion for locks. Both are input-selection problems — decided by three knobs that control which segments a compaction task pulls in and how big the output should be: inputSegmentSizeBytes (a per-task byte ceiling), skipOffsetFromLatest (a protected recent window), and targetRowsPerSegment / maxRowsPerSegment (the output shape). Getting them right is the difference between a compaction duty that quietly keeps the tail in shape and one that either starves or loops. This page sits under compaction threshold tuning and focuses specifically on the input side of the compaction decision — what the duty selects, and what it deliberately leaves alone.

Failure Modes & Diagnostics #

Input-selection misconfiguration is quiet: the API returns 200, the duty runs, and yet a datasource either never converges or re-compacts forever. Diagnose from the shell against the live Coordinator and Overlord.

# Is there a backlog, and is it shrinking across polls? A flat non-zero backlog = starved selection.
curl -s "http://druid-coordinator:8081/druid/coordinator/v1/compaction/status" \
  | jq '.latestStatus[] | select(.dataSource=="analytics_events")
        | {scheduleStatus, bytesAwaitingCompaction, bytesCompacted, bytesSkipped}'

# Fragment count per interval — spot the intervals the duty is failing to consolidate
curl -s "http://druid-coordinator:8081/druid/coordinator/v1/datasources/analytics_events/intervals?full" \
  | jq 'to_entries | map({interval: .key, count: .value.count}) | sort_by(-.count) | .[0:5]'

The four input-side failures:

1. A fragmented interval is never compacted. Symptom: bytesSkipped is high and an interval keeps a large fragment count. Root cause: the combined bytes of that interval's segments exceed inputSegmentSizeBytes, so the duty skips it as too large for a single task. Remediation: raise inputSegmentSizeBytes so the interval fits, or use a coarser segmentGranularity so each interval holds less data.

2. The duty re-compacts the same recent interval every cycle. Symptom: interval/compacted/count climbs but the newest intervals never settle, and compact tasks contend with ingestion for locks. Root cause: skipOffsetFromLatest is smaller than real ingestion lag, so the duty keeps compacting an interval that ingestion is still writing. Remediation: raise skipOffsetFromLatest above the worst-case ingestion watermark lag.

3. An infinite compaction loop on a settled interval. Symptom: the same old interval is rewritten every cycle even though ingestion is done with it. Root cause: targetRowsPerSegment is set so low the output can never satisfy it, so the duty always thinks the interval needs more work. Remediation: raise the row target until output clears the compaction floor; measure real bytes-per-row first (below).

4. TaskLock contention. Symptom: compact task logs show Cannot acquire lock on an interval. Root cause: skipOffsetFromLatest overlaps the interval ingestion still holds. Remediation: the same as failure 2 — widen the skip window.

# Measure real average bytes-per-row so targetRowsPerSegment lands output in the size band
curl -s "http://druid-coordinator:8081/druid/coordinator/v1/datasources/analytics_events/segments?full" \
  | jq '[.[] | {rows: .num_rows, size: .size}]
        | (map(.size)|add) as $b | (map(.rows)|add) as $r
        | {avg_row_bytes: ($b/$r), total_mb: ($b/1048576)}'

Target Spec & Validated JSON #

The input knobs live in the DataSourceCompactionConfig POSTed to POST /druid/coordinator/v1/config/compaction. The block below sets all three deliberately and is copy-ready against current stable Druid; every field is annotated.

{
  "dataSource": "analytics_events",
  "taskPriority": 25,
  "inputSegmentSizeBytes": 419430400,
  "skipOffsetFromLatest": "P1D",
  "tuningConfig": {
    "partitionsSpec": {
      "type": "dynamic",
      "maxRowsPerSegment": 5000000,
      "maxTotalRows": 20000000
    },
    "maxNumConcurrentSubTasks": 4
  },
  "granularitySpec": {
    "segmentGranularity": "DAY",
    "queryGranularity": "HOUR",
    "rollup": true
  },
  "ioConfig": {
    "dropExisting": false
  }
}

• inputSegmentSizeBytes — the maximum total bytes of input segments a single compaction task will pull for one interval. It is a per-task ceiling, not a target: an interval whose segments sum to more than this is skipped entirely rather than partially compacted, so setting it too low starves fragmented intervals. 419430400 (400 MiB) is a conservative example; production values are often much higher (tens of GB) so no reasonable interval is ever skipped. The historical default was effectively unbounded, so most misconfigurations come from setting it too low.
• skipOffsetFromLatest — an ISO-8601 duration measured back from the latest segment; intervals inside it are never auto-compacted. This is the guard that keeps compaction off still-mutating recent data. Size it as at least the segment granularity plus the worst-case ingestion watermark lag plus margin — P1D is the common safe default for DAY segments fed by a stream.
• partitionsSpec.maxRowsPerSegment (or targetRowsPerSegment for range/hashed) — the output shape. maxRowsPerSegment is a hard ceiling for dynamic partitioning; targetRowsPerSegment is a soft target for clustered partitioning. Either way, this is what lands output in the size band, since the byte-based targetCompactionSizeBytes was removed in Druid 0.21 and size is now controlled through row count.
• granularitySpec.segmentGranularity — coarsening this (e.g. ingest HOUR, compact to DAY) is how many small hourly segments collapse into fewer daily ones; it also changes how much data each interval holds relative to inputSegmentSizeBytes.

The relationship among the three is what you actually tune. The skip window sets the boundary between "leave alone" and "eligible"; inputSegmentSizeBytes sets whether an eligible interval fits in one task; and the row target sets the output size. Convert a desired output byte size to a row target with the measured bytes-per-row:

$$ \text{targetRowsPerSegment} \approx \frac{\text{targetBytes}}{\text{avgRowBytes}} $$

For a 700 MB target on data measuring 140 bytes/row compressed, $\text{targetRowsPerSegment} \approx \frac{700 \times 1048576}{140} \approx 5.24 \times 10^{6}$ — round to 5000000. Set inputSegmentSizeBytes to comfortably exceed the total compacted bytes of your largest single interval so no interval is ever skipped as too large; the sizing method for the output band is developed in full under segment size optimization strategies.

Python Automation Script #

Because input tuning is derived from measured data, the pipeline should compute the row target from live segment stats, then apply the config idempotently and poll for convergence. The orchestrator below reads real bytes-per-row, derives maxRowsPerSegment, POSTs the config, and waits for the backlog to drain — refusing to apply a skipOffsetFromLatest smaller than a supplied minimum so a too-aggressive offset cannot slip through. It uses the standard library plus requests, with capped exponential backoff.

import time
import logging
import requests

logger = logging.getLogger("druid.compact_input")

COORDINATOR = "http://druid-coordinator:8081"
MIN_SKIP_ISO = "P1D"  # never accept an offset smaller than this


def _request(method, url, *, retries=5, **kwargs):
    """HTTP with capped exponential backoff on 5xx and connection errors."""
    delay = 1.0
    for attempt in range(1, retries + 1):
        try:
            resp = requests.request(method, url, timeout=30, **kwargs)
            if resp.status_code < 500:
                resp.raise_for_status()
                return resp
        except requests.RequestException as exc:
            logger.warning("%s %s failed: %s (attempt %d)", method, url, exc, attempt)
        if attempt == retries:
            raise RuntimeError(f"{method} {url} failed after {retries} attempts")
        time.sleep(delay)
        delay = min(delay * 2, 30.0)  # cap the backoff at 30s
    return None


def measured_row_bytes(datasource: str) -> float:
    """Average compressed bytes-per-row from live segments; the term that decides output size."""
    segs = _request(
        "GET",
        f"{COORDINATOR}/druid/coordinator/v1/datasources/{datasource}/segments?full",
    ).json()
    total_bytes = sum(int(s["size"]) for s in segs)
    total_rows = sum(int(s["num_rows"]) for s in segs)
    if total_rows == 0:
        raise RuntimeError("no rows found; cannot derive row target")
    return total_bytes / total_rows


def apply_input_tuning(datasource: str, target_mb: int, skip: str,
                       input_ceiling_bytes: int) -> None:
    if skip < MIN_SKIP_ISO:  # crude lexical guard; enforce a real minimum in your validator
        raise ValueError(f"skipOffsetFromLatest {skip} below minimum {MIN_SKIP_ISO}")
    avg = measured_row_bytes(datasource)
    max_rows = max(1_000_000, int(target_mb * 1_048_576 / avg))
    logger.info("avg_row_bytes=%.1f -> maxRowsPerSegment=%d", avg, max_rows)
    config = {
        "dataSource": datasource,
        "taskPriority": 25,
        "inputSegmentSizeBytes": input_ceiling_bytes,
        "skipOffsetFromLatest": skip,
        "tuningConfig": {
            "partitionsSpec": {"type": "dynamic", "maxRowsPerSegment": max_rows},
            "maxNumConcurrentSubTasks": 4,
        },
        "granularitySpec": {"segmentGranularity": "DAY", "rollup": True},
    }
    _request(
        "POST",
        f"{COORDINATOR}/druid/coordinator/v1/config/compaction",
        json=config, headers={"Content-Type": "application/json"},
    )


def await_drain(datasource: str, deadline_s: int = 3600, poll_s: int = 60) -> bool:
    started = time.monotonic()
    while time.monotonic() - started < deadline_s:
        status = _request(
            "GET", f"{COORDINATOR}/druid/coordinator/v1/compaction/status"
        ).json().get("latestStatus", [])
        row = next((r for r in status if r["dataSource"] == datasource), None)
        waiting = int(row.get("bytesAwaitingCompaction", 0)) if row else 0
        logger.info("%s bytesAwaitingCompaction=%d", datasource, waiting)
        if waiting == 0:
            return True
        time.sleep(poll_s)
    return False


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    apply_input_tuning(
        "analytics_events",
        target_mb=700,
        skip="P1D",
        input_ceiling_bytes=50 * 1024 ** 3,  # 50 GB — comfortably above any single interval
    )
    print("drained" if await_drain("analytics_events") else "still draining at deadline")

Deriving maxRowsPerSegment from measured bytes-per-row on every run keeps output in the band as the data's row width drifts, and the skipOffsetFromLatest guard makes it impossible to ship a config that would fight ingestion.

Verification Steps #

Confirm the tuning both consolidated the tail and left recent data alone. First, the fragment count on older intervals should collapse while the newest (protected) intervals keep their fragments:

curl -s "http://druid-coordinator:8081/druid/coordinator/v1/datasources/analytics_events/intervals?full" \
  | jq 'to_entries | map({interval: .key, count: .value.count}) | sort_by(.interval) | .[0:6]'

Expected: aged intervals now show 1–3 segments each, while the last day (inside skipOffsetFromLatest) still shows many — proof compaction consolidated the tail without disturbing live ingestion:

[
  { "interval": "2026-03-10T00:00:00.000Z/2026-03-11T00:00:00.000Z", "count": 3 },
  { "interval": "2026-03-11T00:00:00.000Z/2026-03-12T00:00:00.000Z", "count": 2 },
  { "interval": "2026-03-18T00:00:00.000Z/2026-03-19T00:00:00.000Z", "count": 47 }
]

Next, confirm the backlog reached zero and nothing was skipped as too large:

curl -s "http://druid-coordinator:8081/druid/coordinator/v1/compaction/status" \
  | jq '.latestStatus[] | select(.dataSource=="analytics_events")
        | {bytesAwaitingCompaction, bytesSkipped}'

Expected — a zero backlog and zero skipped bytes means inputSegmentSizeBytes was high enough for every interval to fit:

{ "bytesAwaitingCompaction": 0, "bytesSkipped": 0 }

Finally, audit the output size distribution so the row target actually landed segments in the band:

curl -s "http://druid-coordinator:8081/druid/coordinator/v1/datasources/analytics_events/segments?full" \
  | jq '[.[] | (.size / 1048576 | floor)] | {min: min, max: max, count: length}'

Enforce these as gates: alert if bytesSkipped is ever non-zero (an interval too large for inputSegmentSizeBytes), alert on TaskLock errors in compact logs (skip window too small), and re-derive maxRowsPerSegment from live bytes-per-row rather than hard-coding it. The broader interplay of these thresholds with task-slot budget and parallelism is covered under compaction threshold tuning, and the scheduling of when these tasks fire under automated compaction task scheduling.

Related #

• Compaction Threshold Tuning — the parent guide on the row, size, and concurrency thresholds that decide how aggressively compaction fires.
• Automated Compaction Task Scheduling — how the Coordinator duty decides when to run the tasks these input knobs shape.
• Segment Size Optimization Strategies — the method for choosing the output row target that lands segments in the 500 MB–1 GB band.
• Optimizing Segment Size for Historical Nodes — why oversized and undersized compaction output both hurt Historical query performance.
• Configuring Druid Native Compaction Rules — the full compact-task grammar and locking semantics behind the input selection.

Up one level: Compaction Threshold Tuning.

For authoritative compaction and input-spec semantics, see the official Apache Druid compaction reference.