Apache Airflow Druid Provider

4.5.2 · active · verified Thu Apr 16

The Apache Airflow Druid Provider enables Airflow to interact with Apache Druid, a high-performance, real-time analytics database. It includes hooks and operators for executing Druid queries and loading data. The current version is 4.5.2, and Airflow providers generally follow Airflow's release cadence, with updates for new features and bug fixes.

Common errors

```
ModuleNotFoundError: No module named 'airflow.contrib.operators.druid_operator'
```
cause Using an old import path from Airflow 1.x or before the provider was separated.

fix Install the provider (`pip install apache-airflow-providers-apache-druid`) and update your import statement to `from airflow.providers.apache.druid.operators.druid import DruidOperator`.
```
AirflowException: The Druid connection with conn_id 'my_druid_conn' is not found.
```
cause The specified `druid_conn_id` does not exist in your Airflow connections.

fix Create an Airflow connection with the matching `conn_id` (e.g., 'my_druid_conn') and 'Conn Type' set to 'Druid' via the Airflow UI (Admin -> Connections) or CLI.
```
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
```
cause The `json_query` parameter in `DruidOperator` was provided with an invalid JSON string or object.

fix Ensure the `json_query` argument is a valid Python dictionary representing the Druid native query structure, or a properly formatted JSON string. If using `sql_query`, do not provide `json_query`.

Warnings

breaking Migration from Airflow 1.x to 2.x requires installing providers separately and updating import paths.
Fix: Install `apache-airflow-providers-apache-druid` explicitly (`pip install ...`) and update all imports from `airflow.contrib.*` to `airflow.providers.apache.druid.*`.
gotcha DruidOperator supports both `sql_query` and `json_query`. Ensure you use only one and provide the correct query format.
Fix: Use either `sql_query='SELECT ...'` for SQL or `json_query={...}` for native Druid queries, but not both simultaneously. Validate your JSON query structure against Druid's API documentation.
gotcha The `druid_conn_id` must reference a correctly configured Airflow connection of type 'Druid'. Misconfigured connections lead to errors.
Fix: Verify your Airflow connection: `Admin -> Connections`. Ensure 'Conn Id' matches `druid_conn_id`, 'Conn Type' is 'Druid', and 'Host' and 'Port' are correct for your Druid router/broker.

Install

pip install apache-airflow-providers-apache-druid Install latest version

Imports

DruidHook

wrong:

from airflow.contrib.hooks.druid_hook import DruidHook

correct:

from airflow.providers.apache.druid.hooks.druid import DruidHook

Old import path from Airflow 1.x / contrib, not compatible with Airflow 2.x providers.

DruidOperator

wrong:

from airflow.contrib.operators.druid_operator import DruidOperator

correct:

from airflow.providers.apache.druid.operators.druid import DruidOperator

Old import path from Airflow 1.x / contrib, not compatible with Airflow 2.x providers.

DruidToS3Operator

from airflow.providers.apache.druid.operators.druid_to_s3 import DruidToS3Operator

Quickstart

This quickstart demonstrates how to create an Airflow DAG that uses the `DruidOperator` to execute both SQL and native JSON queries against a Druid cluster. It assumes a Druid connection named `druid_default` is configured in Airflow. The examples show querying a 'wikipedia' datasource.

import os
from datetime import datetime

from airflow.models.dag import DAG
from airflow.providers.apache.druid.operators.druid import DruidOperator

# Ensure you have an Airflow connection named 'druid_default'
# Or set one up: airflow connections add --conn-id druid_default --conn-type Druid --conn-host localhost --conn-port 8082
# For local testing, Druid usually runs on localhost:8082 by default
DRUID_CONN_ID = os.environ.get('AIRFLOW_DRUID_CONN_ID', 'druid_default')

with DAG(
    dag_id='druid_example_dag',
    start_date=datetime(2023, 1, 1),
    schedule=None,
    catchup=False,
    tags=['druid', 'example'],
) as dag:
    # Example of a simple Druid SQL query
    run_druid_sql_query = DruidOperator(
        task_id='run_druid_sql_query',
        druid_conn_id=DRUID_CONN_ID,
        sql_query="SELECT COUNT(*) FROM wikipedia WHERE __time >= CURRENT_TIMESTAMP - INTERVAL '1' DAY",
    )

    # Example of a more complex Druid JSON query
    # This assumes a 'wikipedia' datasource exists
    run_druid_json_query = DruidOperator(
        task_id='run_druid_json_query',
        druid_conn_id=DRUID_CONN_ID,
        json_query={
            "queryType": "timeseries",
            "dataSource": "wikipedia",
            "granularity": "day",
            "intervals": ["2016-06-01T00:00:00.000Z/2016-06-02T00:00:00.000Z"],
            "aggregations": [
                {"type": "count", "name": "total_events"}
            ]
        },
    )

    run_druid_sql_query >> run_druid_json_query

view raw JSON →