DuckDB provider

Konstantin Alpashkin

Overview

The DuckDB provider enables Airflow to execute SQL queries against DuckDB databases directly from DAGs. This provider does not include the DuckDB runtime. Instead, it requires the DuckDB CLI component to be installed on every worker node. Under the hood, the provider runs the CLI, submits an SQL query with a set of parameters to the CLI, receives the query results, and forwards them to Airflow.

Requirements

To use the provider, the following are required:

  • Airflow 3.2.1 or later;

  • the DuckDB CLI component installed on every Airflow worker host;

  • a preconfigured Airflow connection to access a DuckDB database.

Architecture

The provider follows the standard Airflow provider architecture and includes the following modules:

  • hooks. Hooks provide an interface for connecting and interacting with DuckDB.

  • operators. Operators execute DuckDB queries as Airflow tasks.

  • sensors. Sensors wait for a specified condition in DuckDB before triggering downstream processing.

  • utils. The module contains utility functions for working with SQL, parsing results, reporting errors, and so on.

DuckDbHook

The provider exposes only one hook — the DuckDbHook class, which encapsulates most of the provider’s runtime logic. Operators and sensors delegate connection handling, preflight checks, CLI execution, logging, and error reporting to DuckDbHook.

On each run, the hook performs the following:

  1. Reads the Airflow connection object and retrieves the database file path and values from the Extra field.

  2. Verifies cli_params against the banlist and checks that the CLI binary is available.

  3. Сhecks the local database path.

  4. Submits an SQL statement to DuckDB CLI.

  5. Forwards query results to Airflow.

  6. Writes stdout/stderr to the task logs and raises errors, if any.

The main DuckDbHook methods are:

  • run_cli().

    Executes an SQL statement via the DuckDB CLI. After optionally performing %(name)s substitution, the hook writes SQL statement to a temporary file and runs it with the -f flag to allow large SQL submissions. Returns raw stripped stdout as a string. The default output format is JSON.

  • run_file().

    Executes an existing .sql file with the -f flag. Executes the file as-is, without any parameter substitution. Does not specify the default output format. Therefore, output_format="[json|csv]" should be passed explicitly. Raises a configuration error if the specified .sql file does not exist.

  • test_connection()

    Verifies that the DuckDB CLI binary is available and validates the provided connection Extra fields (including CLI parameters banlist and secrets). Runs a test query on an in-memory database. Does not verify whether the target .duckdb file exists and is writable.

  • on_kill()

    Marks the hook as killed and terminates the active DuckDB process group. Used by operators and sensors when the task is killed.

DuckDbOperator

The provider’s only operator DuckDbOperator runs SQL via DuckDB CLI on a worker node and returns raw CLI stdout, which is then sent to XCom.

The main DuckDbOperator methods are:

  • execute()

    Runs an SQL statement and returns a raw stdout string from DuckDB CLI. After that, Airflow pushes that string to XCom. Does not return parsed table rows.

    With the default output_format="json", use airflow.providers.arenadata.duckdb.utils.json_output.parse_json_output — it handles optional non-JSON prefixes and requires a JSON list of rows. For example:

    from airflow.providers.arenadata.duckdb.utils.json_output import parse_json_output

raw = context["ti"].xcom_pull(task_ids="select_rows")
rows = parse_json_output(raw)

    Use one statement per task since _salvage_json() keeps the first JSON array and discards trailing data with a warning. The standard json.loads() function is safe only when the output contains a valid JSON list without additional content.

  • on_kill()

    Terminates the active DuckDB process if execute() has started the hook.

DuckDbSqlSensor

The DuckDbSqlSensor sensor is used to wait until a DuckDB query returns a truthy result. The sensor executes SQL using a DuckDbHook instance. The sensor evaluates the first cell of the first returned row and keeps waiting if:

  • the result set is empty, including empty CLI stdout;

  • the first row is missing, is not a mapping, or has no columns;

  • the first cell is Python-falsy (false, 0, null, or empty string).

NOTE
Non-empty strings such as "0" are considered truthy.

The main sensor method is poke(). On each method invocation, it runs an SQL statement with output_format="json" using the hook. Then, it parses the JSON result set and evaluates the first cell of the first row using the Python truthy check (bool(value)), and returns either False (keeps waiting) or True (succeeds).

CLI and JSON errors fail the task by default. Non-zero DuckDB CLI exit codes and invalid JSON raise exceptions. However, there are sensor parameters(defined in BaseSensorOperator) that can change this behavior:

  • silent_fail=True — exceptions in poke() do not fail the task. An exception is logged, and the sensor waits for the next poke();

  • never_fail=True — exceptions in poke() skip the task;

  • soft_fail=True — CLI/JSON errors do not affect waiting/skipping.

NOTE
By default, a query against a missing table (for example, Catalog Error: Table …​ does not exist) fails the task immediately, and the sensor does not continue waiting. If you need the sensor to wait until data appears, create the table in advance (it can be empty) and poke a condition such as SELECT COUNT(*) > 0 …​.

The sensor parses only the first JSON array in stdout. Later statements are discarded with a warning in logs. Thus, no statement before the condition should return rows. Keep in mind that:

  • INSTALL, LOAD, ATTACH, SET print nothing and may precede the condition.

  • CREATE SECRET returns a success row, which will be treated as the first result set. As a result, the sensor succeeds without evaluating the actual condition, and the task is marked as successful with a warning in the logs.

  • Each task is a separate CLI process. Operations like LOAD/ATTACH must be present in the sensor’s SQL, whereas INSTALL, CREATE PERSISTENT SECRET, and CREATE VIEW can run once in a preceding DuckDbOperator.

When fail_on_empty=True and the query returns no rows, the sensor raises an AirflowFailException. With soft_fail=True, the failure causes the poke to be skipped. When fail_on_empty=False (default), an empty result makes the sensor keep waiting.

Aggregates such as SELECT count(*) …​ always return one row so they are not an empty result set. For such queries, check the truthiness of the first cell, and do not rely on fail_on_empty.

Each poke() invocation creates a hook with lock_retry_attempts=0. Waiting for a locked .duckdb file (or for data to become available) is implemented via Airflow poke_interval and mode="reschedule", not via in-poke CLI lock retries.

Configure a connection

To connect to a DuckDB database, the DuckDB provider needs a preconfigured Airflow connection object. To create one, use Airflow UI. When creating a connection, specify the following parameters.

Field Description

Connection type

The type of connection used by the provider. Must be set to DuckDB

Database file path

Absolute path to a .duckdb file on a worker node or :memory: for in-memory tables

Additional fields

Field Description

DuckDB binary

Path to the DuckDB CLI binary used to execute SQL. The default path (/usr/bin/duckdb) points to the executable installed by the DuckDB CLI component of the DuckDB service

Additional JSON parameters

You can specify additional runtime settings in JSON format in the Extra field.

Parameter Description Default value

timeout

Timeout in seconds for the DuckDB CLI subprocess to complete

300

readonly

Opens the specified database in read-only mode. Performs basic checks that the database file exists and is readable

false

cli_params

Additional DuckDB CLI parameters. The value should be a shell string ("--threads 4") or a JSON array of strings (["--threads", "4"]). Non-string array elements raise a configuration error

 — 

lock_retry_attempts

Defines the number of retries to access the database file in case it is locked

0
Example Extra JSON
{
  "timeout": 300,
  "readonly": false,
  "cli_params": "",
  "lock_retry_attempts": 0
}

Locks and retries

DuckDB takes an exclusive lock on the database file when the CLI opens it. If another process already holds the lock, the CLI fails with an error indicating duckdb_conn_id and db_path.

The lock_retry_attempts property

To manage the locks more flexibly, the lock_retry_attempts connection property (specified in the Extra field) is used.

Its value should be a positive integer that indicates the number of retries in case the database file is locked. If set to 0, the task fails immediately with a clear lock-related message.

The backoff between attempts is 1, 2, 4, 8, 16 seconds (maximum is 16 seconds).

Total task execution time

Each attempt uses its own timeout. With retries enabled, the total task execution time can grow up to . Therefore, set the task’s execution_timeout accordingly.

Locks with ATTACH

The lock-retry mechanism assumes that the CLI opens a connection to a single host database. SQL scripts that include ATTACH 'file.db' operations mid-run can hit the same lock markers after some SQL statements have already been executed. For SQL with ATTACH commands, use lock_retry_attempts=0 (default value).

Logging

The following table describes logging levels for common log events.

Logging information Level

Full CLI command

DEBUG

Run summary (duckdb_conn_id, db_path, return code, duration)

INFO

stdout

INFO truncated to log_output_limit (set as hook/operator/sensor kwarg, defaults to 2000). Full text is available at DEBUG level

stderr on success

INFO truncated to log_output_limit (set as hook/operator/sensor kwarg, defaults to 2000). Full text is available at DEBUG level

stderr on failure

ERROR

Lock conflict retry attempt

WARNING

CLI banlist

Most of the native DuckDB flags, such as SQL submission (-c, -f, -s), output formats (-json, -csv), and others are managed by the provider. Specifying these flags in cli_params raises a configuration error.

Hard-banned CLI flags

  • -c

  • -s

  • -f

  • -cmd

  • -init

  • -json

  • -csv

  • -readonly

  • -bail

  • -no-stdin

TIP
To specify an output format, use DuckDbOperator(output_format=…​).

Mask sensitive data

It is recommended to store credentials and sensitive data in the Extra field using special keys masked by Airflow rather than inline in SQL or as operator/hook parameters.

In case of errors, DuckDB typically echoes the SQL statement in stderr, so a secret may appear as plain text in task logs. Values stored in cli_params with keys shown below are masked with *** in logs.

Masked keys

  • password

  • passwd

  • secret

  • token

  • access_key

  • access_token

  • api_key

  • apikey

  • private_key

  • credential

  • credentials

  • secret_key

  • aws_secret_access_key

For example, the token value will be masked in logs:

{
  "cli_params": ["--token", "my-secret-value"]
}

SQL files

SQL files should end with .sql (case-sensitive). Airflow loads SQL scripts from the DAG bundle (dag.folder and/or template_searchpath), so both the DAG processor and the worker can read them. Use the lowercase .sql extension and keep script files inside the bundle.

If the specified SQL file is missing, the following errors are raised:

Failed to resolve template field 'sql'
jinja2.TemplateNotFound

Lifecycle notes

DuckDbHook runs the CLI with start_new_session=True, so a process-group kill operation can stop an ADO wrapper and its DuckDB child when a task is killed or times out. If a worker is terminated with SIGKILL (for example, because of an OOM exception or a docker kill command), on_kill() is not called and an orphaned DuckDB process may keep the file lock until reaped.

Examples

The provider includes several example DAGs that demonstrate common DuckDB provider operations. To run the example DAGs, modify your Airflow connection to match values used in a DAG (Connection ID, Host)

Basic operations

The following DAG demonstrates basic DuckDB provider operations. It uses a preconfigured Airflow connection to access the database, creates a new table, inserts test data, and fetches test data.

example_duckdb_basic.py
from __future__ import annotations

from datetime import datetime, timedelta

from airflow.providers.arenadata.duckdb.operators.duckdb import DuckDbOperator
from airflow.providers.arenadata.duckdb.version_compat import DAG

CONN_ID = "duckdb_default" (1)

default_args = {
    "owner": "airflow",
    "retries": 1,
    "retry_delay": timedelta(minutes=1),
}

with DAG(
    dag_id="example_duckdb_basic",
    start_date=datetime(2024, 1, 1),
    default_args=default_args,
    schedule=None,
    catchup=False,
    tags=["example", "duckdb"],
) as dag:
    create_table = DuckDbOperator( (2)
        task_id="create_table",
        sql="CREATE OR REPLACE TABLE demo(id INT, name VARCHAR);",
        duckdb_conn_id=CONN_ID,
    )

    insert_data = DuckDbOperator( (3)
        task_id="insert_data",
        sql="INSERT INTO demo VALUES (1, 'alpha'), (2, 'beta');",
        duckdb_conn_id=CONN_ID,
    )

    select_count = DuckDbOperator( (4)
        task_id="select_count",
        sql="SELECT count(*) AS c FROM demo",
        duckdb_conn_id=CONN_ID,
    )

    create_table >> insert_data >> select_count (5)
1 Airflow connection ID used to access the DuckDB database.
2 Task to create a new table.
3 Task to write data to the table.
4 Task to retrieve the number of rows in the table.
5 Task dependency chain.

Work with DuckDbSqlSensor

example_duckdb_sensors.py
from __future__ import annotations

from datetime import datetime, timedelta

from airflow.providers.arenadata.duckdb.sensors.duckdb import DuckDbSqlSensor
from airflow.providers.arenadata.duckdb.version_compat import DAG, Param

CONN_ID = "duckdb_default" (1)

default_args = {
    "owner": "airflow",
    "retries": 1,
    "retry_delay": timedelta(minutes=1),
}

with DAG(
    dag_id="example_duckdb_sensors1",
    start_date=datetime(2024, 1, 1),
    default_args=default_args,
    schedule=None,
    catchup=False,
    tags=["example", "duckdb", "sensor"],
    params={
        "table": Param("events", type="string"),
        "min_id": Param(1, type="integer"),
    },
) as dag:
    wait_inline = DuckDbSqlSensor( (2)
        task_id="wait_inline",
        sql="SELECT count(*) AS ready FROM {{ params.table }}",
        duckdb_conn_id=CONN_ID,
        mode="reschedule",
        poke_interval=30, (3)
        timeout=300,
    )

    wait_from_sql_file = DuckDbSqlSensor(
        task_id="wait_from_sql_file",
        sql="queries/wait_until_ready.sql", (4)
        duckdb_conn_id=CONN_ID,
        mode="reschedule",
        poke_interval=30,
        timeout=300,
    )

    wait_inline >> wait_from_sql_file
1 Airflow connection ID used to access the DuckDB database.
2 The task that waits until the SQL statement returns a truthy result (not False, 0, null, or empty string).
3 Time interval between the pokes.
4 The task that waits until the SQL in file returns a truthy result.

Override database path

A path to the .duckdb database file is specified during Airflow connection creation. However, you can override the database path when instantiating an operator. For example:

select_count = DuckDbOperator(
    task_id="select_count",
    sql="SELECT count(*) AS c FROM demo",
    database="/tmp/example_duckdb_sql_file.duckdb", (1)
    duckdb_conn_id="duckdb_default",
    output_format="json",
)
1 Overrides the database path specified in Airflow connection.
Found a mistake? Seleсt text and press Ctrl+Enter to report it