> For the complete documentation index, see [llms.txt](https://experience-digital-transform.gitbook.io/experience-digital-transformation/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://experience-digital-transform.gitbook.io/experience-digital-transformation/analytics/reference-implementation/workflow-orchestration-subsystem-apache-airflow.md).

# Workflow orchestration subsystem: Apache Airflow

Workflow orchestration is handled by Apache Airflow running on Server #1, allowing programmatic planning, scheduling, and monitoring of Directed Acyclic Graphs (DAGs).

## Airflow architecture and deployment

This implementation uses the official Docker image for Apache Airflow (version 2.9.0 or higher). The Airflow stack involves several coordinated services:

* **Metadata DB (PostgreSQL):** Stores task states, DAG definitions, and connection credentials.
* **Message Broker (Redis):** Handles Celery task queues.
* **Scheduler:** Parses files in `dags/` and schedules tasks.
* **Webserver:** Provides the UI on port 8080.
* **Workers:** Execute individual task instances.
* **Triggerer:** Runs an asynchronous event loop for deferrable operators.

To prevent container-to-host file permission conflicts, map the local user ID by initializing a `.env` file with `AIRFLOW_UID` before launching Docker Compose:

```bash
mkdir -p ./dags ./logs ./plugins
echo -e "AIRFLOW_UID=$(id -u)" > .env
docker-compose up airflow-init
docker-compose up -d
```

To add custom packages or Airflow providers, specify `build: .` in `docker-compose.yaml` and extend the base image via a custom local Dockerfile.

## Integration patterns: connections and plugins

Connecting Airflow to ClickHouse requires installing database provider packages:

1. **`apache-airflow-providers-clickhouse`:** The official community provider package containing `ClickhouseHook` and `ClickhouseOperator`.
2. **`airflow-clickhouse-plugin`:** A popular community plugin providing additional components like `ClickHouseSensor` and `ClickHouseSQLExecuteQueryOperator`.

Airflow serializes connection parameters into a URI format. The connectors map these fields to `clickhouse_driver.Client.__init__` parameters as follows:

| Airflow Connection Attribute | Client Init Argument | Notes                                                             |
| ---------------------------- | -------------------- | ----------------------------------------------------------------- |
| `host`                       | `host`               | ClickHouse server hostname or IP.                                 |
| `port`                       | `port`               | Connection port (defaults to 9000 for Native TCP).                |
| `schema`                     | `database`           | Default database (can be overridden in individual tasks).         |
| `login`                      | `user`               | Database user.                                                    |
| `password`                   | `password`           | Database password.                                                |
| `extra`                      | `**kwargs`           | Extensible JSON settings (e.g. `{"secure": true}` to enable TLS). |

An example connection URI: `AIRFLOW_CONN_CLICKHOUSE_DEFAULT=clickhouse://login:password@clickhouse-host:9000/default?secure=True`. For HTTP connections, use port 8123. In the Airflow UI, select "SQLite" or another standard SQL type if a dedicated ClickHouse connection type is unavailable.

## Developing fault-tolerant DAGs

A core principle of data pipeline design is **idempotency**: running a DAG multiple times for the same logical execution date (`{{ ds }}`) must not duplicate or corrupt data.

For ETL/ELT workflows, use a combination of `ClickHouseSensor` and `ClickHouseOperator` to build reliable dependency flows:

1. **`ClickHouseSensor`:** Polls the source tables, delaying DAG execution until a specific condition is met (e.g. verifying that the raw record count for the execution date `{{ ds }}` exceeds a threshold).
2. **Branching and idempotency:** A `ClickHouseBranchSQLOperator` (or a custom branching operator for Airflow versions < 2.9.3 due to MRO limitations) checks if the target partition or table exists. If it exists, the pipeline truncates or deletes the specific partition to prevent duplicates before performing inserts.
3. **Surgical aggregations:** Run sequential SQL statements within a single execution block:

```python
ClickHouseOperator(
    task_id='update_income_aggregate',
    database='analytics',
    sql=(
        "INSERT INTO aggregate SELECT eventDt, sum(price * qty) AS income FROM sales WHERE eventDt = '{{ ds }}' GROUP BY eventDt",
        "OPTIMIZE TABLE aggregate ON CLUSTER {{ var.value.cluster_name }} PARTITION toDate('{{ execution_date.format('%Y-%m-%01') }}')",
        "SELECT sum(income) FROM aggregate WHERE eventDt BETWEEN '{{ execution_date.start_of('month').to_date_string() }}' AND '{{ execution_date.end_of('month').to_date_string() }}'"
    ),
    clickhouse_conn_id='clickhouse_test',
)
```

The query result of the last execution statement is automatically pushed to the Airflow XCom system, allowing subsequent tasks (like Slack notifications) to retrieve it via `task_instance.xcom_pull(task_ids='update_income_aggregate')`.

{% hint style="danger" %}
🛑 **Important constraint.** Never pass direct string-concatenated values into insert queries (such as `ch_hook.run("INSERT INTO table VALUES (1)")`). The native ClickHouse protocol strictly requires data rows to be provided via structured execution parameters to maintain query compilation integrity and prevent SQL injections.
{% endhint %}


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation by asking a question.

Perform an HTTP GET request on the following URL with the `ask` and `goal` query parameters:

```
GET https://experience-digital-transform.gitbook.io/experience-digital-transformation/analytics/reference-implementation/workflow-orchestration-subsystem-apache-airflow.md?ask=<question>&goal=<user_goal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is what the user is ultimately trying to achieve, the reason they need the answer. Sharing it helps GitBook give you a better, more relevant answer. A goal is most helpful when it describes the outcome the user wants rather than restating the question. For example, with `ask=how do I create an API token`, a goal like `automate deployments from our CI pipeline` lets GitBook tailor the answer to that use case.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.