Integrating Druid Ingestion with Airflow & Dagster

Most production Druid data does not arrive because someone pushed a spec by hand; it arrives because a scheduler decided the upstream table was ready and fired a job. Driving Apache Druid ingestion from an external workflow orchestrator — Apache Airflow or Dagster — turns ingestion into a first-class node in a dependency graph, so a Druid load only runs after its source partition lands, and everything downstream of the load only runs after the segments are queryable. This guide is part of automated ingestion pipeline orchestration, and it focuses on the seam between a general-purpose orchestrator and Druid's own Overlord: how to submit a task over REST, how to poll it to a terminal state and then past handoff, how to model backfills as ordinary DAG runs, and how to keep the whole thing idempotent so a retried task never double-ingests.

Mechanics & Internals #

A workflow orchestrator and the Druid Overlord solve two different problems, and the integration lives in the gap between them. The orchestrator owns when a job runs, what it depends on, and what happens on failure — retries, alerting, SLAs, backfills over a date range. The Overlord owns the execution of a single ingestion task: it accepts a spec, assigns worker capacity, and reports a task state machine of WAITING → PENDING → RUNNING → SUCCESS | FAILED. Neither system knows about the other's internals, so the integration is deliberately thin: it speaks Druid's REST API from inside an orchestrator task, and it translates Druid's task state back into the orchestrator's own notion of success or failure.

The single most important thing this translation must get right is that a task reporting SUCCESS is not the same as the data being queryable. When an index_parallel task finishes, its segments are published in the metadata store, but they are not yet loaded onto a Historical. The Coordinator must reconcile the new metadata against load rules and instruct a Historical to pull each segment before the Broker will route a query to it — the moment called handoff. An orchestrator that marks its ingestion task green on Overlord SUCCESS alone will happily start a downstream dashboard-refresh or a validation query against data that is not yet available, and it will read stale or empty results. The correct integration therefore has three logical steps, not one: submit, poll the task to terminal, and then poll the Coordinator until handoff completes. The distinction between task completion and query availability is important enough to have its own guide in ingestion handoff validation; the orchestrator's job is to make that gate a hard edge in the DAG.

The three REST surfaces the integration touches are stable across the latest Druid release. Submission is POST /druid/indexer/v1/task with the JSON spec as the body; the response returns the accepted {"task": "<taskId>"}. Status is GET /druid/indexer/v1/task/{taskId}/status, whose payload carries status.status (the state) and status.statusCode. The full report — row counts, unparseable rows, error message — is GET /druid/indexer/v1/task/{taskId}/reports, which the orchestrator should attach to a failed task instance for triage. Handoff is confirmed Coordinator-side with GET /druid/coordinator/v1/loadstatus or the per-datasource GET /druid/coordinator/v1/datasources/{ds}/segments. Both Airflow and Dagster express these as ordinary Python running inside an operator or op, so the integration is portable: the same submit-and-poll helper works from an Airflow PythonOperator, an Airflow deferrable sensor, or a Dagster @op.

Two design decisions separate a reliable integration from a flaky one. The first is who owns the poll loop. In Airflow, blocking a worker slot for an hour while a long batch runs is wasteful, so the mature pattern splits submission and waiting into two tasks: a PythonOperator that submits and pushes the task id to XCom, and a sensor (ideally a deferrable one, which releases the worker slot to the triggerer between polls) that waits on that id. Dagster expresses the same split as a submit op that yields the task id and a poll op that consumes it, or a single op that submits then polls with the process parked on the run worker. The second decision is idempotency of the task id, covered in depth by idempotent ingestion retry patterns: if the orchestrator derives the Druid task id deterministically from the datasource, interval, and spec version, then an orchestrator-level retry resubmits the same id, and the Overlord rejects the duplicate rather than running the work twice. This is what makes an at-least-once orchestrator safe to point at Druid without producing duplicate rows.

Backfills fall out of this model naturally. Because each ingestion is parameterized by an interval, a backfill over a date range is simply a set of DAG runs — one logical date per interval — each submitting a spec bounded to its own window with dropExisting: true so the run atomically replaces whatever was previously in that interval. Airflow's data-interval model and catchup/--reset-dagruns backfill machinery, and Dagster's partitioned jobs with a DailyPartitionsDefinition, both map one partition to one Druid interval. The generated spec itself is the concern of dynamic ingestion spec generation; the orchestrator only supplies the interval and reuses one templated builder across every run.

Validated Configuration Spec #

The orchestrator submits an ordinary index_parallel spec, parameterized per run by its interval and input prefix. Below is the complete, copy-ready spec a per-interval DAG run would POST, with all required top-level keys (type, and within spec the dataSchema, ioConfig, and tuningConfig) present. The two fields the orchestrator injects per run are granularitySpec.intervals and ioConfig.inputSource.prefixes; dropExisting: true combined with the bounded interval is what makes a re-run of the same DAG run safe.

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "events_web",
      "timestampSpec": { "column": "event_ts", "format": "iso" },
      "dimensionsSpec": {
        "dimensions": ["country", "device", { "name": "page_id", "type": "long" }]
      },
      "metricsSpec": [
        { "type": "count", "name": "rows" },
        { "type": "longSum", "name": "bytes", "fieldName": "resp_bytes" }
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "HOUR",
        "queryGranularity": "MINUTE",
        "rollup": true,
        "intervals": ["2026-07-16T00:00:00Z/2026-07-17T00:00:00Z"]
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "prefixes": ["s3://lake/events/web/2026-07-16/"]
      },
      "inputFormat": { "type": "json" },
      "appendToExisting": false,
      "dropExisting": true
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxRowsInMemory": 1000000,
      "maxNumConcurrentSubTasks": 4,
      "intermediatePersistPeriod": "PT10M",
      "forceGuaranteedRollup": true,
      "partitionsSpec": {
        "type": "range",
        "partitionDimensions": ["country"],
        "targetRowsPerSegment": 5000000
      }
    }
  }
}

The orchestrator does not hand-edit this JSON per run. It loads one template, substitutes the run's interval and prefix, and — critically — derives a deterministic task id it passes alongside the spec via the top-level "id" field the Overlord honors on submission. Pinning the id in the spec (rather than letting the Overlord mint a random one) is what ties an orchestrator retry to Overlord-side deduplication. The forceGuaranteedRollup: true setting requires the non-dynamic range partitionsSpec and the bounded intervals shown here, both of which every DAG run supplies by construction.

Sizing Heuristics & Formulas #

The orchestrator introduces two sizing dimensions on top of Druid's own segment sizing: how many Druid tasks it may have in flight at once, and how long a sensor should wait before declaring an SLA breach. Start with worker capacity. A MiddleManager runs at most druid.worker.capacity task slots, and a parallel task with maxNumConcurrentSubTasks set to ( k ) consumes up to ( k + 1 ) slots (one supervisor sub-task plus its workers). If the orchestrator fires ( N ) DAG runs concurrently during a backfill, the slot demand is:

$$ \text{slotsNeeded} \approx N \times (k + 1) $$

Exceeding total cluster slots queues tasks in PENDING, so the orchestrator should cap DAG-level concurrency — Airflow's max_active_runs or a pool, Dagster's run-queue tag concurrency — so that ( \text{slotsNeeded} \lesssim \text{totalSlots} ). For a backfill of 90 daily intervals with ( k = 4 ) on a Druid cluster of 40 slots, capping concurrency at 8 keeps demand at ( 8 \times 5 = 40 ) slots, fully utilized but never queued.

The sensor timeout is the second calibration. A poll loop with an initial delay ( d_0 ) that doubles to a ceiling ( d_{max} ) reaches its cap after ( \log_2(d_{max} / d_0) ) polls, after which each poll costs ( d_{max} ). To bound total wait at the SLA horizon ( T ), the number of polls at the ceiling is:

$$ \text{pollsAtCeiling} \approx \frac{T - \sum_{i} d_0 \times 2^{i}}{d_{max}} $$

In practice you set the sensor's overall timeout to a small multiple of the observed p95 task runtime — if a daily batch normally finishes in 12 minutes, a 45-minute sensor timeout absorbs a slow run without pinning the DAG open indefinitely. Finally, the deferrable-sensor choice has a capacity implication of its own: a classic sensor holds a worker slot for the whole wait, so ( W ) concurrent waits consume ( W ) worker slots, whereas a deferrable sensor parks on the triggerer and consumes roughly zero worker slots while waiting — the difference between a scheduler that stalls under many concurrent backfills and one that does not.

Python Orchestration Snippet #

The reusable core is a submit-and-poll helper that any operator or op can call. It uses only the Python standard library plus requests, derives an idempotent task id, submits with capped exponential backoff, polls the Overlord to a terminal state, and then confirms handoff against the Coordinator. A concrete end-to-end DAG built on this helper is walked through in an Airflow DAG for Druid ingestion.

import time
import hashlib
import requests

OVERLORD = "http://overlord:8090"
COORDINATOR = "http://coordinator:8081"
TERMINAL = {"SUCCESS", "FAILED"}


def deterministic_task_id(datasource: str, interval: str, spec_version: str) -> str:
    """Content-hashed id so an orchestrator retry resubmits identical work."""
    key = f"{datasource}:{interval}:{spec_version}".encode()
    return f"idx_{datasource}_{hashlib.sha1(key).hexdigest()[:12]}"


def submit_task(spec: dict, task_id: str, retries: int = 5) -> str:
    """POST the spec with a pinned id; retry transient errors with backoff."""
    spec = {**spec, "id": task_id}
    delay = 2.0
    for attempt in range(retries):
        try:
            r = requests.post(f"{OVERLORD}/druid/indexer/v1/task",
                              json=spec, timeout=30)
            # 400 with an existing-id message means the work already ran.
            if r.status_code == 400 and task_id in r.text:
                return task_id
            r.raise_for_status()
            return r.json()["task"]
        except requests.RequestException:
            if attempt == retries - 1:
                raise
            time.sleep(delay)
            delay = min(delay * 2, 30.0)  # cap backoff at 30s
    raise RuntimeError("unreachable")


def poll_until_terminal(task_id: str, timeout_s: int = 3600) -> str:
    """Poll task status to a terminal state with capped exponential backoff."""
    deadline = time.monotonic() + timeout_s
    delay = 5.0
    while time.monotonic() < deadline:
        r = requests.get(f"{OVERLORD}/druid/indexer/v1/task/{task_id}/status",
                         timeout=15)
        r.raise_for_status()
        state = r.json()["status"]["status"]
        if state in TERMINAL:
            return state
        time.sleep(delay)
        delay = min(delay * 2, 30.0)
    raise TimeoutError(f"{task_id} not terminal within {timeout_s}s")


def confirm_handoff(datasource: str, interval: str, tries: int = 30) -> bool:
    """SUCCESS != queryable: wait until the Coordinator loads the segments."""
    start = interval.split("/")[0]
    delay = 5.0
    for _ in range(tries):
        r = requests.get(
            f"{COORDINATOR}/druid/coordinator/v1/datasources/{datasource}/segments",
            params={"full": "true"}, timeout=15)
        r.raise_for_status()
        loaded = [s for s in r.json()
                  if s.get("interval", "").startswith(start) and s.get("size", 0) > 0]
        if loaded:
            return True
        time.sleep(delay)
        delay = min(delay * 2, 30.0)
    return False


def run_druid_ingestion(spec: dict, datasource: str, interval: str,
                        spec_version: str) -> None:
    """The whole contract an operator/op wraps: submit, wait, gate on handoff."""
    task_id = deterministic_task_id(datasource, interval, spec_version)
    submit_task(spec, task_id)
    state = poll_until_terminal(task_id)
    if state != "SUCCESS":
        raise RuntimeError(f"ingestion {task_id} ended {state}")
    if not confirm_handoff(datasource, interval):
        raise RuntimeError(f"{task_id} SUCCESS but segments never loaded")

Three properties make this safe to embed in any orchestrator. Idempotent submission: the pinned, content-hashed id means an Airflow task retry or a Dagster op re-execution resubmits the same id, and submit_task treats the Overlord's duplicate-rejection as success rather than an error. Capped backoff: both the submit retry and the poll loop start small and ceil at 30 seconds, so a leader election or a transient partition never hammers the Overlord. Handoff as a hard gate: run_druid_ingestion raises unless the Coordinator confirms loaded segments, so the orchestrator marks the task failed — and blocks everything downstream — when a task "succeeds" but never becomes queryable. Wiring the same three checks that async task execution patterns apply to bare submission into an operator is the whole of the integration.

Failure Modes & Diagnostics #

Orchestrator-driven ingestion fails in a handful of recognizable ways, most of them at the seam between the two systems rather than inside Druid itself.

1. The DAG marks the task green but downstream reads stale data. The operator polled Overlord SUCCESS and stopped. Confirm the gap by checking whether the segments actually loaded after the task finished:

curl -s "http://coordinator:8081/druid/coordinator/v1/loadstatus?full" \
  | jq '.events_web // "datasource fully loaded"'

A non-empty per-datasource entry means segments are still pending load — add the handoff gate. Full triage is in verifying segment handoff completion.

2. A retried DAG run produced duplicate rows. The operator minted a fresh random task id on retry, so Druid ran the work twice with appendToExisting. Inspect the live versions for the interval:

curl -s "http://coordinator:8081/druid/coordinator/v1/datasources/events_web/segments?full=true" \
  | jq 'group_by(.interval)[] | {interval: .[0].interval, versions: (map(.version) | unique)}'

More than one live version for one interval means the replace did not supersede. Pin a deterministic id and use dropExisting: true, as detailed in exactly-once Druid backfills.

3. Backfill tasks pile up in PENDING and the DAG stalls. The orchestrator fired more concurrent runs than the Druid cluster has slots. Read current capacity and cap DAG concurrency:

curl -s http://overlord:8090/druid/indexer/v1/workers \
  | jq '[.[] | {host: .worker.host, used: .currCapacityUsed, cap: .worker.capacity}]'

4. The sensor times out even though the task is healthy. The sensor timeout is shorter than the task's real runtime under load, or a classic sensor exhausted worker slots so the task never got scheduled. Confirm the task is actually running and pull its report:

curl -s http://overlord:8090/druid/indexer/v1/task/$TASK_ID/reports \
  | jq '.ingestionStatsAndErrors.payload.rowStats, .ingestionStatsAndErrors.payload.errorMsg'

Switch to a deferrable sensor and raise timeout to a multiple of the p95 runtime.

5. The submit operator retries forever on a 4xx. A genuinely malformed spec returns 400, which is not transient — retrying wastes the SLA. Distinguish an existing-id 400 (safe, already ran) from a validation 400 (fatal) in the operator, and validate the spec pre-flight under schema validation for Druid specs so malformed specs never reach the Overlord.

Automation Checklist #

Wire these gates into every orchestrator that drives Druid, whether Airflow or Dagster.

• Deterministic task id — derive the Druid task id from a content hash of datasource, interval, and spec version, and pin it in the submitted spec's id field.
• Treat duplicate-rejection as success — an Overlord 400 naming an existing id means the work already ran; the operator must not fail on it.
• Split submit from wait — submit in one task and poll in a separate sensor so a long batch does not pin a worker slot; prefer a deferrable sensor.
• Gate on handoff, not SUCCESS — the operator fails unless the Coordinator confirms the interval's segments are loaded and non-zero size.
• Bound the spec to its interval — every per-run spec sets an explicit granularitySpec.intervals and dropExisting: true so a re-run atomically replaces its window.
• Cap DAG concurrency to slot capacity — size max_active_runs / run-queue concurrency so N × (k+1) never exceeds cluster task slots during backfills.
• Set a sensor timeout from p95 runtime — bound the wait at a small multiple of observed runtime and alert on breach rather than waiting indefinitely.
• Attach the failure report — on a failed task, pull …/task/{id}/reports into the task instance log so triage does not require a second round-trip.
• Model backfills as partitioned runs — one interval per DAG run / partition, reusing one templated spec builder rather than bespoke jobs.
• Validate specs pre-flight — reject malformed specs in the orchestrator before submission so a 400 never burns retry budget.

Related #

• An Airflow DAG for Druid ingestion — a concrete DAG with a submit operator and a handoff sensor built on the helper above.
• Idempotent ingestion & retry patterns — content-hashed ids, Overlord deduplication, and safe re-runs that keep an at-least-once orchestrator from double-ingesting.
• Ingestion handoff validation — why SUCCESS is not queryable and how to gate a DAG on actual segment availability.
• Dynamic ingestion spec generation — the templated builder the orchestrator parameterizes per interval.
• Async task execution patterns — the submit-poll-backoff loop that every operator wraps.

Up one level: Automated ingestion pipeline orchestration frames how external orchestrators, spec generation, and handoff validation compose into one deterministic pipeline.