Connecter Apache Airflow à ClickHouse - ClickHouse Documentation

Apache Airflow 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 à l’aide du client clickhouse-connect, 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 :

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 :

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 ou tout cluster auto-hébergé avec TLS activé, définissez secure sur true dans le champ Extra et utilisez le port TLS (8443).

Options de connexion supplémentaires

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`.
`session_settings`	Paramètres de session (JSON)	(vide)	Paramètres de session ClickHouse 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 :

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 :

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 :

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 :

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 :

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 :

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

hook = ClickHouseHook(
    clickhouse_conn_id="clickhouse_default",
    session_settings={"max_execution_time": 60, "max_threads": 4},
)

​Installer le provider

​Créer une connexion à ClickHouse

​Options de connexion supplémentaires

​Définir une connexion sans l’UI

​Exécuter des requêtes avec SQLExecuteQueryOperator

​Utiliser une base de données différente pour chaque tâche

​Utiliser directement le hook

​Appliquer les paramètres de session

​Contenu associé