---
title: "clickhouse integration airflow — 3 Ways to Connect in 2026"
excerpt: "Orchestrate ClickHouse® pipelines from Apache Airflow via the official ClickHouse® Provider, Tinybird REST API, or Python operators. Pick the right clickhouse integration airflow path."
authors: "Tinybird"
categories: "AI Resources"
createdOn: "2026-05-02 00:00:00"
publishedOn: "2026-05-02 00:00:00"
updatedOn: "2026-05-02 00:00:00"
status: "published"
---

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](https://www.oracle.com/database/what-is-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:**

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

**Airflow connection (environment variable format):**

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

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

```python
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:**

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

**DAG example: push transformed data to Tinybird Events API:**

```python
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:**

```python
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:**

```bash
pip install clickhouse-connect
```

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

```python
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**

| Criterion | ClickHouse® Provider | Tinybird REST API | Python operator |
| --- | --- | --- | --- |
| **Setup complexity** | Low (provider install) | Medium (HTTP conn + Tinybird) | Medium (Python code) |
| **Managed connections** | Yes (Airflow conn store) | Yes (HTTP conn) | Manual (Variables/env) |
| **SQL-only pipelines** | Yes | No | No |
| **Python control** | Limited | Limited | Full |
| **API reuse** | No | Yes (same Pipe serves API + Airflow) | No |
| **Airflow retry/SLA** | Yes | Yes | Yes |
| **Ops burden** | Low | Low | Medium |

## **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](https://www.tinybird.co/blog/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](https://www.tinybird.co/blog/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**

```sql
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:

```sql
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:

```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,
    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](https://www.tinybird.co/blog/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](https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/secrets-backend/index.html) (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](https://www.ibm.com/think/topics/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](https://www.tinybird.co/blog/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.