These are the main options for a clickhouse integration airflow setup:

1. Airflow → ClickHouse® via official ClickHouse® Provider (apache-airflow-providers-common-sql + clickhouse-connect)
2. Airflow → Tinybird via REST API using the HTTP operator
3. Airflow → ClickHouse® via Python operator with clickhouse-driver or clickhouse-connect

Apache Airflow is the most widely adopted workflow orchestration platform for data engineering teams. ClickHouse® is a columnar OLAP database that handles billions of rows in sub-second queries. Connecting the two enables scheduled and event-driven data pipelines that load, transform, and query analytical data without manual intervention.

A clickhouse integration airflow pipeline lets data engineers schedule SQL transformations and data loads against ClickHouse®, monitor pipeline health through Airflow's UI, and coordinate ClickHouse® tasks with upstream ingestion and downstream serving steps in a single DAG.

Before you pick a connector, consider these questions:

• Do you need managed connections stored in Airflow's connection store, or are credentials managed externally (Vault, env vars)?
• Does your ClickHouse® instance require SQL-only orchestration, or do you need Python-level control over batching and error handling?
• Do you also need to expose transformed data as REST APIs for applications beyond Airflow pipelines?

Three ways to implement clickhouse integration airflow

This section covers the three main integration paths, with configuration and DAG code for each.

Option 1: Airflow → ClickHouse® — official ClickHouse® Provider

The most direct approach. The apache-airflow-providers-common-sql package combined with clickhouse-connect gives Airflow a managed ClickHouse® connection, a ClickHouseHook, and SQL operators (SQLExecuteQueryOperator) that run queries against ClickHouse® as Airflow tasks.

How it works: install the provider, configure a ClickHouse® connection in Airflow's Connection UI (or via environment variable), then use SQLExecuteQueryOperator or a custom ClickHouseHook to run DDL and DML inside DAG tasks.

Installation:

pip install apache-airflow-providers-common-sql clickhouse-connect

Airflow connection (environment variable format):

AIRFLOW_CONN_CLICKHOUSE_DEFAULT='clickhouse://default:your_password@your-clickhouse-host:8443/default?secure=True'

DAG example: daily aggregation into a ClickHouse® summary table:

from airflow import DAG
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from datetime import datetime, timedelta

default_args = {"owner": "data-engineering", "retries": 2, "retry_delay": timedelta(minutes=5)}

with DAG(
    dag_id="clickhouse_daily_aggregation",
    default_args=default_args,
    start_date=datetime(2026, 1, 1),
    schedule_interval="0 1 * * *",
    catchup=False,
    tags=["clickhouse", "aggregation"],
) as dag:

    insert_daily_summary = SQLExecuteQueryOperator(
        task_id="insert_daily_summary",
        conn_id="clickhouse_default",
        sql="""
            INSERT INTO events_daily_summary
            SELECT
                toDate(event_time)        AS event_date,
                event_type,
                country,
                count()                   AS total_events,
                uniq(user_id)             AS unique_users,
                sum(revenue)              AS total_revenue
            FROM events
            WHERE toDate(event_time) = today() - INTERVAL 1 DAY
            GROUP BY event_date, event_type, country
        """,
    )

    optimize_summary = SQLExecuteQueryOperator(
        task_id="optimize_summary",
        conn_id="clickhouse_default",
        sql="OPTIMIZE TABLE events_daily_summary FINAL",
    )

    insert_daily_summary >> optimize_summary

When this fits:

• You want managed ClickHouse® connections stored in Airflow's connection store with UI visibility
• Your pipeline is SQL-centric — DDL, INSERT INTO ... SELECT, OPTIMIZE TABLE
• You need Airflow task-level retry, alerting, and SLA monitoring on ClickHouse® operations

Trade-offs: SQLExecuteQueryOperator is synchronous — it holds an Airflow worker thread for the duration of the ClickHouse® query. Long-running transformation queries should be batched or use deferred operators. ClickHouse®'s HTTP interface has a default max_execution_time; configure it to avoid silent truncation.

Prerequisites: Airflow 2.5+, apache-airflow-providers-common-sql, clickhouse-connect, ClickHouse® instance reachable from Airflow workers on port 8123 or 8443.

Option 2: Airflow → Tinybird — REST API via HTTP operator

Tinybird sits between your data and Airflow. You define SQL Pipes in Tinybird that query ClickHouse®-backed data sources, publish them as REST API endpoints, and trigger or poll those endpoints from Airflow using the SimpleHttpOperator or HttpSensor.

How it works: Tinybird Pipes expose query results and data ingestion endpoints over HTTPS. Airflow DAG tasks call Tinybird's Events API to push data, or call Tinybird Pipe endpoints to pull results and pass them downstream in XCom.

Airflow connection for Tinybird:

AIRFLOW_CONN_TINYBIRD_DEFAULT='http://https://api.tinybird.co/?headers={"Authorization":"Bearer YOUR_TINYBIRD_TOKEN"}'

DAG example: push transformed data to Tinybird Events API:

from airflow import DAG
from airflow.providers.http.operators.http import SimpleHttpOperator
from airflow.operators.python import PythonOperator
import json
from datetime import datetime, timedelta

default_args = {"owner": "data-engineering", "retries": 2, "retry_delay": timedelta(minutes=5)}

def build_ndjson_payload(**context):
    rows = [
        {"event_date": "2026-05-01", "event_type": "click", "country": "US", "total_events": 4200, "unique_users": 1800},
        {"event_date": "2026-05-01", "event_type": "view",  "country": "DE", "total_events": 8100, "unique_users": 3100},
    ]
    return "\n".join(json.dumps(r) for r in rows)

with DAG(
    dag_id="tinybird_events_push",
    default_args=default_args,
    start_date=datetime(2026, 1, 1),
    schedule_interval="0 2 * * *",
    catchup=False,
    tags=["tinybird", "ingest"],
) as dag:

    prepare_payload = PythonOperator(
        task_id="prepare_ndjson",
        python_callable=build_ndjson_payload,
    )

    push_to_tinybird = SimpleHttpOperator(
        task_id="push_to_tinybird_events_api",
        http_conn_id="tinybird_default",
        endpoint="/v0/events?name=airflow_events",
        method="POST",
        headers={"Authorization": "Bearer YOUR_TINYBIRD_TOKEN", "Content-Type": "application/x-ndjson"},
        data="{{ task_instance.xcom_pull(task_ids='prepare_ndjson') }}",
        response_check=lambda response: response.json().get("quarantined_rows", 0) == 0,
        log_response=True,
    )

    prepare_payload >> push_to_tinybird

DAG example: poll a Tinybird Pipe endpoint and pass results downstream:

from airflow.providers.http.operators.http import SimpleHttpOperator

fetch_summary = SimpleHttpOperator(
    task_id="fetch_tinybird_daily_summary",
    http_conn_id="tinybird_default",
    endpoint="/v0/pipes/daily_summary.json?date=2026-05-01&limit=1000",
    method="GET",
    headers={"Authorization": "Bearer YOUR_TINYBIRD_TOKEN"},
    response_filter=lambda response: response.json()["data"],
    log_response=True,
)

When this fits:

• You already use Tinybird for analytics and want Airflow to push data into Tinybird on a schedule
• You need the same processed data in Airflow pipelines and in REST APIs for application consumers
• You want decoupled ingestion — Airflow orchestrates the schedule, Tinybird manages the data layer

Trade-offs: adds an HTTP layer and a Tinybird dependency. Best for data flows that originate outside ClickHouse® (external APIs, databases, files) and need to land in Tinybird for downstream serving. Not ideal if raw ClickHouse® DDL orchestration is the primary need.

Prerequisites: Airflow 2.5+, apache-airflow-providers-http, Tinybird account with Events API or published Pipes, Tinybird API token with appropriate scopes.

Option 3: Airflow → ClickHouse® — Python operator with clickhouse-connect

For scenarios requiring Python-level control — dynamic query generation, batch inserts, multi-step error handling, or custom retry logic — use the PythonOperator with clickhouse-connect (or clickhouse-driver for native protocol). This approach gives full programmatic control over the ClickHouse® interaction inside an Airflow task.

How it works: the PythonOperator calls a Python function that instantiates a clickhouse_connect client, runs queries, processes results, and returns data to XCom or writes to downstream systems. Credentials are sourced from Airflow Variables, environment variables, or a secrets backend.

Installation:

pip install clickhouse-connect

DAG example: batch INSERT with Python-level error handling:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
import clickhouse_connect
from datetime import datetime, timedelta

default_args = {"owner": "data-engineering", "retries": 3, "retry_delay": timedelta(minutes=2)}

def load_events_batch(batch_date: str, **context):
    client = clickhouse_connect.get_client(
        host=Variable.get("CLICKHOUSE_HOST"),
        port=8443,
        secure=True,
        username="airflow_writer",
        password=Variable.get("CLICKHOUSE_PASSWORD"),
    )

    rows = [
        [1001, "click", "US", "2026-05-01 10:00:00", 0.0],
        [1002, "purchase", "DE", "2026-05-01 10:01:30", 49.99],
    ]
    columns = ["user_id", "event_type", "country", "event_time", "revenue"]

    client.insert("events", rows, column_names=columns)

    count_result = client.query(
        f"SELECT count() FROM events WHERE toDate(event_time) = '{batch_date}'"
    )
    row_count = count_result.first_row[0]
    context["task_instance"].xcom_push(key="rows_loaded", value=row_count)
    return row_count

def validate_load(**context):
    rows_loaded = context["task_instance"].xcom_pull(
        task_ids="load_events_batch", key="rows_loaded"
    )
    if rows_loaded == 0:
        raise ValueError("No rows loaded — pipeline validation failed")

with DAG(
    dag_id="clickhouse_python_batch_load",
    default_args=default_args,
    start_date=datetime(2026, 1, 1),
    schedule_interval="0 3 * * *",
    catchup=False,
    tags=["clickhouse", "batch", "python"],
) as dag:

    load_task = PythonOperator(
        task_id="load_events_batch",
        python_callable=load_events_batch,
        op_kwargs={"batch_date": "{{ ds }}"},
    )

    validate_task = PythonOperator(
        task_id="validate_load",
        python_callable=validate_load,
    )

    load_task >> validate_task

When this fits:

• You need dynamic query generation — batch sizes, partition ranges, or conditional logic not expressible in static SQL strings
• You require Python-level error handling — catching specific ClickHouse® exceptions, partial failure recovery, or row-level validation
• You're building multi-step pipelines that read from one source, transform in Python, and write to ClickHouse®

Trade-offs: more code to maintain than Option 1. Credentials must be managed explicitly (Airflow Variables, environment, or a secrets backend). The Python operator runs synchronously in the Airflow worker — large inserts should use clickhouse-connect's streaming insert to avoid memory pressure.

Prerequisites: Airflow 2.5+, clickhouse-connect (or clickhouse-driver), ClickHouse® instance reachable from Airflow workers, credentials in Airflow Variables or a secrets backend.

Summary: picking the right option

Decision framework: what to choose for clickhouse integration airflow

Pick based on your pipeline complexity, credential management, and downstream consumers:

• ClickHouse® Provider if your pipeline is SQL-centric and you want managed connections with minimal boilerplate. Best for scheduled INSERT INTO ... SELECT transformations, OPTIMIZE TABLE, and DDL tasks.
• Tinybird REST API if Airflow is pushing data into Tinybird and you also serve that data as user-facing analytics or REST APIs. Decouples orchestration from data serving cleanly.
• Python operator if you need dynamic batching, row-level validation, or multi-step Python logic that SQL operators cannot express. Best for complex ETL stages with conditional branching.

Bottom line: for straightforward SQL orchestration, Option 1 is the fastest path. For pipelines that feed application APIs as well as analytics, Option 2 (Tinybird) eliminates a separate serving layer. For complex data transformation logic, Option 3 gives full Python control.

What does clickhouse integration airflow mean (and when should you care)?

A clickhouse integration airflow setup uses Airflow DAGs to orchestrate tasks that read from, write to, or transform data in ClickHouse®. Airflow manages scheduling, retries, alerting, and dependency resolution; ClickHouse® handles the high-performance analytical storage and query execution.

You should care when your ClickHouse® data pipelines grow beyond simple cron jobs. Airflow adds DAG-level visibility, task dependency management, retry policies, and SLA alerts — making ClickHouse® pipelines reproducible and observable in a way that shell scripts or raw cron cannot provide.

The integration also matters when ClickHouse® is one step in a multi-system pipeline — for example, pulling data from Postgres, transforming it in ClickHouse®, and pushing results to an API or S3. Airflow orchestrates the full sequence with dependency-aware scheduling and real-time data processing coordination.

If your ClickHouse® workload is a single daily INSERT INTO ... SELECT with no dependencies, a cron job may be simpler. Airflow earns its weight when you have multiple dependent tasks, SLA requirements, or cross-system pipelines.

Schema and pipeline design

Practical schema rules for Airflow-orchestrated ClickHouse® pipelines

Airflow tasks operate on partitions of data — typically a date range or batch ID. Designing your ClickHouse® schema to align with Airflow's partition model avoids full-table scans and makes retries safe.

Rule 1: partition tables by the Airflow execution date. PARTITION BY toYYYYMM(event_time) or PARTITION BY toDate(event_time) lets Airflow tasks target the exact partition they own. Combined with ReplacingMergeTree, re-running a task for a date is idempotent.

Rule 2: use ReplacingMergeTree for idempotent loads. Airflow retries mean the same data may be inserted multiple times. ReplacingMergeTree(updated_at) deduplicates on the ORDER BY key — retried tasks converge to the correct state.

Rule 3: avoid ALTER TABLE in Airflow tasks. Schema changes during pipeline runs can block queries and affect other tasks. Run DDL in a separate migration DAG with the depends_on_past=True pattern.

Rule 4: use explicit batch markers. Include a batch_id or load_date column to identify which Airflow run inserted each row. This enables precise audit, backfill, and deletion scopes.

Example: Airflow-friendly ClickHouse® schema

CREATE TABLE events (
    event_id     UInt64,
    user_id      UInt64,
    event_type   LowCardinality(String),
    event_time   DateTime,
    country      LowCardinality(String),
    revenue      Float64,
    load_date    Date,
    updated_at   DateTime
)
ENGINE = ReplacingMergeTree(updated_at)
PARTITION BY toYYYYMM(event_time)
ORDER BY (user_id, event_id);

A summary table populated by the Airflow daily aggregation task:

CREATE TABLE events_daily_summary (
    event_date   Date,
    event_type   LowCardinality(String),
    country      LowCardinality(String),
    total_events UInt64,
    unique_users UInt64,
    total_revenue Float64,
    updated_at   DateTime
)
ENGINE = ReplacingMergeTree(updated_at)
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, event_type, country);

Airflow inserts into events_daily_summary with:

INSERT INTO events_daily_summary
SELECT
    toDate(event_time) AS event_date,
    event_type,
    country,
    count()            AS total_events,
    uniq(user_id)      AS unique_users,
    sum(revenue)       AS total_revenue,
    now()              AS updated_at
FROM events
WHERE toDate(event_time) = '{{ ds }}'
GROUP BY event_date, event_type, country

Failure modes

1. Worker thread exhaustion on long queries. SQLExecuteQueryOperator and PythonOperator hold Airflow worker threads for the query duration. Long-running ClickHouse® aggregations on large partitions can starve the worker pool. Mitigation: use deferred operators (Airflow 2.2+), set max_execution_time in ClickHouse® to enforce query limits, and pre-aggregate with materialized views to shorten task duration.

2. Non-idempotent retries causing data duplication. If the Airflow task inserts data and then fails after the INSERT completes, a retry will insert the same rows again. Mitigation: use ReplacingMergeTree with a stable updated_at, scope INSERTs to WHERE load_date = '{{ ds }}', and run OPTIMIZE TABLE FINAL after the insert to force deduplication.

3. ClickHouse® connection pool exhaustion. Many Airflow tasks running in parallel each open a ClickHouse® connection. ClickHouse®'s default max_connections is generous, but high parallelism can exhaust it. Mitigation: use a single ClickHouseHook per task (not per query), set max_connections on the Airflow ClickHouse® connection, and limit DAG parallelism with max_active_tasks.

4. Schema drift between Airflow task SQL and ClickHouse® DDL. Column renames or type changes in ClickHouse® break DAG tasks that reference those columns. Mitigation: treat DAG SQL as a versioned artifact alongside your ClickHouse® migrations; use depends_on_past=True for migration DAGs and test SQL in staging before deploying to production.

5. Silent query failure with no rows affected. ClickHouse® INSERT INTO ... SELECT with a WHERE clause that matches no rows succeeds silently. Mitigation: add a validation task after each insert that checks count() on the target table for the expected partition and raises an exception if zero rows are found.

Why ClickHouse® for Airflow analytics

ClickHouse® is a columnar OLAP database built for the large-scale analytical transformations Airflow pipelines generate. Vectorized execution and columnar compression deliver fast aggregations on billions of rows — INSERT INTO ... SELECT transformations that would take minutes in Postgres or MySQL complete in seconds in ClickHouse®.

ClickHouse®'s MergeTree engine supports time-partitioned data, idempotent writes via ReplacingMergeTree, and pre-aggregation via materialized views — patterns that align directly with Airflow's partition-oriented execution model. For teams that need both scheduled batch pipelines and interactive query performance from the same storage layer, ClickHouse® is among the fastest database for analytics backends available.

Security and operational monitoring

• Authentication: dedicated Airflow-specific ClickHouse® user with GRANT INSERT, SELECT on target tables only. Never use the default admin user in production DAGs.
• TLS: enforce HTTPS on all connections. Port 8443 for Cloud; https_port for self-managed.
• Credential storage: use Airflow's Secrets Backend (HashiCorp Vault, AWS SSM, GCP Secret Manager) — never store passwords in DAG code or Airflow Variables in plain text.
• Query limits: set max_execution_time, max_memory_usage, and max_concurrent_queries on the Airflow ClickHouse® user profile to prevent runaway transformation tasks from impacting other workloads.
• Token hygiene: rotate Tinybird API tokens used in the HTTP operator on schedule; use scoped tokens with minimum permissions per pipeline.

Latency, caching, and freshness considerations

Batch pipelines via the ClickHouse® Provider or Python operator run on Airflow's schedule — data freshness is the schedule interval (hourly, daily). ClickHouse® executes transformations synchronously within the Airflow task; latency is dominated by query complexity and data volume.

Tinybird Pipe caching is relevant for Option 2 — if Airflow pushes data into Tinybird on a schedule, downstream consumers query cached Pipe responses between Airflow runs. This gives near-real-time freshness to API consumers without Airflow running continuously.

For near-real-time pipelines, combine Airflow's scheduling with ClickHouse®'s streaming data ingestion patterns (Kafka Engine, Tinybird Events API) to handle continuous data arrival, and use Airflow only for the periodic aggregation and summary tasks.

Why Tinybird is a strong fit for clickhouse integration airflow

Most data engineering teams using Airflow with ClickHouse® eventually need the same processed data in multiple places — a product dashboard, an embedded chart, or a customer-facing API. Managing a separate API layer means duplicated SQL logic and more infrastructure to operate.

Tinybird solves this by combining a ClickHouse®-powered analytics platform, SQL-based Pipes, and instant REST API publishing in one managed service. Airflow orchestrates the ingestion schedule, Tinybird manages the transformation and serving layer, and downstream consumers — dashboards, APIs, applications — query the same Pipes that Airflow feeds. This is the real-time data ingestion pattern for teams that want orchestration without building a separate data serving layer.

Next step: identify your most critical Airflow DAG that writes to ClickHouse®, model its output as a Tinybird Pipe, and validate API freshness in staging before switching production consumers.

Frequently Asked Questions (FAQs)

How do I connect Airflow to ClickHouse®?

Install apache-airflow-providers-common-sql and clickhouse-connect, then configure a ClickHouse® connection in Airflow's Connection store with the URI clickhouse://user:password@host:8443/database?secure=True. Use SQLExecuteQueryOperator with conn_id pointing to that connection for SQL tasks, or instantiate a clickhouse_connect client directly in a PythonOperator.

Does clickhouse integration airflow support idempotent retries?

Yes, with the right schema design. Use ReplacingMergeTree(updated_at) and scope INSERT INTO ... SELECT statements to a specific date partition (WHERE toDate(event_time) = '{{ ds }}'). Retried tasks insert the same rows, which ReplacingMergeTree deduplicates on the next merge. Run OPTIMIZE TABLE FINAL in a downstream task to force immediate deduplication.

How do I pass ClickHouse® query results between Airflow tasks?

Use XCom. In a PythonOperator, call context["task_instance"].xcom_push(key="result", value=data) after the ClickHouse® query, then pull with xcom_pull(task_ids="upstream_task", key="result") in a downstream task. For large result sets, write intermediate results to S3 or another store and pass only the path through XCom to avoid bloating Airflow's metadata database.

How do I handle long-running ClickHouse® queries in Airflow tasks?

Set max_execution_time on the ClickHouse® user profile to enforce a hard query time limit. For large transformations, break the work into smaller date-partitioned tasks using Airflow's dynamic task mapping (expand). Use Airflow 2.2+ deferred operators to free worker threads while the ClickHouse® query runs asynchronously.

Can I use Airflow to backfill historical data in ClickHouse®?

Yes. Set catchup=True on the DAG and define a start_date covering the historical range. Airflow will generate one DAG run per schedule interval. Scope INSERT INTO ... SELECT statements to WHERE toDate(event_time) = '{{ ds }}' so each run targets exactly its own date partition without overlapping with other runs.

What are the main limitations of clickhouse integration airflow?

SQLExecuteQueryOperator holds an Airflow worker thread for the query duration — long queries can exhaust the worker pool. ClickHouse® does not support transactions, so partial failures in multi-step pipelines require explicit idempotency design. The Tinybird HTTP option adds network latency and a dependency on an external service. Dynamic task mapping with large fan-outs can generate significant metadata overhead in Airflow's database.