Druid Segment Ops Segment & Ingestion Orchestration

Dynamic vs Hash Partitioning Trade-offs

The decision forces itself the moment a datasource needs perfect rollup but ingestion was left on the default: rows that should have collapsed stay duplicated across shards, and aggregate counts drift. Apache Druid splits each time chunk into segments a second time using a partitionsSpec, and the two common choices pull in opposite directions — dynamic optimizes ingestion throughput and produces size-driven shards with only best-effort rollup, while hashed shuffles rows by a hash of their dimensions to guarantee even distribution and, with forceGuaranteedRollup, perfect rollup at the cost of a second processing phase. This page sits under understanding Druid segment granularity, which sets the time boundary; here we decide the secondary partitioning within each of those time chunks.

Dynamic single-pass partitioning versus hashed two-phase partitioning within one time chunk Two ingestion paths for the rows of a single time chunk. The dynamic path is one pass: rows stream straight into size-driven shards as maxRowsPerSegment fills, giving high throughput but only best-effort rollup, so duplicate dimension tuples can land in different shards. The hashed path is two phases: phase one hashes every row by its partition dimensions to determine target shards, phase two shuffles matching rows together so each shard holds a disjoint hash bucket, giving even shards and guaranteed perfect rollup when forceGuaranteedRollup is set, at the cost of the extra shuffle phase. rows of one time chunk dynamic · single pass fill by size: maxRowsPerSegment shard 0 shard 1 high throughput best-effort rollup only dup tuples may split across shards hashed · two phase phase 1: hash by dimensions phase 2: shuffle to hash buckets bucket 0 bucket 1 even shards guaranteed rollup extra shuffle phase = slower

Failure Modes & Diagnostics

Partitioning problems present as drifting aggregates, lopsided shards, or ingestion that runs far slower than expected. The direction tells you which spec is wrong for the job. Start by confirming what partitioning the published segments actually used and how evenly rows landed:

# Shard type and row spread per interval — even shards imply hashed, ragged imply dynamic
curl -s -X POST "http://<coordinator-host>:8081/druid/coordinator/v1/metadata/datasources/events_partitioned/segments?full" \
  -H 'Content-Type: application/json' \
  -d '["2026-01-01T00:00:00.000Z/2026-01-08T00:00:00.000Z"]' \
  | jq '.[] | {interval: .interval, shard: .shardSpec.type, partition: .shardSpec.partitionNum, rows: .["num_rows"], size_mb: (.size/1048576|floor)}'

Rollup not collapsing duplicate rows. Symptom: SELECT COUNT(*) exceeds the count of distinct dimension tuples that rollup should have merged. Confirm best-effort rollup let identical tuples split across shards:

# Raw rows vs distinct dimension tuples — a gap means rollup did not fully collapse
curl -s "http://<broker-host>:8082/druid/v2/sql" -H 'Content-Type: application/json' \
  -d '{"query":"SELECT COUNT(*) AS raw, COUNT(DISTINCT country || device) AS tuples FROM events_partitioned WHERE __time >= TIMESTAMP '\''2026-01-01'\''"}'

Root cause is dynamic partitioning (which only rolls up within a shard) on a datasource that needs perfect rollup. Remediation: switch to hashed with forceGuaranteedRollup: true, or re-compact the interval with that spec.

Skewed shard sizes / a hot shard. Symptom: one segment per interval is far larger and slower to scan. With dynamic partitioning a size-driven cut can co-locate heavy tuples; with hashed it usually means too few numShards for the cardinality. Inspect the spread with the first command above — a wide size_mb range confirms skew. Remediation: move to hashed and either set numShards explicitly or supply partitionDimensions that spread the load.

Ingestion far slower than a comparable job. Symptom: wall-clock time roughly doubles versus a size-only job. Check whether the task ran the extra shuffle phase:

curl -s "http://<overlord-host>:8090/druid/indexer/v1/task/<task_id>/reports" \
  | jq '.ingestionStatsAndErrors.payload.ingestionState, .ingestionStatsAndErrors.payload.rowStats'

Root cause is hashed with forceGuaranteedRollup paying for a two-phase shuffle you may not need. Remediation: if perfect rollup is not required, use dynamic for the throughput. The time boundary that feeds all of this is set by segment granularity; partitioning only subdivides within it.

Target Spec & Validated JSON

partitionsSpec lives in tuningConfig. Use dynamic when throughput and best-effort rollup are acceptable — it is a single pass and size-driven:

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "events_partitioned",
      "timestampSpec": { "column": "ts", "format": "iso" },
      "granularitySpec": { "type": "uniform", "segmentGranularity": "DAY", "rollup": true }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": { "type": "s3", "prefixes": ["s3://analytics-bucket/raw/"] },
      "inputFormat": { "type": "json" }
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxRowsInMemory": 1000000,
      "forceGuaranteedRollup": false,
      "partitionsSpec": {
        "type": "dynamic",
        "maxRowsPerSegment": 5000000,
        "maxTotalRows": 20000000
      }
    }
  }
}

When you need even shards and perfect rollup, switch to hashed. This requires forceGuaranteedRollup: true and, for batch, an explicit or profiled interval set so Druid can run the shuffle:

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "events_partitioned",
      "timestampSpec": { "column": "ts", "format": "iso" },
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "rollup": true,
        "intervals": ["2026-01-01/2026-01-08"]
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": { "type": "s3", "prefixes": ["s3://analytics-bucket/raw/"] },
      "inputFormat": { "type": "json" }
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxRowsInMemory": 1000000,
      "forceGuaranteedRollup": true,
      "maxNumConcurrentSubTasks": 4,
      "partitionsSpec": {
        "type": "hashed",
        "targetRowsPerSegment": 5000000,
        "partitionDimensions": ["country", "device"]
      }
    }
  }
}

The fields that decide the trade-off:

  • partitionsSpec.type — dynamic (single pass, size-driven, best-effort rollup) vs hashed (two-phase shuffle, even hash buckets, perfect rollup). A third option, range (formerly single_dim), sorts on chosen dimensions so that common range/IN filters prune whole segments — pick it when queries filter heavily on a low-cardinality dimension and you want both pruning and guaranteed rollup.
  • forceGuaranteedRollup — must be true for hashed/range to guarantee rollup; it is what triggers the second phase. Leave false for dynamic.
  • targetRowsPerSegment vs maxRowsPerSegment — hashed/range size by targetRowsPerSegment (or fixed numShards); dynamic caps by maxRowsPerSegment plus maxTotalRows.
  • partitionDimensions — for hashed, restricts the hash to these dimensions so identical tuples always co-locate; omit to hash all dimensions.

Because hashed needs the row set up front, it pairs naturally with a bounded interval; the interval arithmetic is covered under how Druid segments map to time intervals, and the encoding that rollup then shrinks is detailed in columnar storage formats in Druid.

Python Automation Script

The choice should be data-driven, not hardcoded. The helper below picks a partitionsSpec from whether perfect rollup is required, submits the task, applies capped exponential backoff, and polls to a terminal state. It uses only the standard library plus requests.

import time
import logging
import requests

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("druid_partition")


def build_partitions_spec(require_perfect_rollup, target_rows=5_000_000,
                          partition_dimensions=None):
    """Return a partitionsSpec and the forceGuaranteedRollup flag it needs."""
    if require_perfect_rollup:
        spec = {"type": "hashed", "targetRowsPerSegment": target_rows}
        if partition_dimensions:
            spec["partitionDimensions"] = partition_dimensions
        return spec, True
    return {
        "type": "dynamic",
        "maxRowsPerSegment": target_rows,
        "maxTotalRows": target_rows * 4,
    }, False


class DruidPartitionPipeline:
    def __init__(self, overlord_url, auth=None):
        self.overlord = overlord_url.rstrip("/")
        self.session = requests.Session()
        self.session.auth = auth

    def submit(self, spec):
        resp = self.session.post(
            f"{self.overlord}/druid/indexer/v1/task", json=spec, timeout=30
        )
        resp.raise_for_status()
        task_id = resp.json()["task"]
        logger.info("Submitted ingestion task %s", task_id)
        return task_id

    def poll_until_terminal(self, task_id, base_delay=5, max_delay=60, max_wait=7200):
        start = time.time()
        delay = base_delay
        while time.time() - start < max_wait:
            resp = self.session.get(
                f"{self.overlord}/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 terminal state: %s", task_id, status)
                return status
            logger.debug("Task %s pending (%s); sleeping %ss", task_id, status, delay)
            time.sleep(delay)
            delay = min(delay * 2, max_delay)  # exponential backoff, capped
        raise TimeoutError(f"Task {task_id} exceeded {max_wait}s")


# Usage
# pipe = DruidPartitionPipeline("http://overlord:8090")
# pspec, force = build_partitions_spec(require_perfect_rollup=True,
#                                      partition_dimensions=["country", "device"])
# spec["spec"]["tuningConfig"]["partitionsSpec"] = pspec
# spec["spec"]["tuningConfig"]["forceGuaranteedRollup"] = force
# if pipe.poll_until_terminal(pipe.submit(spec)) != "SUCCESS":
#     raise SystemExit("Ingestion failed; partitioning not applied")

Because a non-SUCCESS state raises loudly, a mis-partitioned run never silently ships. For templating this across environments see dynamic ingestion spec generation, and for structural validation before submission see schema validation for Druid specs.

Verification Steps

Confirm rollup and shard evenness after the task completes. First verify that raw rows now equal distinct tuples — proof that hashed + forceGuaranteedRollup collapsed every duplicate:

curl -s "http://<broker-host>:8082/druid/v2/sql" -H 'Content-Type: application/json' \
  -d '{"query":"SELECT COUNT(*) AS raw_rows, COUNT(DISTINCT country || '\''|'\'' || device) AS tuples FROM events_partitioned WHERE __time >= TIMESTAMP '\''2026-01-01'\'' AND __time < TIMESTAMP '\''2026-01-08'\''"}'

Expected output shows the two counts matching (perfect rollup) rather than raw_rows exceeding tuples:

[
  {
    "raw_rows": 184320,
    "tuples": 184320
  }
]

Then confirm the shards came out even — hashed distribution should hold size variance tight:

curl -s -X POST "http://<coordinator-host>:8081/druid/coordinator/v1/metadata/datasources/events_partitioned/segments?full" \
  -H 'Content-Type: application/json' \
  -d '["2026-01-01T00:00:00.000Z/2026-01-08T00:00:00.000Z"]' \
  | jq '[.[] | (.size / 1048576 | floor)] | {shards: length, min_mb: min, max_mb: max}'

A tight min_mb/max_mb spread confirms even hash buckets; a wide spread means numShards is too low for the cardinality or a hot partitionDimensions value dominates. Reference the Druid partitioning documentation for hashed, range, and dynamic semantics and the two-phase shuffle requirements.

Back to Apache Druid Segment Architecture & Lifecycle Fundamentals