Druid Segment Ops Segment & Ingestion Orchestration

Building an Airflow DAG for Druid Ingestion

The recurring question when wiring Apache Airflow to Druid is not whether an operator can POST a spec — it obviously can — but how to structure the DAG so that a long-running index_parallel task does not pin a worker slot for an hour, and so that everything downstream waits for the data to be queryable rather than merely for the task to report SUCCESS. This page gives a complete, realistic DAG that submits a Druid batch task with a PythonOperator, passes the task id through XCom, and waits on a custom sensor that polls first the Overlord and then the Coordinator. It is the concrete build behind the reusable helper described in the parent guide on integrating Druid ingestion with Airflow & Dagster. Everything inside the operators is the Python standard library plus requests, so it runs on a stock Airflow worker with no Druid provider package required.

Airflow DAG task graph for a Druid ingestion with a submit operator and a handoff sensor Four Airflow tasks run left to right. A wait_for_source sensor confirms the upstream S3 prefix exists. A submit_druid_task PythonOperator POSTs the index_parallel spec to the Overlord and pushes the returned task id to XCom. A wait_for_handoff sensor pulls the task id from XCom and polls the Overlord status endpoint until SUCCESS, then polls the Coordinator until the segments load. Only then does the downstream refresh_dashboards task run. A dashed edge shows XCom passing the task id from the submit operator to the sensor. wait_for_source S3 prefix sensor submit_druid_task PythonOperator wait_for_handoff custom sensor refresh_dashboards downstream XCom: task_id

Failure Modes & Diagnostics

Before the DAG, understand the two structural mistakes it exists to prevent, because both are silent. The first is blocking a worker slot on a long poll. A naive DAG submits and then loops on the task status inside the same PythonOperator, so an hour-long batch holds an Airflow worker slot for an hour. Under a backfill of ninety days that exhausts the worker pool and the scheduler stalls with everything queued. Confirm the pathology by watching how many task instances sit in running versus how many slots exist:

airflow jobs check --job-type SchedulerJob   # scheduler alive?
airflow tasks states-for-dag-run druid_web_ingest 2026-07-16 | grep running

The fix is the split shown here — a fast submit operator plus a separate sensor — and, in production, a deferrable sensor so the wait parks on the triggerer instead of a worker.

The second mistake is completing on SUCCESS instead of handoff. If wait_for_handoff only polled the Overlord, the downstream dashboard refresh would query segments that are published but not yet loaded onto a Historical, returning empty or partial results. Reproduce the gap by checking the Coordinator immediately after a task finishes:

# Right after the task reports SUCCESS, is the interval actually loaded?
curl -s "http://coordinator:8081/druid/coordinator/v1/loadstatus?full" | jq '.events_web'

A non-empty entry means segments are still queued for load. The sensor below closes that gap by requiring both an Overlord terminal SUCCESS and a Coordinator confirmation that the interval's segments are present with non-zero size. A third, subtler failure is a retry minting a new task id: if the submit operator called the Overlord without a pinned id, an Airflow retry would submit a second, differently-named task and double-ingest. The DAG derives the id deterministically so a retry resubmits the same id and the Overlord rejects the duplicate.

Target Spec & Validated JSON

The DAG loads one templated index_parallel spec and injects the run's interval and input prefix from the Airflow logical date. The spec below is complete and copy-ready — every required key (type, and within spec the dataSchema, ioConfig, tuningConfig) is present. dropExisting: true with the bounded intervals makes a re-run of the same logical date atomically replace its window.

{
  "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
      }
    }
  }
}

Python Automation Script

The DAG file below is realistic Airflow. It defines a helper module inline for clarity — in a real repo that would be an importable package — then wires four tasks: an S3 key sensor, a PythonOperator that submits and pushes the task id to XCom, a custom sensor subclassing BaseSensorOperator that gates on handoff, and a downstream placeholder. Retries and an SLA are set at the task level, and the submit uses a deterministic id so a retry is idempotent.

import hashlib
import time
from datetime import datetime, timedelta

import requests
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.sensors.base import BaseSensorOperator

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


def _task_id(datasource: str, interval: str) -> str:
    key = f"{datasource}:{interval}:{SPEC_VERSION}".encode()
    return f"idx_{datasource}_{hashlib.sha1(key).hexdigest()[:12]}"


def _build_spec(interval_start: str, interval_end: str) -> dict:
    day = interval_start[:10]
    return {
        "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": [f"{interval_start}/{interval_end}"]},
            },
            "ioConfig": {
                "type": "index_parallel",
                "inputSource": {"type": "s3",
                                "prefixes": [f"s3://lake/events/web/{day}/"]},
                "inputFormat": {"type": "json"},
                "appendToExisting": False, "dropExisting": True,
            },
            "tuningConfig": {
                "type": "index_parallel", "maxRowsInMemory": 1_000_000,
                "maxNumConcurrentSubTasks": 4, "forceGuaranteedRollup": True,
                "partitionsSpec": {"type": "range",
                                   "partitionDimensions": ["country"],
                                   "targetRowsPerSegment": 5_000_000},
            },
        },
    }


def submit_druid_task(data_interval_start, data_interval_end, ti, **_):
    """Submit with a pinned, deterministic id; push the id to XCom."""
    start = data_interval_start.strftime("%Y-%m-%dT%H:%M:%SZ")
    end = data_interval_end.strftime("%Y-%m-%dT%H:%M:%SZ")
    task_id = _task_id("events_web", start)
    spec = {**_build_spec(start, end), "id": task_id}
    delay = 2.0
    for attempt in range(5):
        r = requests.post(f"{OVERLORD}/druid/indexer/v1/task", json=spec, timeout=30)
        if r.status_code == 400 and task_id in r.text:
            break                      # already submitted: idempotent no-op
        if r.ok:
            break
        if attempt == 4:
            r.raise_for_status()
        time.sleep(delay)
        delay = min(delay * 2, 30.0)
    ti.xcom_push(key="druid_task_id", value=task_id)
    ti.xcom_push(key="druid_interval", value=start)


class DruidHandoffSensor(BaseSensorOperator):
    """Poke returns True only when the task is SUCCESS AND segments are loaded."""

    def __init__(self, datasource: str, **kwargs):
        super().__init__(**kwargs)
        self.datasource = datasource

    def poke(self, context) -> bool:
        ti = context["ti"]
        task_id = ti.xcom_pull(task_ids="submit_druid_task", key="druid_task_id")
        interval = ti.xcom_pull(task_ids="submit_druid_task", key="druid_interval")
        status = requests.get(
            f"{OVERLORD}/druid/indexer/v1/task/{task_id}/status", timeout=15)
        status.raise_for_status()
        state = status.json()["status"]["status"]
        if state == "FAILED":
            raise RuntimeError(f"druid task {task_id} FAILED")
        if state != "SUCCESS":
            return False               # still WAITING/PENDING/RUNNING; re-poke
        segs = requests.get(
            f"{COORDINATOR}/druid/coordinator/v1/datasources/{self.datasource}/segments",
            params={"full": "true"}, timeout=15)
        segs.raise_for_status()
        return any(s.get("interval", "").startswith(interval) and s.get("size", 0) > 0
                   for s in segs.json())


default_args = {
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "sla": timedelta(minutes=45),
}

with DAG(
    dag_id="druid_web_ingest",
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=True,
    max_active_runs=8,               # cap slot demand during backfills
    default_args=default_args,
    tags=["druid", "ingestion"],
) as dag:

    wait_for_source = S3KeySensor(
        task_id="wait_for_source",
        bucket_key="s3://lake/events/web//_SUCCESS",
        poke_interval=60, timeout=3600, mode="reschedule",
    )

    submit = PythonOperator(
        task_id="submit_druid_task",
        python_callable=submit_druid_task,
    )

    wait_for_handoff = DruidHandoffSensor(
        task_id="wait_for_handoff",
        datasource="events_web",
        poke_interval=30, timeout=2700, mode="reschedule",
    )

    refresh = PythonOperator(
        task_id="refresh_dashboards",
        python_callable=lambda **_: None,
    )

    wait_for_source >> submit >> wait_for_handoff >> refresh

Three details make this production-grade rather than a toy. mode="reschedule" on both sensors frees the worker slot between pokes, so a backfill does not exhaust the pool — the same win a deferrable sensor gives, available without the triggerer. The submit callable treats a 400 naming the existing id as success, so an Airflow retry after a partial failure never double-ingests, honoring the idempotent ingestion & retry patterns contract. And max_active_runs=8 caps how many intervals ingest at once during catchup, keeping Druid slot demand bounded.

Verification Steps

After a DAG run finishes, confirm the run behaved as designed: the sensor gated on handoff, exactly one version loaded, and the interval is queryable. First, confirm the task the DAG submitted reached SUCCESS:

TASK_ID=$(airflow tasks render druid_web_ingest submit_druid_task 2026-07-16 2>/dev/null; \
  curl -s "http://overlord:8090/druid/indexer/v1/tasks?state=complete&datasource=events_web" \
  | jq -r '.[0].id')
curl -s "http://overlord:8090/druid/indexer/v1/task/${TASK_ID}/status" \
  | jq '{id: .status.id, state: .status.status, duration: .status.duration}'

Expected output — a terminal success with a bounded duration:

{
  "id": "idx_events_web_9f3c2a1b7e40",
  "state": "SUCCESS",
  "duration": 512000
}

Next, confirm the sensor's handoff gate was truthful: the interval must have exactly one live version loaded with non-zero size.

curl -s "http://coordinator:8081/druid/coordinator/v1/datasources/events_web/segments?full=true" \
  | jq '[.[] | select(.interval | startswith("2026-07-16"))]
        | {segments: length, versions: (map(.version) | unique), mb: (map(.size) | add / 1048576 | floor)}'

Expected output — one version, a sane segment count and size band:

{
  "segments": 3,
  "versions": ["2026-07-16T04:11:07.482Z"],
  "mb": 1180
}

A single entry in versions proves the dropExisting replace superseded cleanly rather than leaving overlapping data, and a non-zero mb confirms the segments are real and loaded. Deeper handoff verification — including the loadstatus percentage and serverview checks — is covered in verifying segment handoff completion.

Up one level: Integrating Druid ingestion with Airflow & Dagster.

Back to Automated Ingestion Pipeline Orchestration