> ## Documentation Index
> Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.
> Orchestrez les requêtes ClickHouse et les chargements de données depuis Apache Airflow à l’aide du provider ClickHouse
# Connecter Apache Airflow à ClickHouse
export const ClickHouseSupportedBadge = () => {
return
Compatible avec ClickHouse
;
};
[Apache Airflow](https://airflow.apache.org/) est une plateforme open-source permettant de définir, d’ordonnancer et de superviser des workflows sous forme de code. Les workflows sont définis comme des graphes orientés acycliques (DAG) de tâches écrites en Python.
Le provider `apache-airflow-providers-clickhousedb` connecte Airflow à ClickHouse, ce qui vous permet d’exécuter des requêtes, de créer des tables et de charger des données dans le cadre d’un DAG. Il se connecte via l’[interface HTTP](/docs/fr/concepts/features/interfaces/http) à l’aide du client [`clickhouse-connect`](/docs/fr/integrations/language-clients/python/index), et expose ClickHouse via le framework SQL commun d’Airflow, de sorte que l’opérateur standard `SQLExecuteQueryOperator` gère les requêtes DDL, DML et analytiques, sans nécessiter d’opérateur spécifique à ClickHouse.
## Installer le provider
Installez le provider dans l’environnement où s’exécutent le scheduler Airflow et les workers :
```bash theme={null}
pip install apache-airflow-providers-clickhousedb
```
Le provider dépend de `apache-airflow-providers-common-sql` et de `clickhouse-connect`, qui sont installés en même temps que lui. Pour transmettre les résultats de la requête à des DataFrames pandas ou polars, installez les extras facultatifs :
```bash theme={null}
pip install 'apache-airflow-providers-common-sql[pandas,polars]'
```
## Créer une connexion à ClickHouse
Le provider enregistre un type de connexion `clickhouse`. Créez une connexion depuis l'UI d'Airflow, dans **Admin > Connections**, ou définissez-en une via la CLI ou une variable d'environnement.
Dans l'UI, sélectionnez **ClickHouse** comme type de connexion et renseignez les champs suivants :
| Champ | Description | Par défaut |
| ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------- |
| **Host** | Nom d'hôte du serveur ClickHouse, par exemple `abc123.clickhouse.cloud` | `localhost` |
| **Port** | Port HTTP(S) | `8123` (en clair), `8443` (TLS) |
| **Login** | Nom d'utilisateur ClickHouse | `default` |
| **Password** | Mot de passe de l'utilisateur ClickHouse | (vide) |
| **Database** | Base de données par défaut de la connexion. Dans l'UI, ce champ s'appelle **Database** ; il correspond au champ `schema` lorsque vous définissez la connexion via un URI ou du JSON. | `default` |
Pour [ClickHouse Cloud](/docs/fr/products/cloud/getting-started/intro) ou tout cluster auto-hébergé avec TLS activé, définissez `secure` sur `true` dans le champ **Extra** et utilisez le port TLS (`8443`).
Le provider expose des options supplémentaires sous forme de champs dédiés dans le formulaire de connexion. Si vous définissez la connexion via un URI, du JSON ou une variable d’environnement, indiquez-les plutôt comme clés dans l’objet JSON `extra`. Elles sont toutes facultatives :
| clé `extra` | Champ de l’UI | Par défaut | Description |
| ---------------------- | ------------------------------------------ | ---------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `secure` | Utiliser TLS (HTTPS) | `false` | Active HTTPS/TLS. |
| `verify` | Vérifier le certificat SSL | `true` | Vérifie le certificat TLS du serveur lorsque `secure` vaut `true`. Définissez cette option sur `false` pour les certificats autosignés. |
| `connect_timeout` | Délai d’expiration de connexion (secondes) | `10` | Délai d’expiration de la connexion HTTP, en secondes. |
| `send_receive_timeout` | Délai d’expiration de requête (secondes) | `300` | Délai d’expiration en lecture/écriture des requêtes, en secondes. Augmentez cette valeur pour les requêtes analytiques de longue durée. |
| `compress` | Activer la compression LZ4 | `true` | Active la compression LZ4 des résultats. |
| `client_name` | Nom du client | (vide) | Libellé ajouté à l’identifiant de version d’Airflow dans le `User-Agent` de ClickHouse et dans la colonne `client_name` de [`system.query_log`](/docs/fr/reference/system-tables/query_log). |
| `session_settings` | Paramètres de session (JSON) | (vide) | [Paramètres de session ClickHouse](/docs/fr/reference/settings/session-settings) appliqués à chaque requête de la connexion, par exemple `{"max_execution_time": 300, "max_threads": 8}`. |
| `client_kwargs` | kwargs du client (JSON) | (vide) | Arguments nommés supplémentaires transmis à `clickhouse_connect.get_client()`, par exemple un `http_proxy`. |
### Définir une connexion sans l’UI
Configurez la connexion à l’aide d’une variable d’environnement. Le format URI inclut l’hôte, les identifiants et la base de données :
```bash theme={null}
export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='clickhouse://default:password@localhost:8123/my_database'
```
Tous les éléments de l’URI doivent être codés en URL. Pour TLS, les délais d’expiration ou les paramètres de session, utilisez le format JSON, qui expose les champs **Extra** :
```bash theme={null}
export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='{
"conn_type": "clickhouse",
"host": "abc123.clickhouse.cloud",
"port": 8443,
"login": "default",
"password": "secret",
"schema": "my_database",
"extra": {
"secure": true,
"session_settings": {
"max_execution_time": 300,
"max_memory_usage": 10000000000
}
}
}'
```
Tous les hooks et les opérateurs utilisent l’identifiant de connexion `clickhouse_default`, sauf si vous en indiquez un autre.
## Exécuter des requêtes avec SQLExecuteQueryOperator
Définissez le `conn_id` de l’opérateur sur votre connexion à ClickHouse. Le DAG suivant crée une table, y insère des lignes, les relit, puis supprime la table :
```python theme={null}
from datetime import datetime
from airflow import DAG
from airflow.providers.common.sql.hooks.sql import fetch_all_handler
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
CLICKHOUSE_CONN_ID = "clickhouse_default"
CLICKHOUSE_TABLE = "airflow_example"
with DAG(
dag_id="example_clickhouse",
start_date=datetime(2021, 1, 1),
default_args={"conn_id": CLICKHOUSE_CONN_ID},
schedule="@once",
catchup=False,
) as dag:
create_table = SQLExecuteQueryOperator(
task_id="create_table",
sql=f"""
CREATE TABLE IF NOT EXISTS {CLICKHOUSE_TABLE} (
id UInt32,
name String,
ts DateTime DEFAULT now()
) ENGINE = MergeTree()
ORDER BY id
""",
)
insert_rows = SQLExecuteQueryOperator(
task_id="insert_rows",
sql=f"""
INSERT INTO {CLICKHOUSE_TABLE} (id, name) VALUES
(1, 'Alice'),
(2, 'Bob'),
(3, 'Charlie')
""",
)
read_rows = SQLExecuteQueryOperator(
task_id="read_rows",
sql=f"SELECT id, name FROM {CLICKHOUSE_TABLE} ORDER BY id",
handler=fetch_all_handler,
)
drop_table = SQLExecuteQueryOperator(
task_id="drop_table",
sql=f"DROP TABLE IF EXISTS {CLICKHOUSE_TABLE}",
)
create_table >> insert_rows >> read_rows >> drop_table
```
Le résultat de la requête est récupéré avec le `handler` par défaut (`fetch_all_handler`). Pour renvoyer autre chose que l’ensemble complet des résultats, transmettez un autre `handler`, par exemple `fetch_one_handler` pour récupérer uniquement la première ligne.
### Utiliser une base de données différente pour chaque tâche
Lorsqu'une connexion pointe vers un cluster et que certaines tâches interrogent différentes bases de données, remplacez la base de données via `hook_params` au lieu de créer une connexion distincte :
```python theme={null}
read_rows = SQLExecuteQueryOperator(
task_id="read_rows",
conn_id=CLICKHOUSE_CONN_ID,
sql="SELECT count() FROM events",
hook_params={"database": "analytics"},
)
```
## Utiliser directement le hook
Pour les opérations qui ne se prêtent pas à un opérateur SQL — insertions en masse, streaming ou appels client propres à ClickHouse — utilisez `ClickHouseHook` dans une tâche Python.
La méthode `bulk_insert_rows` du hook utilise le mécanisme d’insertion colonnaire natif de `clickhouse-connect`, bien plus rapide que les insertions ligne par ligne pour de grands jeux de données. Définissez `batch_size` afin de limiter le pic de mémoire sur des entrées très volumineuses :
```python theme={null}
from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook
hook = ClickHouseHook(clickhouse_conn_id="clickhouse_default")
hook.bulk_insert_rows(
table="events",
rows=[("user1", "click"), ("user2", "view")],
column_names=["user_id", "action"],
batch_size=1000,
)
```
Appelez `get_client()` pour accéder au client `clickhouse-connect` sous-jacent afin de gérer tout ce que le hook n’expose pas directement :
```python theme={null}
client = hook.get_client()
total = client.query("SELECT count() FROM events").result_rows[0][0]
```
### Appliquer les paramètres de session
Transmettez des [paramètres de session](/docs/fr/reference/settings/session-settings) lors de l’initialisation du hook, soit directement, soit via le `hook_params` d’un opérateur. Les paramètres transmis au constructeur sont fusionnés avec les `session_settings` définis dans le champ **Extra** de la connexion, et les valeurs du constructeur prévalent en cas de conflit de clés :
```python theme={null}
hook = ClickHouseHook(
clickhouse_conn_id="clickhouse_default",
session_settings={"max_execution_time": 60, "max_threads": 4},
)
```
## Contenu associé
* [Client Python `clickhouse-connect`](/docs/fr/integrations/language-clients/python/index)
* [Interface HTTP de ClickHouse](/docs/fr/concepts/features/interfaces/http)
* [Référence des paramètres de session de ClickHouse](/docs/fr/reference/settings/session-settings)
* [Documentation de référence pour `apache-airflow-providers-clickhousedb`](https://airflow.apache.org/docs/apache-airflow-providers-clickhousedb/)
* [Paquet du provider sur PyPI](https://pypi.org/project/apache-airflow-providers-clickhousedb/)