Druid Segment Ops Segment & Ingestion Orchestration

Optimizing Segment Size for Historical Nodes

Historical nodes serve as the primary query execution layer in Apache Druid, relying on deterministic segment caching to maintain low-latency OLAP performance. Segment sizing directly dictates memory allocation, garbage collection behavior, and Broker routing efficiency. When segment footprints drift outside the 500–750 MB operational sweet spot, resource contention compounds rapidly. Understanding the Apache Druid Segment Architecture & Lifecycle Fundamentals is prerequisite to implementing strict ingestion controls, automated compaction workflows, and CI/CD validation gates.

Failure Modes & Diagnostics

Oversized segments (>1.5 GB) trigger predictable JVM heap exhaustion during the load phase, causing prolonged stop-the-world garbage collection pauses and OutOfMemoryError: Java heap space exceptions during handoff. Conversely, undersized segments (<200 MB) fragment the underlying Columnar Storage Formats in Druid, inflating metadata overhead in the relational metadata store and degrading Broker-to-Historical discovery latency. Diagnostic workflows must isolate these conditions before they cascade into query timeouts or node evictions.

# Inspect segment size distribution per datasource via Coordinator API
curl -s "http://<coordinator-host>:8081/druid/coordinator/v1/datasources/<datasource>/segments" | \
  jq '.[] | {id: .id, size_mb: (.size / 1048576 | floor)}' | sort -k2 -n

# Monitor Historical JVM heap pressure and segment load latency
jstat -gc <historical_pid> 1000 5
curl -s "http://<historical-host>:8083/druid/v2/segments/loaded" | jq 'length'

Target Sizing & Validated JSON Specs

Deterministic partitioning requires calibrating targetRowsPerSegment and maxRowsPerSegment against empirical row sizes and compression ratios. The baseline calculation targetRows = (TargetMB * 1048576) / AvgRowBytes must be validated post-ingestion, as Parquet dictionary encoding and bitmap compression can yield 30–60% variance from raw estimates. For batch and stream ingestion, the tuningConfig must enforce strict boundaries to prevent runaway partitioning.

{
  "type": "index_parallel",
  "spec": {
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": { "type": "s3", "prefixes": ["s3://analytics-bucket/raw/"] },
      "inputFormat": { "type": "parquet" }
    },
    "tuningConfig": {
      "type": "index_parallel",
      "partitionsSpec": {
        "type": "dynamic",
        "maxRowsPerSegment": 5000000,
        "maxTotalRows": 20000000
      },
      "maxRowsInMemory": 1000000,
      "forceGuaranteedRollup": false,
      "logParseExceptions": true,
      "maxParseExceptions": 0,
      "maxSavedParseExceptions": 0
    }
  }
}

Compression behavior should be validated against official Apache Parquet documentation to ensure dictionary thresholds and encoding strategies align with Druid's columnar expectations. Post-ingestion, segment sizes must be audited; if compression yields deviate from baseline, adjust maxRowsPerSegment incrementally rather than modifying targetRowsPerSegment mid-pipeline.

Python API Orchestration

Pipeline builders must automate compaction to correct drift without manual intervention. The following production-ready Python orchestrator submits compaction tasks, implements exponential backoff for API rate limits, and polls for terminal states. It integrates directly with CI/CD validation gates to block deployments if historical segment sizes exceed defined thresholds.

import requests
import time
import logging
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DruidCompactionOrchestrator:
    def __init__(self, overlord_url: str, datasource: str, interval: str, auth: tuple = None):
        self.base_url = overlord_url.rstrip("/")
        self.datasource = datasource
        self.interval = interval
        self.session = requests.Session()
        self.session.auth = auth
        retry_strategy = Retry(total=3, backoff_factor=1.5, status_forcelist=[429, 500, 502, 503, 504])
        self.session.mount("http://", HTTPAdapter(max_retries=retry_strategy))
        self.session.mount("https://", HTTPAdapter(max_retries=retry_strategy))

    def submit_compaction(self) -> str:
        payload = {
            "type": "compact",
            "dataSource": self.datasource,
            "ioConfig": {"type": "compact", "inputSpec": {"type": "interval", "interval": self.interval}},
            "tuningConfig": {
                "type": "index_parallel",
                "partitionsSpec": {"type": "dynamic", "maxRowsPerSegment": 5000000, "maxTotalRows": 20000000},
                "maxRowsInMemory": 1000000
            }
        }
        resp = self.session.post(f"{self.base_url}/druid/indexer/v1/task", json=payload, timeout=30)
        resp.raise_for_status()
        task_id = resp.json()["task"]
        logger.info("Compaction task submitted: %s", task_id)
        return task_id

    def poll_until_terminal(self, task_id: str, poll_interval: int = 15, max_wait: int = 3600) -> str:
        start = time.time()
        while time.time() - start < max_wait:
            resp = self.session.get(f"{self.base_url}/druid/indexer/v1/task/{task_id}/status", timeout=10)
            resp.raise_for_status()
            status = resp.json().get("status", {}).get("status")
            if status in ("SUCCESS", "FAILED", "INTERRUPTED"):
                logger.info("Task %s reached terminal state: %s", task_id, status)
                return status
            logger.debug("Task %s current state: %s", task_id, status)
            time.sleep(poll_interval)
        raise TimeoutError(f"Compaction task {task_id} exceeded max wait time")

# Usage in pipeline
# orchestrator = DruidCompactionOrchestrator("http://overlord:8090", "events_raw", "2024-01-01/2024-01-02")
# task_id = orchestrator.submit_compaction()
# final_state = orchestrator.poll_until_terminal(task_id)

Recovery Patterns & Validation Gates

When historical nodes encounter handoff failures due to oversized segments, immediate recovery requires isolating the affected datasource and forcing a state transition. Use the Coordinator API to mark problematic segments as unused, drop them from the metadata store, and trigger a targeted re-ingestion or compaction pass.

# Force segment state transition for recovery
curl -X POST "http://<coordinator-host>:8081/druid/coordinator/v1/datasources/<datasource>/markUnused" \
  -H "Content-Type: application/json" \
  -d '{"interval": "2024-01-01/2024-01-02"}'

# Trigger immediate coordinator balance cycle
curl -X POST "http://<coordinator-host>:8081/druid/coordinator/v1/loadQueue?immediate=true"

Recovery pipelines should enforce strict validation gates:

  1. Pre-flight Sizing Check: Query druid/coordinator/v1/datasources/<ds>/segments and reject ingestion if >15% of segments exceed 1.2 GB.
  2. Heap Guardrails: Configure Historical druid.processing.buffer.sizeBytes and druid.server.http.maxQueuedBytes to prevent OOM during concurrent segment loads.
  3. Automated Rollback: If compaction fails or produces segments outside the 500–750 MB range, the orchestrator must halt downstream Broker routing updates and revert to the previous stable segment set using metadata snapshots.

Reference the official Druid compaction documentation for advanced configuration of maxRowsPerSegment overrides and coordinator balancing intervals. Consistent enforcement of these patterns ensures predictable query latency, stable JVM memory profiles, and resilient segment lifecycle management across production clusters.

Back to Apache Druid Segment Architecture & Lifecycle Fundamentals